Merge pull request #70 from simple-login/api-alias

Api alias

README.md

@@ -737,6 +737,43 @@ Output:
 The `api_key` is used in all subsequent requests. It's empty if MFA is enabled.
 If user hasn't enabled MFA, `mfa_key` is empty.
 
+#### GET /api/aliases
+
+Get user aliases.
+
+Input:
+- `Authentication` header that contains the api key
+- `page_id` used for the pagination. The endpoint returns maximum 20 aliases for each page. `page_id` starts at 0.
+
+Output:
+If success, 200 with the list of aliases, for example:
+
+```json
+{
+    "aliases": [
+        {
+            "creation_date": "2020-02-04 16:23:02+00:00",
+            "creation_timestamp": 1580833382,
+            "email": "e3@.alo@sl.local",
+            "id": 4,
+            "nb_block": 0,
+            "nb_forward": 0,
+            "nb_reply": 0
+        },
+        {
+            "creation_date": "2020-02-04 16:23:02+00:00",
+            "creation_timestamp": 1580833382,
+            "email": "e2@.meo@sl.local",
+            "id": 3,
+            "nb_block": 0,
+            "nb_forward": 0,
+            "nb_reply": 0
+        }
+    ]
+}
+```
+
+
 ### Database migration
 
 The database migration is handled by `alembic`

app/api/__init__.py

@@ -5,4 +5,5 @@ from .views import (
     user_info,
     auth_login,
     auth_mfa,
+    alias,
 )

app/api/views/alias.py

@@ -0,0 +1,58 @@
+from flask import g
+from flask import jsonify, request
+from flask_cors import cross_origin
+
+from app.api.base import api_bp, verify_api_key
+from app.config import MAX_NB_EMAIL_FREE_PLAN
+from app.dashboard.views.custom_alias import verify_prefix_suffix
+from app.dashboard.views.index import get_alias_info, AliasInfo
+from app.extensions import db
+from app.log import LOG
+from app.models import GenEmail, AliasUsedOn
+from app.utils import convert_to_id
+
+
+@api_bp.route("/aliases")
+@cross_origin()
+@verify_api_key
+def get_aliases():
+    """
+    Get aliases
+    Input:
+        page_id: in query
+    Output:
+        - aliases: list of alias:
+            - id
+            - email
+            - creation_date
+            - creation_timestamp
+            - nb_forward
+            - nb_block
+            - nb_reply
+
+    """
+    user = g.user
+    try:
+        page_id = int(request.args.get("page_id"))
+    except (ValueError, TypeError):
+        return jsonify(error="page_id must be provided in request query"), 400
+
+    aliases: [AliasInfo] = get_alias_info(user.id, page_id=page_id)
+
+    return (
+        jsonify(
+            aliases=[
+                {
+                    "id": alias.id,
+                    "email": alias.gen_email.email,
+                    "creation_date": alias.gen_email.created_at.format(),
+                    "creation_timestamp": alias.gen_email.created_at.timestamp,
+                    "nb_forward": alias.nb_forward,
+                    "nb_block": alias.nb_blocked,
+                    "nb_reply": alias.nb_reply,
+                }
+                for alias in aliases
+            ]
+        ),
+        200,
+    )

app/api/views/alias_options.py

@@ -26,6 +26,7 @@ def options():
         existing: array of existing aliases
 
     """
+    LOG.error("/v2/alias/options should be used instead")
     user = g.user
     hostname = request.args.get("hostname")
 

app/config.py

@@ -170,3 +170,6 @@ FLASK_PROFILER_PASSWORD = os.environ.get("FLASK_PROFILER_PASSWORD")
 
 # Job names
 JOB_ONBOARDING_1 = "onboarding-1"
+
+# for pagination
+PAGE_LIMIT = 20

app/dashboard/views/alias_log.py

@@ -2,12 +2,11 @@ import arrow
 from flask import render_template, flash, redirect, url_for
 from flask_login import login_required, current_user
 
+from app.config import PAGE_LIMIT
 from app.dashboard.base import dashboard_bp
 from app.extensions import db
 from app.models import GenEmail, ForwardEmailLog, ForwardEmail
 
-_LIMIT = 15
-
 
 class AliasLog:
     website_email: str
@@ -54,7 +53,7 @@ def alias_log(alias_id, page_id):
     email_replied = base.filter(ForwardEmailLog.is_reply == True).count()
     email_blocked = base.filter(ForwardEmailLog.blocked == True).count()
     last_page = (
-        len(logs) < _LIMIT
+        len(logs) < PAGE_LIMIT
     )  # lightweight pagination without counting all objects
 
     return render_template("dashboard/alias_log.html", **locals())
@@ -68,8 +67,8 @@ def get_alias_log(gen_email: GenEmail, page_id=0):
         .filter(ForwardEmail.id == ForwardEmailLog.forward_id)
         .filter(ForwardEmail.gen_email_id == gen_email.id)
         .order_by(ForwardEmailLog.id.desc())
-        .limit(_LIMIT)
-        .offset(page_id * _LIMIT)
+        .limit(PAGE_LIMIT)
+        .offset(page_id * PAGE_LIMIT)
     )
 
     for fe, fel in q:

app/dashboard/views/index.py

@@ -4,6 +4,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import joinedload
 
 from app import email_utils
+from app.config import PAGE_LIMIT
 from app.dashboard.base import dashboard_bp
 from app.extensions import db
 from app.log import LOG
@@ -18,6 +19,7 @@ from app.models import (
 
 
 class AliasInfo:
+    id: int
     gen_email: GenEmail
     nb_forward: int
     nb_blocked: int
@@ -143,27 +145,35 @@ def index():
     )
 
 
-def get_alias_info(user_id, query=None, highlight_gen_email_id=None) -> [AliasInfo]:
+def get_alias_info(
+    user_id, query=None, highlight_gen_email_id=None, page_id=None
+) -> [AliasInfo]:
     if query:
         query = query.strip().lower()
 
     aliases = {}  # dict of alias and AliasInfo
+
     q = (
         db.session.query(GenEmail, ForwardEmail, ForwardEmailLog)
-        .filter(
-            GenEmail.user_id == user_id,
-            GenEmail.id == ForwardEmail.gen_email_id,
-            ForwardEmail.id == ForwardEmailLog.forward_id,
+        .join(ForwardEmail, GenEmail.id == ForwardEmail.gen_email_id, isouter=True)
+        .join(
+            ForwardEmailLog, ForwardEmail.id == ForwardEmailLog.forward_id, isouter=True
         )
+        .filter(GenEmail.user_id == user_id)
         .order_by(GenEmail.created_at.desc())
     )
 
     if query:
         q = q.filter(GenEmail.email.contains(query))
 
+    # pagination activated
+    if page_id is not None:
+        q = q.limit(PAGE_LIMIT).offset(page_id * PAGE_LIMIT)
+
     for ge, fe, fel in q:
         if ge.email not in aliases:
             aliases[ge.email] = AliasInfo(
+                id=ge.id,
                 gen_email=ge,
                 nb_blocked=0,
                 nb_forward=0,
@@ -172,6 +182,9 @@ def get_alias_info(user_id, query=None, highlight_gen_email_id=None) -> [AliasIn
             )
 
         alias_info = aliases[ge.email]
+        if not fel:
+            continue
+
         if fel.is_reply:
             alias_info.nb_reply += 1
         elif fel.blocked:
@@ -179,25 +192,6 @@ def get_alias_info(user_id, query=None, highlight_gen_email_id=None) -> [AliasIn
         else:
             alias_info.nb_forward += 1
 
-    # also add alias that has no forward email or log
-    q = (
-        db.session.query(GenEmail)
-        .filter(GenEmail.email.notin_(aliases.keys()))
-        .filter(GenEmail.user_id == user_id)
-    ).order_by(GenEmail.created_at.desc())
-
-    if query:
-        q = q.filter(GenEmail.email.contains(query))
-
-    for ge in q:
-        aliases[ge.email] = AliasInfo(
-            gen_email=ge,
-            nb_blocked=0,
-            nb_forward=0,
-            nb_reply=0,
-            highlight=ge.id == highlight_gen_email_id,
-        )
-
     ret = list(aliases.values())
 
     # make sure the highlighted alias is the first element

tests/api/test_alias.py

@@ -0,0 +1,56 @@
+from flask import url_for
+
+from app.config import EMAIL_DOMAIN, MAX_NB_EMAIL_FREE_PLAN, PAGE_LIMIT
+from app.extensions import db
+from app.models import User, ApiKey, GenEmail
+from app.utils import random_word
+
+
+def test_error_without_pagination(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    # create api_key
+    api_key = ApiKey.create(user.id, "for test")
+    db.session.commit()
+
+    r = flask_client.get(
+        url_for("api.get_aliases"), headers={"Authentication": api_key.code},
+    )
+
+    assert r.status_code == 400
+    assert r.json["error"]
+
+
+def test_success_with_pagination(flask_client):
+    user = User.create(
+        email="a@b.c", password="password", name="Test User", activated=True
+    )
+    db.session.commit()
+
+    # create api_key
+    api_key = ApiKey.create(user.id, "for test")
+    db.session.commit()
+
+    # create more aliases than PAGE_LIMIT
+    for _ in range(PAGE_LIMIT + 1):
+        GenEmail.create_new_random(user.id)
+    db.session.commit()
+
+    # get aliases on the 1st page, should return PAGE_LIMIT aliases
+    r = flask_client.get(
+        url_for("api.get_aliases", page_id=0), headers={"Authentication": api_key.code},
+    )
+    assert r.status_code == 200
+    assert len(r.json["aliases"]) == PAGE_LIMIT
+
+    # get aliases on the 2nd page, should return 2 aliases
+    # as the total number of aliases is PAGE_LIMIT +2
+    # 1 alias is created when user is created
+    r = flask_client.get(
+        url_for("api.get_aliases", page_id=1), headers={"Authentication": api_key.code},
+    )
+    assert r.status_code == 200
+    assert len(r.json["aliases"]) == 2