Move methods to api/serializer.py

app/api/serializer.py

@@ -0,0 +1,109 @@
+from dataclasses import dataclass
+
+from sqlalchemy import or_
+from sqlalchemy.orm import joinedload
+
+from app.config import PAGE_LIMIT
+from app.extensions import db
+
+from app.models import Alias, Mailbox, Contact, EmailLog
+
+
+@dataclass
+class AliasInfo:
+    alias: Alias
+
+    nb_forward: int
+    nb_blocked: int
+    nb_reply: int
+
+
+def serialize_alias_info(alias_info: AliasInfo) -> dict:
+    return {
+        # Alias field
+        "id": alias_info.alias.id,
+        "email": alias_info.alias.email,
+        "creation_date": alias_info.alias.created_at.format(),
+        "creation_timestamp": alias_info.alias.created_at.timestamp,
+        "enabled": alias_info.alias.enabled,
+        "note": alias_info.alias.note,
+        # activity
+        "nb_forward": alias_info.nb_forward,
+        "nb_block": alias_info.nb_blocked,
+        "nb_reply": alias_info.nb_reply,
+    }
+
+
+def serialize_contact(contact: Contact) -> dict:
+    res = {
+        "id": contact.id,
+        "creation_date": contact.created_at.format(),
+        "creation_timestamp": contact.created_at.timestamp,
+        "last_email_sent_date": None,
+        "last_email_sent_timestamp": None,
+        "contact": contact.website_email,
+        "reverse_alias": contact.website_send_to(),
+    }
+
+    email_log: EmailLog = contact.last_reply()
+    if email_log:
+        res["last_email_sent_date"] = email_log.created_at.format()
+        res["last_email_sent_timestamp"] = email_log.created_at.timestamp
+
+    return res
+
+
+def get_alias_infos_with_pagination(user, page_id=0, query=None) -> [AliasInfo]:
+    ret = []
+    q = (
+        db.session.query(Alias)
+        .filter(Alias.user_id == user.id)
+        .order_by(Alias.created_at.desc())
+    )
+
+    if query:
+        q = q.filter(
+            or_(Alias.email.ilike(f"%{query}%"), Alias.note.ilike(f"%{query}%"))
+        )
+
+    q = q.limit(PAGE_LIMIT).offset(page_id * PAGE_LIMIT)
+
+    for alias in q:
+        ret.append(get_alias_info(alias))
+
+    return ret
+
+
+def get_alias_info(alias: Alias) -> AliasInfo:
+    q = (
+        db.session.query(Contact, EmailLog)
+        .filter(Contact.alias_id == alias.id)
+        .filter(EmailLog.contact_id == Contact.id)
+    )
+
+    alias_info = AliasInfo(alias=alias, nb_blocked=0, nb_forward=0, nb_reply=0,)
+
+    for _, el in q:
+        if el.is_reply:
+            alias_info.nb_reply += 1
+        elif el.blocked:
+            alias_info.nb_blocked += 1
+        else:
+            alias_info.nb_forward += 1
+
+    return alias_info
+
+
+def get_alias_contacts(alias, page_id: int) -> [dict]:
+    q = (
+        Contact.query.filter_by(alias_id=alias.id)
+        .order_by(Contact.id.desc())
+        .limit(PAGE_LIMIT)
+        .offset(page_id * PAGE_LIMIT)
+    )
+
+    res = []
+    for fe in q.all():
+        res.append(serialize_contact(fe))
+
+    return res

app/api/views/alias.py

@@ -2,21 +2,22 @@ from flask import g
 from flask import jsonify
 from flask import request
 from flask_cors import cross_origin
-from sqlalchemy import or_
-from sqlalchemy.orm import joinedload
 
 from app.api.base import api_bp, verify_api_key
-from app.config import EMAIL_DOMAIN
+from app.api.serializer import (
-from app.config import PAGE_LIMIT
-from app.dashboard.views.alias_log import get_alias_log
-from app.dashboard.views.index import (
     AliasInfo,
+    serialize_alias_info,
+    serialize_contact,
+    get_alias_infos_with_pagination,
+    get_alias_info,
+    get_alias_contacts,
 )
+from app.config import EMAIL_DOMAIN
+from app.dashboard.views.alias_log import get_alias_log
 from app.email_utils import parseaddr_unicode
 from app.extensions import db
 from app.log import LOG
 from app.models import Alias, Contact
-from app.models import EmailLog
 from app.utils import random_string
 
 
@@ -63,20 +64,6 @@ def get_aliases():
     )
 
 
-def serialize_alias_info(alias_info: AliasInfo) -> dict:
-    return {
-        "id": alias_info.id,
-        "email": alias_info.alias.email,
-        "creation_date": alias_info.alias.created_at.format(),
-        "creation_timestamp": alias_info.alias.created_at.timestamp,
-        "nb_forward": alias_info.nb_forward,
-        "nb_block": alias_info.nb_blocked,
-        "nb_reply": alias_info.nb_reply,
-        "enabled": alias_info.alias.enabled,
-        "note": alias_info.note,
-    }
-
-
 @api_bp.route("/aliases/<int:alias_id>", methods=["DELETE"])
 @cross_origin()
 @verify_api_key
@@ -235,41 +222,6 @@ def get_alias(alias_id):
     return jsonify(**serialize_alias_info(get_alias_info(alias))), 200
 
 
-def serialize_contact(contact: Contact) -> dict:
-
-    res = {
-        "id": contact.id,
-        "creation_date": contact.created_at.format(),
-        "creation_timestamp": contact.created_at.timestamp,
-        "last_email_sent_date": None,
-        "last_email_sent_timestamp": None,
-        "contact": contact.website_email,
-        "reverse_alias": contact.website_send_to(),
-    }
-
-    email_log: EmailLog = contact.last_reply()
-    if email_log:
-        res["last_email_sent_date"] = email_log.created_at.format()
-        res["last_email_sent_timestamp"] = email_log.created_at.timestamp
-
-    return res
-
-
-def get_alias_contacts(alias, page_id: int) -> [dict]:
-    q = (
-        Contact.query.filter_by(alias_id=alias.id)
-        .order_by(Contact.id.desc())
-        .limit(PAGE_LIMIT)
-        .offset(page_id * PAGE_LIMIT)
-    )
-
-    res = []
-    for fe in q.all():
-        res.append(serialize_contact(fe))
-
-    return res
-
-
 @api_bp.route("/aliases/<int:alias_id>/contacts")
 @cross_origin()
 @verify_api_key
@@ -382,53 +334,3 @@ def delete_contact(contact_id):
     db.session.commit()
 
     return jsonify(deleted=True), 200
-
-
-def get_alias_infos_with_pagination(user, page_id=0, query=None) -> [AliasInfo]:
-    ret = []
-    q = (
-        db.session.query(Alias)
-        .options(joinedload(Alias.mailbox))
-        .filter(Alias.user_id == user.id)
-        .order_by(Alias.created_at.desc())
-    )
-
-    if query:
-        q = q.filter(
-            or_(Alias.email.ilike(f"%{query}%"), Alias.note.ilike(f"%{query}%"))
-        )
-
-    q = q.limit(PAGE_LIMIT).offset(page_id * PAGE_LIMIT)
-
-    for alias in q:
-        ret.append(get_alias_info(alias))
-
-    return ret
-
-
-def get_alias_info(alias: Alias) -> AliasInfo:
-    q = (
-        db.session.query(Contact, EmailLog)
-        .filter(Contact.alias_id == alias.id)
-        .filter(EmailLog.contact_id == Contact.id)
-    )
-
-    alias_info = AliasInfo(
-        id=alias.id,
-        alias=alias,
-        mailbox=alias.mailbox,
-        note=alias.note,
-        nb_blocked=0,
-        nb_forward=0,
-        nb_reply=0,
-    )
-
-    for _, el in q:
-        if el.is_reply:
-            alias_info.nb_reply += 1
-        elif el.blocked:
-            alias_info.nb_blocked += 1
-        else:
-            alias_info.nb_forward += 1
-
-    return alias_info

app/api/views/new_custom_alias.py

@@ -3,7 +3,7 @@ from flask import jsonify, request
 from flask_cors import cross_origin
 
 from app.api.base import api_bp, verify_api_key
-from app.api.views.alias import serialize_alias_info, get_alias_info
+from app.api.serializer import serialize_alias_info, get_alias_info
 from app.config import MAX_NB_EMAIL_FREE_PLAN, ALIAS_DOMAINS
 from app.dashboard.views.custom_alias import verify_prefix_suffix
 from app.extensions import db

app/api/views/new_random_alias.py

@@ -3,7 +3,7 @@ from flask import jsonify, request
 from flask_cors import cross_origin
 
 from app.api.base import api_bp, verify_api_key
-from app.api.views.alias import serialize_alias_info, get_alias_info
+from app.api.serializer import serialize_alias_info, get_alias_info
 from app.config import MAX_NB_EMAIL_FREE_PLAN
 from app.extensions import db
 from app.log import LOG

app/dashboard/views/alias_log.py

@@ -61,7 +61,7 @@ def alias_log(alias_id, page_id):
     return render_template("dashboard/alias_log.html", **locals())
 
 
-def get_alias_log(alias: Alias, page_id=0):
+def get_alias_log(alias: Alias, page_id=0) -> [AliasLog]:
     logs: [AliasLog] = []
     mailbox = alias.mailbox_email()