refactor: change env vars to use pydantic settings

pyproject.toml

@@ -8,7 +8,7 @@ packages = [{ include = "regtech-user-fi-management" }]
 
 [tool.poetry.dependencies]
 python = "^3.11"
-fastapi = "^0.97.0"
+fastapi = "^0.103.1"
 uvicorn = "^0.22.0"
 python-dotenv = "^1.0.0"
 python-keycloak = "^3.0.0"
@@ -18,6 +18,7 @@ python-jose = "^3.3.0"
 requests = "^2.31.0"
 asyncpg = "^0.27.0"
 alembic = "^1.12.0"
+pydantic-settings = "^2.0.3"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.0.278"

src/.env.local

@@ -14,4 +14,6 @@ INST_DB_PWD=fi
 INST_DB_HOST=localhost:5432
 INST_DB_SCHEMA=public
 INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
-JWT_OPTS=verify_at_hash:False,verify_aud:False,verify_iss:False
+JWT_OPTS_VERIFY_AT_HASH="false"
+JWT_OPTS_VERIFY_AUD="false"
+JWT_OPTS_VERIFY_ISS="false"

src/.env.template

@@ -13,5 +13,4 @@ INST_DB_USER=
 INST_DB_PWD=
 INST_DB_HOST=
 INST_DB_SCHEMA=
-INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}
-JWT_OPTS=
+INST_CONN=postgresql+asyncpg://${INST_DB_USER}:${INST_DB_PWD}@${INST_DB_HOST}/${INST_DB_NAME}

src/config.py

@@ -0,0 +1,41 @@
+import os
+from pydantic import TypeAdapter
+
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+JWT_OPTS_PREFIX = "jwt_opts_"
+
+env_files_to_load = [".env"]
+if os.getenv("ENV", "LOCAL") == "LOCAL":
+    env_files_to_load.append(".env.local")
+
+
+class Settings(BaseSettings):
+    inst_conn: str = ""
+    inst_db_schema: str = "public"
+    auth_client: str = ""
+    auth_url: str = ""
+    token_url: str = ""
+    certs_url: str = ""
+    kc_url: str = ""
+    kc_realm: str = ""
+    kc_admin_client_id: str = ""
+    kc_admin_client_secret: str = ""
+    kc_realm_url: str = ""
+
+    def __init__(self, **data):
+        super().__init__(**data)
+        jwt_opts_adapter = TypeAdapter(int | bool)
+        self.jwt_opts = {
+            key.replace(JWT_OPTS_PREFIX, ""): jwt_opts_adapter.validate_python(value)
+            for (key, value) in self.model_extra.items()
+            if key.startswith(JWT_OPTS_PREFIX)
+        }
+
+    model_config = SettingsConfigDict(env_file=env_files_to_load, extra="allow")
+
+
+try:
+    settings = Settings()
+except Exception as e:
+    raise SystemExit(f"failed to set up settings [{e}]")

src/entities/engine/engine.py

@@ -5,10 +5,11 @@
     async_scoped_session,
 )
 from asyncio import current_task
+from config import settings
 
-DB_URL = os.getenv("INST_CONN")
-DB_SCHEMA = os.getenv("INST_DB_SCHEMA", "public")
-engine = create_async_engine(DB_URL, echo=True).execution_options(schema_translate_map={None: DB_SCHEMA})
+engine = create_async_engine(settings.inst_conn, echo=True).execution_options(
+    schema_translate_map={None: settings.inst_db_schema}
+)
 SessionLocal = async_scoped_session(async_sessionmaker(engine, expire_on_commit=False), current_task)
 
 

src/entities/models/dto.py

@@ -15,7 +15,7 @@ class FinancialInsitutionDomainDto(FinancialInsitutionDomainBase):
     lei: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class FinancialInstitutionBase(BaseModel):
@@ -26,7 +26,7 @@ class FinancialInstitutionDto(FinancialInstitutionBase):
     lei: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class FinancialInstitutionWithDomainsDto(FinancialInstitutionDto):
@@ -37,13 +37,13 @@ class DeniedDomainDto(BaseModel):
     domain: str
 
     class Config:
-        orm_mode = True
+        from_attributes = True
 
 
 class UserProfile(BaseModel):
     first_name: str
     last_name: str
-    leis: Optional[Set[str]]
+    leis: Set[str] = None
 
     def to_keycloak_user(self):
         return {"firstName": self.first_name, "lastName": self.last_name}

src/main.py

@@ -1,6 +1,4 @@
-import os
 import logging
-import env  # noqa: F401
 from http import HTTPStatus
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse
@@ -12,6 +10,8 @@
 
 from oauth2 import BearerTokenAuthBackend
 
+from config import settings
+
 log = logging.getLogger()
 
 app = FastAPI()
@@ -32,7 +32,7 @@ async def general_exception_handler(request: Request, exception: Exception) -> J
     )
 
 
-oauth2_scheme = OAuth2AuthorizationCodeBearer(authorizationUrl=os.getenv("AUTH_URL"), tokenUrl=os.getenv("TOKEN_URL"))
+oauth2_scheme = OAuth2AuthorizationCodeBearer(authorizationUrl=settings.auth_url, tokenUrl=settings.token_url)
 
 app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend(oauth2_scheme))
 app.add_middleware(

src/oauth2/oauth2_admin.py

@@ -10,42 +10,19 @@
 
 from keycloak import KeycloakAdmin, KeycloakOpenIDConnection, exceptions as kce
 
-log = logging.getLogger(__name__)
-
-
-def get_jwt_opts(opts_string: str) -> Dict[str, bool | int]:
-    """
-    Parses out the opts_string into JWT options dictionary.
-
-    Args:
-        opts_string (str): comma separated key value pairs in the form of "key1:True,key2:False"
-            all options are boolean, with exception of 'leeway' being int
-            valid options can be found here:
-            https://github.com/mpdavis/python-jose/blob/4b0701b46a8d00988afcc5168c2b3a1fd60d15d8/jose/jwt.py#L81
+from config import settings
 
-    Returns:
-        dict: dictionary of options supported by jwt, mentioned in link above
-    """
-    jwt_opts = {}
-    if opts_string:
-        pairs = opts_string.split(",")
-        for pair in pairs:
-            [key, value] = pair.split(":", 1)
-            jwt_opts[key] = ast.literal_eval(value)
-    return jwt_opts
-
-
-JWT_OPTS = get_jwt_opts(os.getenv("JWT_OPTS", ""))
+log = logging.getLogger(__name__)
 
 
 class OAuth2Admin:
     def __init__(self) -> None:
         self._keys = None
         conn = KeycloakOpenIDConnection(
-            server_url=os.getenv("KC_URL"),
-            realm_name=os.getenv("KC_REALM"),
-            client_id=os.getenv("KC_ADMIN_CLIENT_ID"),
-            client_secret_key=os.getenv("KC_ADMIN_CLIENT_SECRET"),
+            server_url=settings.kc_url,
+            realm_name=settings.kc_realm,
+            client_id=settings.kc_admin_client_id,
+            client_secret_key=settings.kc_admin_client_secret,
         )
         self._admin = KeycloakAdmin(connection=conn)
 
@@ -54,16 +31,16 @@ def get_claims(self, token: str) -> Dict[str, str] | None:
             return jose.jwt.decode(
                 token=token,
                 key=self._get_keys(),
-                issuer=os.getenv("KC_REALM_URL"),
-                audience=os.getenv("AUTH_CLIENT"),
-                options=JWT_OPTS,
+                issuer=settings.kc_realm_url,
+                audience=settings.auth_client,
+                options=settings.jwt_opts,
             )
         except jose.ExpiredSignatureError:
             pass
 
     def _get_keys(self) -> Dict[str, Any]:
         if self._keys is None:
-            response = requests.get(os.getenv("CERTS_URL"))
+            response = requests.get(settings.certs_url)
             self._keys = response.json()
         return self._keys