file upload service

Author: Jens Kürten

csfunctions/handler.py

@@ -96,9 +96,7 @@ def execute(function_name: str, request_body: str, function_dir: str = "src") ->
         link_objects(request.event)
 
         function_callback = get_function_callable(function_name, function_dir)
-        service = Service(
-            str(request.metadata.service_url) if request.metadata.service_url else None, request.metadata.service_token
-        )
+        service = Service(metadata=request.metadata)
 
         response = function_callback(request.metadata, request.event, service)
 

csfunctions/service/__init__.py

@@ -1,3 +1,5 @@
+from csfunctions.metadata import MetaData
+from csfunctions.service.file_upload import FileUploadService
 from csfunctions.service.numgen import NumberGeneratorService
 
 
@@ -6,5 +8,6 @@ class Service:
     Provides access to services on the elements instance, e.g. generating numbers.
     """
 
-    def __init__(self, service_url: str | None, service_token: str | None):
-        self.generator = NumberGeneratorService(service_url, service_token)
+    def __init__(self, metadata: MetaData):
+        self.generator = NumberGeneratorService(metadata=metadata)
+        self.file_upload = FileUploadService(metadata=metadata)

csfunctions/service/base.py

@@ -2,36 +2,58 @@
 
 import requests
 
+from csfunctions.metadata import MetaData
+
 
 class Unauthorized(Exception):
     pass
 
 
+class Conflict(Exception):
+    pass
+
+
+class NotFound(Exception):
+    pass
+
+
+class UnprocessableEntity(Exception):
+    pass
+
+
 class BaseService:
     """
     Base class for services.
     """
 
-    def __init__(self, service_url: str | None, service_token: str | None):
-        self.service_url = service_url
-        self.service_token = service_token
+    def __init__(self, metadata: MetaData):
+        # Store full metadata for services that need additional fields (e.g. app_user)
+        self.metadata = metadata
 
-    def request(self, endpoint: str, method: str = "GET", params: Optional[dict] = None) -> dict | list:
+    def request(
+        self, endpoint: str, method: str = "GET", params: Optional[dict] = None, json: Optional[dict] = None
+    ) -> dict | list:
         """
         Make a request to the access service.
         """
-        if self.service_url is None:
+        if self.metadata.service_url is None:
             raise ValueError("No service url given.")
-        if self.service_token is None:
+        if self.metadata.service_token is None:
             raise ValueError("No service token given.")
 
-        headers = {"Authorization": f"Bearer {self.service_token}"}
+        headers = {"Authorization": f"Bearer {self.metadata.service_token}"}
         params = params or {}
-        url = self.service_url.rstrip("/") + "/" + endpoint.lstrip("/")
-        response = requests.request(method, url=url, params=params, headers=headers, timeout=10)
+        url = str(self.metadata.service_url).rstrip("/") + "/" + endpoint.lstrip("/")
+        response = requests.request(method, url=url, params=params, headers=headers, timeout=10, json=json)
 
         if response.status_code == 401:
             raise Unauthorized
+        elif response.status_code == 409:
+            raise Conflict
+        elif response.status_code == 404:
+            raise NotFound
+        elif response.status_code == 422:
+            raise UnprocessableEntity(response.text)
         if response.status_code == 200:
             return response.json()
         else:

csfunctions/service/file_upload.py

@@ -0,0 +1,161 @@
+import hashlib
+from copy import deepcopy
+from random import choice
+from string import ascii_letters
+from typing import BinaryIO
+
+import requests
+
+from csfunctions.service.base import BaseService
+from csfunctions.service.file_upload_schemas import (
+    CompleteFileUploadRequest,
+    CreateNewFileRequest,
+    CreateNewFileResponse,
+    GeneratePresignedUrlRequest,
+    PresignedWriteUrls,
+)
+
+
+def _generate_lock_id():
+    return "".join(choice(ascii_letters) for i in range(12))  # nosec
+
+
+class FileUploadService(BaseService):
+    def _create_new_file(self, filename: str, parent_object_id: str, persno: str, check_access: bool = True) -> str:
+        """Creates a new (empty) file attached to the parent object and returns the cdb_object_id."""
+        response_json = self.request(
+            endpoint="/file_upload/create",
+            method="POST",
+            json=CreateNewFileRequest(
+                parent_object_id=parent_object_id, filename=filename, persno=persno, check_access=check_access
+            ).model_dump(),
+        )
+        data = CreateNewFileResponse.model_validate(response_json)
+        return data.file_object_id
+
+    def _get_presigned_write_urls(
+        self, file_object_id: str, filesize: int, lock_id: str, persno: str, check_access: bool = True
+    ) -> PresignedWriteUrls:
+        response_json = self.request(
+            endpoint=f"/file_upload/{file_object_id}/generate_presigned_url",
+            method="POST",
+            json=GeneratePresignedUrlRequest(
+                check_access=check_access, persno=persno, filesize=filesize, lock_id=lock_id
+            ).model_dump(),
+        )
+
+        return PresignedWriteUrls.model_validate(response_json)
+
+    def _upload_file_content(
+        self, presigned_urls: PresignedWriteUrls, stream: BinaryIO
+    ) -> tuple[PresignedWriteUrls, str]:
+        """Upload file stream in chunks using presigned URLs and return updated context + sha256 hash."""
+        etags: list[str] = []
+        sha256 = hashlib.sha256()
+        for url in presigned_urls.urls:
+            data: bytes = stream.read(presigned_urls.chunksize)
+            sha256.update(data)
+            resp = requests.put(url, data=data, headers=presigned_urls.headers, timeout=20)
+            # 20 second timeout to stay below 30s max execution time of the Function
+            # otherwise we won't get a proper error message in the logs
+            resp.raise_for_status()
+            etag = resp.headers.get("ETag")
+            if etag:
+                etags.append(etag)
+        updated = deepcopy(presigned_urls)
+        if etags:
+            updated.etags = etags
+        return updated, sha256.hexdigest()
+
+    @staticmethod
+    def _get_stream_size(stream: BinaryIO) -> int:
+        if not stream.seekable():
+            raise ValueError("Stream is not seekable; size cannot be determined.")
+        current_pos = stream.tell()
+        stream.seek(0, 2)
+        size = stream.tell()
+        stream.seek(current_pos)
+        return size
+
+    def _complete_upload(
+        self,
+        file_object_id: str,
+        filesize: int,
+        lock_id: str,
+        presigned_urls: PresignedWriteUrls,
+        persno: str,
+        check_access: bool = True,
+        sha256: str | None = None,
+        delete_derived_files: bool = True,
+    ) -> None:
+        self.request(
+            endpoint=f"/file_upload/{file_object_id}/complete",
+            method="POST",
+            json=CompleteFileUploadRequest(
+                filesize=filesize,
+                check_access=check_access,
+                persno=persno,
+                presigned_write_urls=presigned_urls,
+                lock_id=lock_id,
+                sha256=sha256,
+                delete_derived_files=delete_derived_files,
+            ).model_dump(),
+        )
+
+    def upload_file_content(
+        self,
+        file_object_id: str,
+        stream: BinaryIO,
+        persno: str | None = None,
+        check_access: bool = True,
+        filesize: int | None = None,
+        delete_derived_files: bool = True,
+    ) -> None:
+        persno = persno or self.metadata.app_user
+        if filesize is None:
+            filesize = self._get_stream_size(stream)
+        lock_id = _generate_lock_id()
+        presigned = self._get_presigned_write_urls(
+            file_object_id=file_object_id,
+            filesize=filesize,
+            lock_id=lock_id,
+            persno=persno,
+            check_access=check_access,
+        )
+        presigned_with_etags, sha256 = self._upload_file_content(presigned_urls=presigned, stream=stream)
+        self._complete_upload(
+            file_object_id=file_object_id,
+            filesize=filesize,
+            lock_id=lock_id,
+            presigned_urls=presigned_with_etags,
+            persno=persno,
+            check_access=check_access,
+            sha256=sha256,
+            delete_derived_files=delete_derived_files,
+        )
+
+    def upload_new_file(
+        self,
+        parent_object_id: str,
+        filename: str,
+        stream: BinaryIO,
+        persno: str | None = None,
+        check_access: bool = True,
+        filesize: int | None = None,
+    ) -> str:
+        persno = persno or self.metadata.app_user
+        file_object_id = self._create_new_file(
+            filename=filename,
+            parent_object_id=parent_object_id,
+            persno=persno,
+            check_access=check_access,
+        )
+        self.upload_file_content(
+            file_object_id=file_object_id,
+            stream=stream,
+            persno=persno,
+            check_access=check_access,
+            filesize=filesize,
+            delete_derived_files=False,
+        )
+        return file_object_id

csfunctions/service/file_upload_schemas.py

@@ -0,0 +1,73 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class PresignedWriteUrls(BaseModel):
+    """
+    Context object for managing upload information.
+    """
+
+    blob_id: str
+    urls: list[str]
+    chunksize: int
+    upload_id: Optional[str] = None
+    etags: Optional[list[str]] = None
+    block_ids: Optional[list[str]] = None
+    headers: Optional[dict[str, str]] = None
+    metadata: Optional[dict[str, str]] = None
+    signature: Optional[str] = None
+
+
+class GeneratePresignedUrlRequest(BaseModel):
+    """
+    Response model for generating presigned URLs.
+    """
+
+    filesize: int = Field(..., description="The size of the file you want to upload in bytes.", ge=0)
+    check_access: bool = Field(..., description="Whether to check access permissions.")
+    persno: str = Field(..., description="The persno of the user who is uploading the file.")
+    lock_id: str = Field(..., description="Provide some random string to lock the file for upload.")
+
+
+class CompleteFileUploadRequest(BaseModel):
+    """
+    Request model for completing a file upload.
+    """
+
+    filesize: int = Field(..., description="The size of the file you want to upload in bytes.", ge=0)
+    check_access: bool = Field(..., description="Whether to check access permissions.")
+    persno: str = Field(..., description="The persno of the user who is uploading the file.")
+    presigned_write_urls: PresignedWriteUrls = Field(..., description="The presigned write URLs for the file upload.")
+    sha256: Optional[str] = Field(None, description="The SHA256 hash of the file content.")
+    lock_id: str = Field(..., description="The lock ID the file was locked with")
+    delete_derived_files: bool = Field(True, description="Whether to delete derived files (e.g. converted pdfs).")
+
+
+class AbortFileUploadRequest(BaseModel):
+    """
+    Request model for aborting a file upload.
+    """
+
+    presigned_write_urls: PresignedWriteUrls = Field(..., description="The presigned write URLs for the file upload.")
+    lock_id: str = Field(..., description="The lock ID the file was locked with")
+    persno: str = Field(..., description="The persno of the user who is uploading the file.")
+
+
+class CreateNewFileRequest(BaseModel):
+    """
+    Request model for creating a new file.
+    """
+
+    parent_object_id: str = Field(..., description="cdb_object_id of the object the file should be attached to.")
+    filename: str = Field(..., description="The name of the file to create.")
+    persno: str = Field(..., description="The persno of the user creating the file.")
+    check_access: bool = Field(..., description="Whether to check access permissions.")
+
+
+class CreateNewFileResponse(BaseModel):
+    """
+    Response model for creating a new file.
+    """
+
+    file_object_id: str = Field(..., description="The cdb_object_id of the newly created file.")

poetry.lock

@@ -34,6 +34,7 @@ mkdocs = "^1.6.1"
 mkdocs-material = "^9.6.14"
 mkdocs-link-marker = "^0.1.3"
 types-requests = "^2.32.4.20250809"
+types-pyyaml = "^6.0.12.20250822"
 
 [tool.ruff]
 line-length = 120