feat: add pytest-env, and use pydantic settings types

pyproject.toml

@@ -19,6 +19,7 @@ requests = "^2.31.0"
 asyncpg = "^0.27.0"
 alembic = "^1.12.0"
 pydantic-settings = "^2.0.3"
+pytest-env = "^1.0.1"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.0.278"
@@ -44,6 +45,18 @@ addopts = [
   "-rfE",
 ]
 testpaths = ["tests"]
+env = [
+  "INST_CONN=postgresql+asyncpg://localhost",
+  "KC_URL=http://localhost",
+  "KC_REALM=",
+  "KC_ADMIN_CLIENT_ID=",
+  "KC_ADMIN_CLIENT_SECRET=",
+  "KC_REALM_URL=http://localhost",
+  "AUTH_URL=http://localhost",
+  "TOKEN_URL=http://localhost",
+  "CERTS_URL=http://localhost",
+  "AUTH_CLIENT=",
+]
 
 [tool.black]
 line-length = 120

src/config.py

@@ -1,7 +1,9 @@
 import os
-from typing import Dict
-from pydantic import TypeAdapter
+from typing import Dict, Any
 
+from pydantic import TypeAdapter
+from pydantic.networks import HttpUrl, PostgresDsn
+from pydantic.types import SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 JWT_OPTS_PREFIX = "jwt_opts_"
@@ -12,17 +14,17 @@
 
 
 class Settings(BaseSettings):
-    inst_conn: str = ""
+    inst_conn: PostgresDsn
     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 = ""
+    auth_client: str
+    auth_url: HttpUrl
+    token_url: HttpUrl
+    certs_url: HttpUrl
+    kc_url: HttpUrl
+    kc_realm: str
+    kc_admin_client_id: str
+    kc_admin_client_secret: SecretStr
+    kc_realm_url: HttpUrl
     jwt_opts: Dict[str, bool | int] = {}
 
     def __init__(self, **data):
@@ -31,22 +33,29 @@ def __init__(self, **data):
 
     def set_jwt_opts(self) -> None:
         """
-        Converts `jwt_opts_` prefixed settings into JWT options dictionary.
+        Converts `jwt_opts_` prefixed settings, and env vars into JWT options dictionary.
         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
+
+        Because we're using model_extra to load in jwt_opts as a dynamic dictionary,
+        normal env overrides does not take place on top of dotenv files,
+        so we're merging settings.model_extra with environment variables.
         """
         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)
+            **self.parse_jwt_vars(jwt_opts_adapter, self.model_extra.items()),
+            **self.parse_jwt_vars(jwt_opts_adapter, os.environ.items()),
+        }
+
+    def parse_jwt_vars(self, type_adapter: TypeAdapter, setting_variables: Dict[str, Any]) -> Dict[str, bool | int]:
+        return {
+            key.lower().replace(JWT_OPTS_PREFIX, ""): type_adapter.validate_python(value)
+            for (key, value) in setting_variables
+            if key.lower().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}]")
+settings = Settings()

src/entities/engine/engine.py

@@ -6,7 +6,7 @@
 from asyncio import current_task
 from config import settings
 
-engine = create_async_engine(settings.inst_conn, echo=True).execution_options(
+engine = create_async_engine(settings.inst_conn.unicode_string(), 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/main.py

@@ -32,7 +32,9 @@ async def general_exception_handler(request: Request, exception: Exception) -> J
     )
 
 
-oauth2_scheme = OAuth2AuthorizationCodeBearer(authorizationUrl=settings.auth_url, tokenUrl=settings.token_url)
+oauth2_scheme = OAuth2AuthorizationCodeBearer(
+    authorizationUrl=settings.auth_url.unicode_string(), tokenUrl=settings.token_url.unicode_string()
+)
 
 app.add_middleware(AuthenticationMiddleware, backend=BearerTokenAuthBackend(oauth2_scheme))
 app.add_middleware(

src/oauth2/oauth2_admin.py

@@ -17,10 +17,10 @@ class OAuth2Admin:
     def __init__(self) -> None:
         self._keys = None
         conn = KeycloakOpenIDConnection(
-            server_url=settings.kc_url,
+            server_url=settings.kc_url.unicode_string(),
             realm_name=settings.kc_realm,
             client_id=settings.kc_admin_client_id,
-            client_secret_key=settings.kc_admin_client_secret,
+            client_secret_key=settings.kc_admin_client_secret.get_secret_value(),
         )
         self._admin = KeycloakAdmin(connection=conn)
 
@@ -29,7 +29,7 @@ def get_claims(self, token: str) -> Dict[str, str] | None:
             return jose.jwt.decode(
                 token=token,
                 key=self._get_keys(),
-                issuer=settings.kc_realm_url,
+                issuer=settings.kc_realm_url.unicode_string(),
                 audience=settings.auth_client,
                 options=settings.jwt_opts,
             )