updating file management auth. api failsafe if env's are not present

Author: amyfromandi

services/api-v3/api/routes/object.py

@@ -6,20 +6,46 @@
 import starlette.requests
 from api.database import get_async_session, get_engine
 from api.routes.security import has_access
-from api.settings import settings
 from fastapi import (
     APIRouter,
     Depends,
-    File,
     HTTPException,
-    Request,
-    Response,
-    UploadFile,
 )
 from minio import Minio
 from sqlalchemy import text
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from functools import lru_cache
 from starlette.datastructures import UploadFile as StarletteUploadFile
 
+
+class FileSettings(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="ignore",
+    )
+
+    s3_access_key: str = Field(alias="S3_ACCESS_KEY")
+    s3_secret_key: str = Field(alias="S3_SECRET_KEY")
+    s3_bucket: str = Field(alias="BUCKET")
+    s3_endpoint: str = Field(alias="S3_HOST")
+    s3_secure: bool = Field(default=True, alias="S3_SECURE")
+
+
+@lru_cache
+def get_file_settings() -> FileSettings:
+    try:
+        return FileSettings()
+    except Exception as e:
+        print(e)
+        raise HTTPException(
+            status_code=503,
+            detail=f"File management is not configured.",
+        )
+
+
+
 router = APIRouter(
     prefix="/object",
     tags=["file"],
@@ -51,6 +77,7 @@ def sha256_of_uploadfile(
 
 
 def get_s3_client():
+    settings = get_file_settings()
     return Minio(
         endpoint=settings.s3_endpoint,
         access_key=settings.s3_access_key,
@@ -66,6 +93,7 @@ def get_storage_host_bucket() -> tuple[str, str]:
     Also ensures host has no port (hostname only) if storage_host ever includes one.
     """
     import urllib.parse
+    settings = get_file_settings()
 
     raw_host = settings.s3_endpoint
     parsed = urllib.parse.urlparse(
@@ -214,7 +242,6 @@ async def get_object(id: int):
 async def create_object(
     request: starlette.requests.Request,
     user_has_access: bool = Depends(has_access),
-    object: UploadFile | None = File(default=None),
 ):
     """
     Upload to s3 and register in storage.objects.

services/api-v3/api/routes/security.py

@@ -358,6 +358,7 @@ async def redirect_callback(code: str, state: Optional[str] = None):
             )
 
             # validate jwt https://dev.macrostrat.org/dev/me
+            #TODO remove the groups and sub and add the user_name
             access_token = create_access_token(
                 data={
                     "sub": user.sub,
@@ -397,7 +398,7 @@ async def redirect_callback(code: str, state: Optional[str] = None):
                 samesite=samesite,
                 secure=secure,
             )
-
+            #TODO remove the token type
             refresh_jwt = jwt.encode(
                 {
                     "user_id": user.id,
@@ -463,6 +464,7 @@ async def refresh_token(
         else "web_user"
     )
     # setting new access cookie
+    #TODO can remove groups, sub. add user_name.
     access_token = create_access_token(
         data={"sub": user.sub, "role": role, "user_id": user.id, "groups": list(ids)}
     )

services/api-v3/api/settings.py

@@ -1,46 +1,24 @@
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-
 class Settings(BaseSettings):
-    """
-    Centralized application settings for API v3.
-    Values are loaded from environment variables and .env.
-    """
-
     model_config = SettingsConfigDict(
         env_file=".env",
         env_file_encoding="utf-8",
         extra="ignore",
     )
 
-    # Environment
     environment: str = Field(default="development", alias="ENVIRONMENT")
-
-    # Database
     database_uri: str = Field(alias="uri")
 
-    # OAuth
     redirect_uri: str = Field(alias="REDIRECT_URI_ENV")
     oauth_authorization_url: str = Field(alias="OAUTH_AUTHORIZATION_URL")
     oauth_token_url: str = Field(alias="OAUTH_TOKEN_URL")
     oauth_userinfo_url: str = Field(alias="OAUTH_USERINFO_URL")
     oauth_client_id: str = Field(alias="OAUTH_CLIENT_ID")
     oauth_client_secret: str = Field(alias="OAUTH_CLIENT_SECRET")
 
-    # JWT
     jwt_secret_key: str = Field(alias="SECRET_KEY")
-    jwt_algorithm: str = Field(
-        default="HS256",
-        alias="JWT_ENCRYPTION_ALGORITHM",
-    )
-
-    # S3 / MinIO
-    s3_access_key: str = Field(alias="ACCESS_KEY")
-    s3_secret_key: str = Field(alias="SECRET_KEY")
-    s3_bucket: str = Field(alias="BUCKET")
-    s3_endpoint: str = Field(alias="S3_HOST")
-    s3_secure: bool = Field(default=True, alias="S3_SECURE")
-
+    jwt_algorithm: str = Field(default="HS256", alias="JWT_ENCRYPTION_ALGORITHM")
 
 settings = Settings()