respond to /contents POST instantly and process file fetching in background

Author: Smehnov

storage/db/migrations/versions/1674283621_5314853acecf_add_enum_value_to_content_availability.py

@@ -0,0 +1,38 @@
+"""add enum value to content availability
+
+Revision ID: 5314853acecf
+Revises: 209e2baad6e1
+Create Date: 2023-01-21 09:47:01.173291
+
+"""
+from alembic import op
+
+from storage.db.multitenancy import for_each_tenant_schema
+
+# revision identifiers, used by Alembic.
+revision = "5314853acecf"
+down_revision = "209e2baad6e1"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute("ALTER TYPE contentavailability ADD VALUE 'PENDING'")
+
+
+@for_each_tenant_schema
+def downgrade(schema: str):
+    op.drop_column("contents", "availability", schema=schema)
+    op.execute("ALTER TYPE contentavailability RENAME TO contentavailability_old")
+    op.execute(
+        "CREATE TYPE contentavailability "
+        "AS ENUM('INSTANT', 'ENCRYPTED', 'ARCHIVE', 'ABSENT')"
+    )
+    op.execute(
+        (
+            f"ALTER TABLE {schema}.contents "
+            "ALTER COLUMN availability TYPE contentavailability USING "
+            "contentavailability::text::contentavailability"
+        )
+    )
+    op.execute("DROP TYPE contentavailability_old")

storage/db/models/content.py

@@ -9,6 +9,7 @@
 
 
 class ContentAvailability(str, enum.Enum):
+    PENDING = "pending"
     INSTANT = "instant"
     ENCRYPTED = "encrypted"
     ARCHIVE = "archive"

storage/upload.py

@@ -0,0 +1,28 @@
+import io
+
+import httpx
+
+from storage.config import settings
+from storage.db.models import Content
+from storage.db.models.content import ContentAvailability
+from storage.logging import log
+
+
+async def process_data_from_origin(origin: str, content_id: int, db) -> None:
+    log.debug(f"fetching content {content_id} {origin}")
+    async with httpx.AsyncClient() as client:
+        r = await client.get(origin)
+        content_file = io.BytesIO(r.content)
+    async with httpx.AsyncClient(
+        base_url=settings.IPFS_HTTP_PROVIDER,
+    ) as client:
+        response = await client.post(
+            "/api/v0/add", files={"upload-files": content_file}
+        )
+
+    content: Content = db.query(Content).filter(Content.id == content_id).first()
+    content.ipfs_cid = response.json()["Hash"]
+    content.availability = ContentAvailability.INSTANT
+    db.commit()
+    db.refresh(content)
+    log.debug(f"fetched {content=}")

storage/web/routers/contents.py

@@ -1,7 +1,5 @@
-import io
-
 import httpx
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, BackgroundTasks, Depends, status
 from fastapi.exceptions import HTTPException
 from fastapi.responses import RedirectResponse, StreamingResponse
 
@@ -12,6 +10,7 @@
 from storage.db.session import SessionLocal
 from storage.logging import log
 from storage.schemas import content as schemas
+from storage.upload import process_data_from_origin
 from storage.web import deps
 
 router = APIRouter()
@@ -65,6 +64,7 @@ async def read_contents(
 async def create_content(
     *,
     db: SessionLocal = Depends(deps.get_db),
+    background_tasks: BackgroundTasks,
     current_user: User = Depends(deps.get_current_user),
     content_in: schemas.ContentCreate,
 ):
@@ -86,24 +86,20 @@ async def create_content(
             f"/contents/{content_id}",
             status_code=status.HTTP_303_SEE_OTHER,
         )
-    async with httpx.AsyncClient() as client:
-        r = await client.get(content_in.origin)
-        content_file = io.BytesIO(r.content)
-    async with httpx.AsyncClient(
-        base_url=settings.IPFS_HTTP_PROVIDER,
-    ) as client:
-        response = await client.post(
-            "/api/v0/add", files={"upload-files": content_file}
-        )
+
     content = Content(
         origin=content_in.origin,
-        ipfs_cid=response.json()["Hash"],
-        availability=ContentAvailability.INSTANT,
+        availability=ContentAvailability.PENDING,
         owner_id=current_user.id,
     )
     db.add(content)
     db.commit()
     db.refresh(content)
+
+    background_tasks.add_task(
+        process_data_from_origin, origin=content.origin, content_id=content.id, db=db
+    )
+
     return content