Added upload file funcitonality

pyproject.toml

@@ -20,6 +20,8 @@ python-multipart = "^0.0.6"
 boto3 = "^1.33.12"
 alembic = "^1.12.0"
 async-lru = "^2.0.4"
+fsspec = "^2024.2.0"
+s3fs = "^2024.2.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.3"
@@ -28,6 +30,7 @@ pytest-mock = "^3.12.0"
 pytest-env = "^1.1.3"
 pytest-alembic = "^0.10.7"
 pytest-asyncio = "^0.23.2"
+moto = "^4.2.13"
 aiosqlite = "^0.19.0"
 httpx = "^0.26.0"
 
@@ -68,6 +71,8 @@ env = [
   "TOKEN_URL=http://localhost",
   "CERTS_URL=http://localhost",
   "AUTH_CLIENT=",
+  "UPLOAD_FS_PROTOCOL=file",
+  "UPLOAD_FS_ROOT=../upload",
 ]
 testpaths = ["tests"]
 

src/.env.local

@@ -2,4 +2,19 @@ DB_NAME=filing
 DB_USER=filing_user
 DB_PWD=filing_user
 DB_HOST=localhost:5432
-DB_SCHEMA=filing
+DB_SCHEMA=filing
+KC_URL=http://localhost:8880
+KC_REALM=regtech
+KC_ADMIN_CLIENT_ID=admin-cli
+KC_ADMIN_CLIENT_SECRET=local_test_only
+KC_REALM_URL=${KC_URL}/realms/${KC_REALM}
+KC_REALM_ADMIN_URL=${KC_URL}/admin/realms/${KC_REALM}
+AUTH_URL=${KC_REALM_URL}/protocol/openid-connect/auth
+AUTH_CLIENT=regtech-client
+TOKEN_URL=${KC_REALM_URL}/protocol/openid-connect/token
+CERTS_URL=${KC_REALM_URL}/protocol/openid-connect/certs
+JWT_OPTS_VERIFY_AT_HASH="false"
+JWT_OPTS_VERIFY_AUD="false"
+JWT_OPTS_VERIFY_ISS="false"
+UPLOAD_FS_PROTOCOL=file
+UPLOAD_FS_ROOT=../upload

src/.env.template

@@ -3,4 +3,16 @@ DB_USER=
 DB_PWD=
 DB_HOST=
 DB_SCHEMA=
-# DB_SCHEME= can be used to override postgresql+asyncpg if needed
+# DB_SCHEME= can be used to override postgresql+asyncpg if needed
+KC_URL=
+KC_REALM=
+KC_ADMIN_CLIENT_ID=
+KC_ADMIN_CLIENT_SECRET=
+KC_REALM_URL=${KC_URL}/realms/${KC_REALM}
+KC_REALM_ADMIN_URL=${KC_URL}/admin/realms/${KC_REALM}
+AUTH_URL=${KC_REALM_URL}/protocol/openid-connect/auth
+AUTH_CLIENT=
+TOKEN_URL=${KC_REALM_URL}/protocol/openid-connect/token
+CERTS_URL=${KC_REALM_URL}/protocol/openid-connect/certs
+UPLOAD_FS_PROTOCOL=file
+UPLOAD_FS_ROOT=../upload

src/config.py

@@ -1,3 +1,4 @@
+from enum import StrEnum
 import os
 from urllib import parse
 from typing import Any
@@ -13,6 +14,11 @@
     env_files_to_load.append(".env.local")
 
 
+class FsProtocol(StrEnum):
+    FILE = "file"
+    S3 = "s3"
+
+
 class Settings(BaseSettings):
     db_schema: str = "public"
     db_name: str
@@ -21,6 +27,18 @@ class Settings(BaseSettings):
     db_host: str
     db_scheme: str = "postgresql+asyncpg"
     conn: PostgresDsn | None = None
+    """
+    upload_fs_protocol: to be used with fsspec, and s3fs
+    `file` is for local file system
+    `s3` is for AWS S3
+    """
+    upload_fs_protocol: FsProtocol = FsProtocol.FILE
+    """
+    upload_fs_root: root of the upload folder in file system
+    with `file` protocol, this can be any directory you specific (e.g. `../upload`)
+    if using `s3` for the protocol, this should be the bucket name (e.g. `my-s3-bucket`)
+    """
+    upload_fs_root: str
 
     def __init__(self, **data):
         super().__init__(**data)

src/routers/filing.py

@@ -41,7 +41,7 @@ async def upload_file(
     request: Request, lei: str, submission_id: str, file: UploadFile, background_tasks: BackgroundTasks
 ):
     content = await file.read()
-    await submission_processor.upload_to_storage(lei, submission_id, content)
+    await submission_processor.upload_to_storage(lei, submission_id, content, file.filename.split(".")[-1])
     background_tasks.add_task(submission_processor.validate_submission, lei, submission_id, content)
 
 

src/services/submission_processor.py

@@ -1,6 +1,21 @@
-async def upload_to_storage(lei: str, submission_id: str, content: bytes):
-    # implement uploading process here
-    pass
+from http import HTTPStatus
+from fastapi import HTTPException
+import logging
+from fsspec import AbstractFileSystem, filesystem
+from config import settings
+
+log = logging.getLogger(__name__)
+
+
+async def upload_to_storage(lei: str, submission_id: str, content: bytes, extension: str = "csv"):
+    try:
+        fs: AbstractFileSystem = filesystem(settings.upload_fs_protocol.value)
+        fs.mkdirs(f"{settings.upload_fs_root}/{lei}", exist_ok=True)
+        with fs.open(f"{settings.upload_fs_root}/{lei}/{submission_id}.{extension}", "wb") as f:
+            f.write(content)
+    except Exception as e:
+        log.error("Failed to upload file", e, exc_info=True, stack_info=True)
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail="Failed to upload file")
 
 
 async def validate_submission(lei: str, submission_id: str, content: bytes):

tests/services/test_submission_processor.py

@@ -0,0 +1,37 @@
+from fastapi import HTTPException
+import pytest
+from unittest.mock import Mock, ANY
+from pytest_mock import MockerFixture
+from services.submission_processor import upload_to_storage
+
+
+@pytest.fixture
+def mock_fs(mocker: MockerFixture) -> Mock:
+    fs_mock_patch = mocker.patch("services.submission_processor.AbstractFileSystem")
+    return fs_mock_patch.return_value
+
+
+@pytest.fixture
+def mock_fs_func(mocker: MockerFixture, mock_fs: Mock) -> Mock:
+    fs_func_mock = mocker.patch("services.submission_processor.filesystem")
+    fs_func_mock.return_value = mock_fs
+    return fs_func_mock
+
+
+async def test_upload(mocker: MockerFixture, mock_fs_func: Mock, mock_fs: Mock):
+    with mocker.mock_open(mock_fs.open):
+        await upload_to_storage("test", "test", b"test content")
+    mock_fs_func.assert_called()
+    mock_fs.mkdirs.assert_called()
+    mock_fs.open.assert_called_with(ANY, "wb")
+    file_handle = mock_fs.open()
+    file_handle.write.assert_called_with(b"test content")
+
+
+async def test_upload_failure(mocker: MockerFixture, mock_fs_func: Mock, mock_fs: Mock):
+    log_mock = mocker.patch("services.submission_processor.log")
+    mock_fs.mkdirs.side_effect = IOError("test")
+    with pytest.raises(Exception) as e:
+        await upload_to_storage("test", "test", b"test content")
+    log_mock.error.assert_called_with("Failed to upload file", ANY, exc_info=True, stack_info=True)
+    assert isinstance(e.value, HTTPException)