models

Author: sanderegg

services/datcore-adapter/src/simcore_service_datcore_adapter/models/domains/__init__.py renamed to services/datcore-adapter/src/simcore_service_datcore_adapter/errors/__init__.py

@@ -0,0 +1,32 @@
+from botocore.exceptions import ClientError
+from fastapi import FastAPI
+from fastapi.encoders import jsonable_encoder
+from servicelib.fastapi.http_error import set_app_default_http_error_handlers
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_500_INTERNAL_SERVER_ERROR
+
+
+async def botocore_exceptions_handler(
+    _: Request,
+    exc: Exception,
+) -> JSONResponse:
+    assert isinstance(exc, ClientError)  # nosec
+    assert "Error" in exc.response  # nosec
+    assert "Code" in exc.response["Error"]  # nosec
+    error_content = {"errors": [f"{exc}"]}
+    if exc.response["Error"]["Code"] == "NotAuthorizedException":
+        return JSONResponse(
+            content=jsonable_encoder({"error": error_content}),
+            status_code=HTTP_401_UNAUTHORIZED,
+        )
+    return JSONResponse(
+        content=jsonable_encoder({"error": error_content}),
+        status_code=HTTP_500_INTERNAL_SERVER_ERROR,
+    )
+
+
+def set_exception_handlers(app: FastAPI) -> None:
+    set_app_default_http_error_handlers(app)
+
+    app.add_exception_handler(ClientError, botocore_exceptions_handler)

services/datcore-adapter/src/simcore_service_datcore_adapter/errors/handlers.py

@@ -0,0 +1,36 @@
+import datetime
+from pathlib import Path
+from typing import Annotated
+
+from models_library.api_schemas_datcore_adapter.datasets import PackageMetaData
+from pydantic import AnyUrl, BaseModel, ByteSize, Field
+
+
+class FileDownloadOut(BaseModel):
+    link: AnyUrl
+
+
+class DatCorePackageMetaData(BaseModel):
+    id: int
+    path: Path
+    display_path: Path
+    package_id: Annotated[str, Field(alias="packageId")]
+    name: str
+    filename: str
+    s3_bucket: Annotated[str, Field(alias="s3bucket")]
+    size: ByteSize
+    created_at: Annotated[datetime.datetime, Field(alias="createdAt")]
+    updated_at: Annotated[datetime.datetime, Field(alias="updatedAt")]
+
+    def to_api_model(self) -> PackageMetaData:
+        return PackageMetaData(
+            path=self.path,
+            display_path=self.display_path,
+            package_id=self.package_id,
+            name=self.name,
+            filename=self.filename,
+            s3_bucket=self.s3_bucket,
+            size=self.size,
+            created_at=self.created_at,
+            updated_at=self.updated_at,
+        )