refactor: streamline session management and remove unused code

Author: getmarkus

.env.example

@@ -0,0 +1,23 @@
+# Application Settings
+APP_PROJECT_NAME=python-template
+# sqlite:// sqlite:///./test.db
+APP_DATABASE_URL=postgresql+psycopg://app_user:change_this_password@localhost:5432/app_database
+APP_DATABASE_SCHEMA=issue_analysis
+APP_MIGRATE_DATABASE=false
+APP_SQLITE_WAL_MODE=false
+# collection, sqlmodel
+APP_DATABASE_TYPE=sqlmodel
+
+# PostgreSQL Configuration (for reference)
+POSTGRES_SUPER_USER=app_user
+POSTGRES_SUPER_PASSWORD=change_this_password
+POSTGRES_DB=app_database
+POSTGRES_HOST=localhost
+POSTGRES_PORT=5432
+
+# CORS Settings
+APP_BACKEND_CORS_ORIGINS=["http://localhost:8000","http://localhost:3000"]
+APP_CORS_ALLOW_CREDENTIALS=false
+APP_CORS_ALLOW_METHODS=["*"]
+APP_CORS_ALLOW_HEADERS=["X-Forwarded-For","Authorization","Content-Type","X-Request-ID"]
+APP_CORS_EXPOSE_HEADERS=["X-Process-Time","X-Request-ID"]

.gitignore

@@ -121,6 +121,7 @@ celerybeat.pid
 
 # Environments
 .env
+.env.testing
 .venv
 env/
 venv/
@@ -204,4 +205,5 @@ Temporary Items
 *.sqlite3
 *.sqlite-journal
 *.sqlite3-journal
-*.db-journal
+*.db-journal
+*.sql

.vscode/settings.json

@@ -1,11 +1,5 @@
 {
-    "python.testing.pytestArgs": [
-        "app"
-    ],
-    "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true,
-    "launch": {
-        "configurations": [],
-        "compounds": []
-    }
-}
+  "python.testing.pytestArgs": ["app"],
+  "python.testing.unittestEnabled": false,
+  "python.testing.pytestEnabled": true
+}

README.md

@@ -11,9 +11,12 @@ Simple python template I am experimenting with around a set of overlapping conce
 uv run fastapi dev main.py
 uv run pytest
 
-uvx dynaconf list
-
 atlas schema inspect -u "sqlite://issues.db" --format "{{ sql . }}" > migrate.sql
+
+atlas schema apply --url "sqlite://issues.db" --to "file://migrate.sql" --dev-url "sqlite://file?mode=memory" --dry-run
+atlas schema apply --url "sqlite://issues.db" --to "file://migrate.sql" --dev-url "sqlite://file?mode=memory"
+
+atlas schema apply --url "postgres://postgres:pass@localhost:5432/database?search_path=public&sslmode=disable" --to "file://schema.sql" --dev-url "docker://postgres/17"
 ```
 
 ```mermaid

app/core/factory.py

@@ -1,28 +1,20 @@
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
 
 from app.interface_adapters import api_router
-from app.interface_adapters.containers import Container
 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
+from config import Settings
 
 
-def create_container():
-    container = Container()
-    container.wire(packages=["app"])
-    return container
-
-
-def create_app(lifespan_handler=None) -> FastAPI:
+def create_app(settings: Settings, lifespan_handler=None) -> FastAPI:
     if lifespan_handler is None:
 
         @asynccontextmanager
         async def default_lifespan(app: FastAPI):
-            # Initialize container and wire dependencies
-            container = create_container()
 
             # Initialize database if using SQLModel
             get_engine()
@@ -32,7 +24,6 @@ async def default_lifespan(app: FastAPI):
             yield
 
             app.state.running = False
-            container.unwire()
 
         lifespan_handler = default_lifespan
 
@@ -48,4 +39,18 @@ async def default_lifespan(app: FastAPI):
     # Register routes
     app.include_router(api_router, prefix="/v1")
 
+    # Configure CORS
+    app.add_middleware(
+        CORSMiddleware,
+        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,
+    )
+
     return app

app/domain/aggregate_root.py

@@ -4,8 +4,6 @@
 
 from sqlmodel import SQLModel
 
-from config import settings
-
 
 class BaseCommand(Protocol):
     command_id: str
@@ -21,11 +19,11 @@ class BaseEvent(Protocol):
 
 
 class AggregateRoot(SQLModel, abc.ABC):
-    # This sets the database schema in which to create tables for all subclasses of BaseModel
-    # __tablename__ = "sometable"
-    __table_args__ = {"schema": settings.database_schema}
     version: int = 0
 
+    def __init__(self, **data):
+        super().__init__(**data)
+
     # or handle()
     @abc.abstractmethod
     def process(self, command: BaseCommand) -> list[BaseEvent]:

app/interface_adapters/containers.py

@@ -1,23 +1,23 @@
 from typing import Annotated
 
-from dependency_injector.wiring import inject
 from fastapi import APIRouter, Depends
 from loguru import logger
 
 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.containers import Container
+from app.resource_adapters.persistence.sqlmodel.database import SessionDep
+from app.resource_adapters.persistence.sqlmodel.issues import SQLModelIssueRepository
 
 router = APIRouter()
 
 
-def get_repository() -> IssueRepository:
-    return Container.issue_repository()
+def get_repository(session: SessionDep) -> IssueRepository:
+    with SQLModelIssueRepository(session) as repo:
+        yield repo
 
 
 @router.post("/issues/{issue_number}/analyze", response_model=Issue)
-@inject
 def analyze_issue(
     issue_number: int,
     repo: Annotated[IssueRepository, Depends(get_repository)],

app/interface_adapters/endpoints/issues.py

@@ -1,30 +1,41 @@
-from typing import Generator
+from typing import Annotated, Generator
 
+from fastapi import Depends
 from loguru import logger
-from sqlalchemy import inspect, text
+from sqlalchemy import inspect, schema, text
 from sqlalchemy.engine import Engine
 from sqlalchemy.pool import StaticPool
 from sqlmodel import Session, SQLModel, create_engine
 
-from config import settings
+from config import Settings, get_settings
 
 _engine: Engine | None = None
 
 
-def get_db() -> Generator[Session, None, None]:
+def get_session(
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> Generator[Session, Settings, None]:
     with Session(_engine) as session:
+        session.connection(
+            execution_options={"schema_translate_map": {None: settings.database_schema}}
+        )
         yield session
 
 
-def get_engine(database_url: str | None = None) -> Engine:
+SessionDep = Annotated[Session, Depends(get_session)]
+
+
+def get_engine(_settings: Settings | None = None) -> Engine:
     """Get or create SQLModel engine instance."""
     global _engine
 
     if _engine is not None:
         return _engine
 
-    if database_url is None:
-        database_url = settings.database_url
+    if _settings is None:
+        _settings = get_settings()
+
+    database_url = _settings.database_url
 
     # Configure engine based on database type
     engine_args = {"echo": True}
@@ -38,29 +49,38 @@ def get_engine(database_url: str | None = None) -> Engine:
     _engine = create_engine(database_url, **engine_args)
 
     # Enable WAL mode if configured
-    if settings.sqlite_wal_mode:
+    if _settings.sqlite_wal_mode:
         with _engine.connect() as conn:
             # https://www.sqlite.org/pragma.html
             conn.execute(text("PRAGMA journal_mode=WAL"))
             # conn.execute(text("PRAGMA synchronous=OFF"))
             logger.info("SQLite WAL mode enabled")
 
     # Initialize database if using SQLModel
-    if settings.model_config == "sqlmodel" and not settings.migrate_database:
+    if (
+        _settings.database_type == "sqlmodel"
+        and _settings.database_schema
+        and not _settings.database_url.startswith("sqlite")
+    ):
+        SQLModel.metadata.schema = _settings.database_schema
+        with _engine.connect() as conn:
+            conn.execution_options = {
+                "schema_translate_map": {None: _settings.database_schema}
+            }
+            if not conn.dialect.has_schema(conn, _settings.database_schema):
+                logger.warning(
+                    f"Schema '{_settings.database_schema}' not found in database. Creating..."
+                )
+                conn.execute(schema.CreateSchema(_settings.database_schema))
+                conn.commit()
+
+    if _settings.database_type == "sqlmodel" and not _settings.migrate_database:
         logger.info("Creating database tables...")
 
-        if settings.database_schema:
-            SQLModel.metadata.schema = settings.database_schema
-            # with self.engine.connect() as conn:
-            # if not conn.dialect.has_schema(conn, db_schema):
-            # logger.warning(f"Schema '{db_schema}' not found in database. Creating...")
-            # conn.execute(sa.schema.CreateSchema(db_schema))
-            # conn.commit()
-
         # Check if tables exist before creating them
         inspector = inspect(_engine)
         existing_tables = inspector.get_table_names(
-            schema=settings.database_schema if settings.database_schema else None
+            schema=_settings.database_schema if _settings.database_schema else None
         )
         if not existing_tables:
             # if settings.database_schema:

app/resource_adapters/persistence/sqlmodel/database.py

@@ -1,28 +1,33 @@
 import os
 from contextlib import asynccontextmanager
+from pathlib import Path
 
 import pytest
 from _pytest.config import Config
+from dotenv import load_dotenv
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from loguru import logger
 from sqlmodel import Session, delete
 
 from app.core.factory import create_app
 from app.resource_adapters.persistence.sqlmodel.database import get_engine
 from app.resource_adapters.persistence.sqlmodel.issues import Issue
-from config import settings
+from config import Settings
 
+# Specify the custom .env file
+dotenv_path = Path(".env.testing")
+load_dotenv(dotenv_path=dotenv_path, override=True)
 
-@pytest.fixture(scope="session", autouse=True)
-def set_test_settings():
-    """Fixture to force the use of the 'testing' environment for all tests."""
-    settings.configure(FORCE_ENV_FOR_DYNACONF="testing")
+
+settings = Settings()
 
 
 def pytest_unconfigure(config: Config) -> None:
     """Clean up after each test."""
     # Remove test database if it exists
     if os.path.exists("test.db"):
+        logger.info("Removing test database")
         os.remove("test.db")
 
 
@@ -37,13 +42,13 @@ async def test_lifespan(app: FastAPI):
 @pytest.fixture(name="app")
 def test_app():
     """Create test app instance only during test execution."""
-    return create_app(lifespan_handler=test_lifespan)
+    return create_app(settings, lifespan_handler=test_lifespan)
 
 
 @pytest.fixture(name="session")
 def test_session():
     """Session fixture for testing environment using test database."""
-    with Session(get_engine()) as session:
+    with Session(get_engine(settings)) as session:
         yield session
         statement = delete(Issue)
         session.exec(statement)