Remove need for decopule dep

Author: ChrisLovering

api/__init__.py

@@ -1,12 +1,12 @@
 import logging
 
-from api import constants
+from api import settings
 
 format_string = "[%(asctime)s] [%(process)d] [%(levelname)s] %(name)s - %(message)s"
 date_format_string = "%Y-%m-%d %H:%M:%S %z"
 
 logging.basicConfig(
-    format=format_string, datefmt=date_format_string, level=getattr(logging, constants.Config.LOG_LEVEL.upper())
+    format=format_string, datefmt=date_format_string, level=getattr(logging, settings.Server.LOG_LEVEL.upper())
 )
 
 logging.getLogger().info("Logging initialization complete")

api/constants.py

@@ -2,17 +2,14 @@
 
 from fastapi import Depends
 from sqlalchemy import BigInteger, Boolean, Column, Enum, ForeignKey, Index, Integer, PrimaryKeyConstraint, Text, text
-from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import relationship, sessionmaker
+from sqlalchemy.orm import relationship
 
-from api.constants import Config
 from api.dependencies import get_db_session
 
-engine = create_async_engine(Config.DATABASE_URL)
 Base = declarative_base()
 
-Session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
 DBSession = Annotated[AsyncSession, Depends(get_db_session)]
 
 

api/database.py

@@ -1,15 +1,15 @@
 from fastapi import FastAPI
 from starlette.middleware.authentication import AuthenticationMiddleware
 
-from api.constants import Config
 from api.middleware import TokenAuthentication, on_auth_error
 from api.routers import codejams, infractions, teams, users, winners
+from api.settings import Server
 
 app = FastAPI(redoc_url="/", docs_url="/swagger")
 
 app.add_middleware(
     AuthenticationMiddleware,
-    backend=TokenAuthentication(token=Config.TOKEN),
+    backend=TokenAuthentication(token=Server.API_TOKEN.get_secret_value()),
     on_error=on_auth_error,
 )
 

api/main.py

@@ -3,7 +3,7 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from api.constants import Config
+from api.settings import Server
 
 NO_AUTHORIZATION_HEADER = "no `Authorization` header in request."
 INVALID_CREDENTIALS = "invalid credentials."
@@ -17,7 +17,7 @@ def __init__(self, token: str) -> None:
 
     async def authenticate(self, request: Request) -> tuple[AuthCredentials, SimpleUser]:
         """Authenticate the request based on the Authorization header."""
-        if Config.DEBUG:
+        if Server.DEBUG:
             credentials = AuthCredentials(scopes=["debug"])
             user = SimpleUser(username="api_client")
             return credentials, user

api/middleware.py

@@ -0,0 +1,32 @@
+import pydantic
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from api.utils.settings import CJMSBaseSettings
+
+
+class _ConnectionURLs(CJMSBaseSettings):
+    """URLs for connecting to other internal services."""
+
+    DATABASE_URL: pydantic.SecretStr
+    SENTRY_DSN: pydantic.SecretStr | None
+
+
+ConnectionURLs = _ConnectionURLs()
+
+
+class Connections:
+    """How to connect to other, internal services."""
+
+    DB_ENGINE = create_async_engine(ConnectionURLs.DATABASE_URL.get_secret_value())
+    DB_SESSION = async_sessionmaker(DB_ENGINE)
+
+
+class _Server(CJMSBaseSettings):
+    """Basic configuration for the Code Jam Management System."""
+
+    LOG_LEVEL: str = "INFO"
+    DEBUG: bool = False
+    API_TOKEN: pydantic.SecretStr
+
+
+Server = _Server(API_TOKEN="badbot13m0n8f570f942013fc818f234916ca531")

api/settings.py

@@ -0,0 +1,57 @@
+"""Utilities for working with the project's constants."""
+import logging
+import typing
+from collections.abc import Sequence
+
+import pydantic
+from pydantic.error_wrappers import ErrorWrapper
+
+logger = logging.getLogger("cjms.settings")
+
+# This is available in pydantic as pydantic.error_wrappers.ErrorList
+# but is typehinted as a Sequence[any], due to being a recursive type.
+# This makes it harder to handle the types.
+# For our purposes, a fully accurate representation is not necessary.
+_PYDANTIC_ERROR_TYPE = Sequence[ErrorWrapper | Sequence[ErrorWrapper]]
+
+
+class CJMSBaseSettings(pydantic.BaseSettings):
+    """Base class for settings with .env support and nicer error messages."""
+
+    @staticmethod
+    def __log_missing_errors(base_error: pydantic.ValidationError, errors: _PYDANTIC_ERROR_TYPE) -> bool:
+        """
+        Log out a nice representation for missing environment variables.
+
+        Returns false if none of the errors were caused by missing variables.
+        """
+        found_relevant_errors = False
+        for error in errors:
+            if isinstance(error, Sequence):
+                found_relevant_errors = (
+                    CJMSBaseSettings.__log_missing_errors(base_error, error) or found_relevant_errors
+                )
+            elif isinstance(error.exc, pydantic.MissingError):
+                logger.error(f"Missing environment variable {base_error.args[1].__name__}.{error.loc_tuple()[0]}")
+                found_relevant_errors = True
+
+        return found_relevant_errors
+
+    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        """Try to instantiate the class, and print a nicer message for unset variables."""
+        try:
+            super().__init__(*args, **kwargs)
+        except pydantic.ValidationError as error:
+            if CJMSBaseSettings.__log_missing_errors(error, error.raw_errors):
+                exit(1)
+            else:
+                # The validation error is not due to an unset environment variable, propagate the error as normal
+                raise error from None
+
+    class Config:
+        """Enable env files."""
+
+        frozen = True
+
+        env_file = ".env"
+        env_file_encoding = "utf-8"

api/utils/settings.py

@@ -253,9 +253,6 @@ pydantic==1.10.6 ; python_full_version >= "3.11.0" and python_full_version < "3.
 python-dateutil==2.8.2 ; python_full_version >= "3.11.0" and python_full_version < "3.12.0" \
     --hash=sha256:0123cacc1627ae19ddf3c27a5de5bd67ee4586fbdd6440d9748f8abb483d3e86 \
     --hash=sha256:961d03dc3453ebbc59dbdea9e4e11c5651520a876d0f4db161e8674aae935da9
-python-decouple==3.8 ; python_full_version >= "3.11.0" and python_full_version < "3.12.0" \
-    --hash=sha256:ba6e2657d4f376ecc46f77a3a615e058d93ba5e465c01bbe57289bfb7cce680f \
-    --hash=sha256:d0d45340815b25f4de59c974b855bb38d03151d81b037d9e3f463b0c9f8cbd66
 python-dotenv==1.0.0 ; python_full_version >= "3.11.0" and python_full_version < "3.12.0" \
     --hash=sha256:a8df96034aae6d2d50a4ebe8216326c61c3eb64836776504fcca410e5937a3ba \
     --hash=sha256:f5971a9226b701070a4bf2c38c89e5a3f0d64de8debda981d1db98583009122a

main-requirements.txt

@@ -8,7 +8,7 @@
 
 # This is a required step by Alembic to properly generate migrations
 from api import database
-from api.constants import Config
+from api.settings import ConnectionURLs
 
 target_metadata = database.Base.metadata
 
@@ -19,7 +19,7 @@
 # Interpret the config file for Python logging.
 # This line sets up loggers basically.
 fileConfig(config.config_file_name)
-config.set_main_option("sqlalchemy.url", Config.DATABASE_URL)
+config.set_main_option("sqlalchemy.url", ConnectionURLs.DATABASE_URL.get_secret_value())
 
 
 def run_migrations_offline() -> None: