Implement personal data archive (Closes #22)

pyproject.toml

@@ -33,6 +33,7 @@ nanoid = "^2.0.0"
 python-slugify = "^4.0.1"
 fastapi_utils = "^0.2.1"
 fastapi-etag = "^0.2.2"
+pyzip = "^0.2.0"
 
 [tool.poetry.dev-dependencies]
 black = "^19.10b0"

schoolsyst_api/accounts/users.py

@@ -210,22 +210,3 @@ async def delete_current_user(
         db.collection(c).delete_match({"owner_key": user.key})
 
     return Response(status_code=status.HTTP_204_NO_CONTENT)
-
-
-@router.get("/personal_data_archive")
-async def get_personal_data_archive(
-    user: User = Depends(get_current_confirmed_user),
-    db: StandardDatabase = Depends(database.get),
-) -> dict:
-    """
-    Get an archive of all of the data linked to the user.
-    """
-    data = {}
-    # The user's data
-    data["user"] = db.collection("users").get(user.key)
-    # the data of which the user is the owner for every collection
-    for c in COLLECTIONS:
-        if c == "users":
-            continue
-        data[c] = [batch for batch in db.collection(c).find({"owner_key": user.key})]
-    return data

schoolsyst_api/main.py

@@ -2,6 +2,7 @@
 
 import schoolsyst_api.grades.routes
 import schoolsyst_api.homework.routes
+import schoolsyst_api.personal_archive.routes
 import schoolsyst_api.schedule.routes
 import schoolsyst_api.settings.routes
 import schoolsyst_api.statistics.routes
@@ -31,6 +32,7 @@
 api.add_middleware(**cors.middleware_params)
 # Include routes
 api.include_router(accounts.router, tags=["Accounts"])
+api.include_router(schoolsyst_api.personal_archive.routes.router, tags=["Accounts"])
 api.include_router(schoolsyst_api.subjects.routes.router, tags=["Subjects"])
 api.include_router(schoolsyst_api.homework.routes.router, tags=["Homework"])
 api.include_router(schoolsyst_api.settings.routes.router, tags=["Settings"])

schoolsyst_api/personal_archive/models.py

@@ -0,0 +1,23 @@
+from typing import List
+
+from schoolsyst_api.accounts.models import User
+from schoolsyst_api.grades.models import Grade
+from schoolsyst_api.homework.models import Homework
+from schoolsyst_api.learn.models import Quiz
+from schoolsyst_api.models import BaseModel
+from schoolsyst_api.notes.models import Note
+from schoolsyst_api.schedule.models import Event, EventMutation
+from schoolsyst_api.settings.models import Settings
+from schoolsyst_api.subjects.models import Subject
+
+
+class PersonalArchive(BaseModel):
+    subjects: List[Subject]
+    users: List[User]
+    settings: List[Settings]
+    quizzes: List[Quiz]
+    notes: List[Note]
+    grades: List[Grade]
+    homework: List[Homework]
+    events: List[Event]
+    event_mutations: List[EventMutation]

schoolsyst_api/personal_archive/routes.py

@@ -0,0 +1,47 @@
+from arango.database import StandardDatabase
+from fastapi import Depends, Response, status
+from fastapi_utils.inferring_router import InferringRouter
+from pyzip import PyZip
+from schoolsyst_api import database
+from schoolsyst_api.accounts.models import User
+from schoolsyst_api.accounts.users import get_current_confirmed_user
+from schoolsyst_api.database import COLLECTIONS
+from schoolsyst_api.personal_archive.models import PersonalArchive
+
+router = InferringRouter()
+
+
+@router.get(
+    "/personal_data_archive",
+    status_code=status.HTTP_201_CREATED,
+    description=f"""\
+Get an archive of all the data owned by the user.
+The response is a zip file containing a JSON response, which is
+an object associating keys {', '.join(COLLECTIONS)}
+to lists of the corresponding objects.""",
+)
+async def get_personal_data_archive(
+    filename: str = "schoolsyst_data_archive.zip",
+    user: User = Depends(get_current_confirmed_user),
+    db: StandardDatabase = Depends(database.get),
+):
+    """
+    Get an archive of all of the data linked to the user.
+    """
+    data = {}
+    # The user's data
+    data["users"] = [db.collection("users").get(user.key)]
+    # the data of which the user is the owner for every collection
+    for c in COLLECTIONS:
+        if c == "users":
+            continue
+        data[c] = [batch for batch in db.collection(c).find({"owner_key": user.key})]
+    # zip the data
+    zip_file = PyZip()
+    zip_file["data.json"] = PersonalArchive(**data).json(by_alias=True).encode("utf-8")
+    return Response(
+        content=zip_file.to_bytes(),
+        status_code=status.HTTP_201_CREATED,
+        headers={"Content-Disposition": f"attachment; filename={filename}"},
+        media_type="application/zip",
+    )