refactor: simplify settings retrieval and enhance info endpoint

Author: getmarkus

app/interface_adapters/endpoints/issues.py

@@ -4,14 +4,14 @@
 from loguru import logger
 from sqlmodel import Session
 
-from config import Settings
 from app.core.ports.repositories.issues import IssueRepository
 from app.core.usecases.analyze_issue import AnalyzeIssue
 from app.domain.issue import Issue
 from app.interface_adapters.exceptions import UnsupportedOperationException
 from app.resource_adapters.persistence.in_memory.issues import InMemoryIssueRepository
 from app.resource_adapters.persistence.sqlmodel.database import get_db
 from app.resource_adapters.persistence.sqlmodel.issues import SQLModelIssueRepository
+from config import settings
 
 router = APIRouter()
 
@@ -20,7 +20,7 @@
 def configure_repository(
     session: Annotated[Session, Depends(get_db)]
 ) -> IssueRepository:
-    execution_mode = Settings.get_settings().execution_mode
+    execution_mode = settings.execution_mode
     if execution_mode == "in-memory":
         return InMemoryIssueRepository()
     elif execution_mode == "sqlmodel":
@@ -34,8 +34,7 @@ def configure_repository(
 
 @router.post("/issues/{issue_number}/analyze", response_model=Issue)
 def analyze_issue(
-    issue_number: int,
-    repo: Annotated[IssueRepository, Depends(configure_repository)]
+    issue_number: int, repo: Annotated[IssueRepository, Depends(configure_repository)]
 ) -> Issue:
     logger.info(f"analyzing issue: {issue_number}")
     use_case = AnalyzeIssue(issue_number=issue_number, repo=repo)

app/resource_adapters/persistence/sqlmodel/database.py

@@ -4,7 +4,7 @@
 from sqlalchemy.engine import Engine
 from sqlmodel import Session, SQLModel, create_engine
 
-from config import Settings
+from config import settings
 
 _engine: Engine | None = None
 
@@ -22,14 +22,11 @@ def get_engine(database_url: str | None = None) -> Engine:
         return _engine
 
     if database_url is None:
-        database_url = Settings.get_settings().database_url
+        database_url = settings.database_url
 
     _engine = create_engine(database_url, echo=True)
     # Initialize database if using SQLModel
-    if (
-        Settings.get_settings().execution_mode == "sqlmodel"
-        and not Settings.get_settings().migrate_database
-    ):
+    if settings.execution_mode == "sqlmodel" and not settings.migrate_database:
         logger.info("Creating database tables...")
         SQLModel.metadata.create_all(_engine)
         logger.info("Database tables created successfully")

app/tests/conftest.py

@@ -22,4 +22,3 @@ def client_fixture():
 
     with TestClient(app) as client:
         yield client
-

config.py

@@ -1,40 +1,41 @@
-import tomllib
 from functools import lru_cache
 from typing import List
 
+from dynaconf import Dynaconf, Validator
 from pydantic import AnyHttpUrl
-from pydantic_settings import BaseSettings, SettingsConfigDict
 
+# order matters
+# [default] settings in settings.toml are loaded first
+# [APP_ENV] settings override defaults
+# .env file overrides those
+# Environment variables have the highest priority
+# CLI arguments override those
 
-# https://fastapi.tiangolo.com/advanced/settings/
-# https://docs.pydantic.dev/latest/concepts/pydantic_settings/
-# https://docs.pydantic.dev/latest/concepts/pydantic_settings/#command-line-support
-class Settings(BaseSettings):
-    # https://stackoverflow.com/questions/70852331/how-to-define-class-attributes-after-inheriting-pydantics-basemodel
-    # https://stackoverflow.com/questions/69388833/triggering-a-function-on-creation-of-an-pydantic-object
-    pyproject_file: str = ""
-    execution_mode: str = ""
-    env_smoke_test: str = ""
-    project_name: str = ""
-    database_url: str = "sqlite:///./issues.db"
-    migrate_database: bool = False
-    # BACKEND_CORS_ORIGINS is a JSON-formatted list of origins
-    # e.g: '["http://localhost", "http://localhost:4200", "http://localhost:3000", \
-    # "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
-    backend_cors_origins: List[AnyHttpUrl] = []
+settings = Dynaconf(
+    envvar_prefix="APP",
+    settings_files=["settings.toml", ".secrets.toml"],
+    environments=True,
+    load_dotenv=True,
+    validators=[
+        Validator("database_url", must_exist=True),
+        Validator("project_name", must_exist=True),
+        Validator("migrate_database", is_type_of=bool),
+        Validator("backend_cors_origins", is_type_of=list),
+    ],
+)
 
-    model_config = SettingsConfigDict(env_file=".env", extra="allow")
-    # `.env.prod` takes priority over `.env`
-    #   env_file=('.env', '.env.prod')
 
-    def setFromTOML(self):
-        with open(self.pyproject_file, "rb") as f:
-            _META = tomllib.load(f)
-        self.project_name = _META["project"]["name"]
+# Singleton pattern to ensure only one settings instance
+@lru_cache()
+def get_settings():
+    return settings
 
-    @classmethod
-    @lru_cache
-    def get_settings(cls):
-        settings = Settings(pyproject_file="pyproject.toml")
-        settings.setFromTOML()
-        return settings
+
+# Type hints for your settings (optional but recommended)
+class Settings:
+    project_name: str
+    database_url: str
+    migrate_database: bool
+    backend_cors_origins: List[AnyHttpUrl]
+    execution_mode: str
+    env_smoke_test: str

main.py

@@ -1,28 +1,30 @@
 import datetime
 import uuid
 from contextlib import asynccontextmanager
-from typing import Annotated, Any, Dict, Optional
+from typing import Any, Dict, Optional
 
-from fastapi import Depends, FastAPI
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from loguru import logger
 from pydantic import BaseModel, Field
 
-from config import Settings
 from app.interface_adapters import api_router
 from app.interface_adapters.exceptions import AppException
 from app.interface_adapters.middleware.error_handler import app_exception_handler
 from app.resource_adapters.persistence.sqlmodel.database import get_engine
+from config import settings
 
 # https://brandur.org/logfmt
 # https://github.com/Delgan/loguru
 # https://betterstack.com/community/guides/logging/loguru/
 
+
 def isRunning() -> bool:
-    #logger.info(f"App state: {dict(app.state.__dict__)}")
+    # logger.info(f"App state: {dict(app.state.__dict__)}")
     running = getattr(app.state, "running", False)
     logger.info(f"Running state: {running}")
+    logger.info(f"Environment: {settings.current_env}")
     return running
 
 
@@ -35,8 +37,7 @@ async def lifespan(app: FastAPI):
 
     app.state.running = True
     logger.info("Lifespan started")
-    logger.info(f"App state: {dict(app.state.__dict__)}")
-    logger.info(app.state.running)
+    isRunning()
 
     yield
 
@@ -45,8 +46,8 @@ async def lifespan(app: FastAPI):
 
 
 def is_configured() -> bool:
-    logger.info(f"configured: {Settings.get_settings().env_smoke_test == "configured"}")
-    return Settings.get_settings().env_smoke_test == "configured"
+    logger.info(f"configured: {settings.env_smoke_test == 'configured'}")
+    return settings.env_smoke_test == "configured"
 
 
 """ {
@@ -86,7 +87,7 @@ def create_health_response(is_healthy: bool, check_name: str) -> JSONResponse:
     status = "pass" if is_healthy else "fail"
     response = HealthCheck(
         status=status,
-        version=Settings.get_settings().project_name,
+        version=settings.project_name,
         description=f"Service {status}",
         checks={
             check_name: {
@@ -105,7 +106,7 @@ def create_health_response(is_healthy: bool, check_name: str) -> JSONResponse:
 
 app = FastAPI(
     lifespan=lifespan,
-    title=Settings.get_settings().project_name,
+    title=settings.project_name,  # Use settings directly
     openapi_url="/v1/openapi.json",
 )
 
@@ -129,6 +130,7 @@ def create_health_response(is_healthy: bool, check_name: str) -> JSONResponse:
 # gzip compression - GZipMiddleware
 # ssl enforcement - HTTPSRedirectMiddleware
 
+
 @app.middleware("http")
 async def add_request_id(request, call_next):
     request_id = str(uuid.uuid4())
@@ -138,47 +140,49 @@ async def add_request_id(request, call_next):
     logger.info(f"Request {request_id} to {request.url.path}")
     return response
 
+
 @app.middleware("http")
 async def add_process_time_header(request, call_next):
     import time
+
     start_time = time.perf_counter()
     response = await call_next(request)
     process_time = time.perf_counter() - start_time
     response.headers["X-Process-Time"] = str(process_time)
     logger.info(f"Request to {request.url.path} took {process_time:.4f} seconds")
     return response
 
+
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],  # Set this to lock down if applicable to your use case
-    # allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
-    allow_credentials=False,  # Must be false if all origins (*) is allowed
-    allow_methods=["*"],
-    allow_headers=["X-Forwarded-For", "Authorization", "Content-Type", "X-Request-ID"],
-    expose_headers=["X-Process-Time", "X-Request-ID"]
+    allow_origins=(
+        [str(origin) for origin in settings.backend_cors_origins]
+        if settings.backend_cors_origins
+        else ["*"]
+    ),
+    allow_credentials=settings.cors_allow_credentials,
+    allow_methods=settings.cors_allow_methods,
+    allow_headers=settings.cors_allow_headers,
+    expose_headers=settings.cors_expose_headers,
 )
 
 
 @app.get("/")
-async def root(
-    settings: Annotated[Settings, Depends(Settings.get_settings)],
-) -> Dict[str, Any]:
+async def root() -> Dict[str, Any]:
     return {
         "app_name": settings.project_name,
         "system_time": datetime.datetime.now(),
     }
 
 
 @app.get("/info")
-async def info(
-    settings: Annotated[Settings, Depends(Settings.get_settings)],
-) -> Dict[str, Any]:
-    logger.info(settings.model_dump())
+async def info() -> Dict[str, Any]:
     return {
         "app_name": settings.project_name,
         "system_time": datetime.datetime.now(),
         "execution_mode": settings.execution_mode,
         "env_smoke_test": settings.env_smoke_test,
+        "env": settings.current_env,
     }
 
 

settings.toml

@@ -0,0 +1,32 @@
+[default]
+project_name = "python-template"
+database_url = "sqlite:///./issues.db"
+migrate_database = false
+execution_mode = "sqlmodel"
+env_smoke_test = ""
+
+# CORS Settings
+backend_cors_origins = []
+cors_allow_credentials = false
+cors_allow_methods = ["*"]
+cors_allow_headers = [
+  "X-Forwarded-For",
+  "Authorization",
+  "Content-Type",
+  "X-Request-ID",
+]
+cors_expose_headers = ["X-Process-Time", "X-Request-ID"]
+
+[development]
+# Development-specific settings
+backend_cors_origins = ["http://localhost:8000", "http://localhost:3000"]
+
+[production]
+# Production-specific settings
+backend_cors_origins = [] # Set specific origins in production
+cors_allow_methods = [
+  "GET",
+  "POST",
+  "PUT",
+  "DELETE",
+] # Be more restrictive in production