wip

Author: getmarkus

.env.example

@@ -1,7 +1,11 @@
 # 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
+# sqlite:// will silently swallow exceptions, better not to use for in-memory database
+# APP_DATABASE_URL=sqlite:// 
+# sqlite:///:memory: follows normal exception handling rules
+APP_DATABASE_URL=sqlite:///:memory:
+# APP_DATABASE_URL=sqlite:///./issues.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

.trunk/trunk.yaml

@@ -2,7 +2,7 @@
 # To learn more about the format of this file, see https://docs.trunk.io/reference/trunk-yaml
 version: 0.1
 cli:
-  version: 1.22.10
+  version: 1.22.12
 # Trunk provides extensibility via plugins. (https://docs.trunk.io/plugins)
 plugins:
   sources:
@@ -20,16 +20,16 @@ lint:
     - bandit
     - git-diff-check
   enabled:
-    - osv-scanner@1.9.2
-    - yamllint@1.35.1
+    - osv-scanner@2.0.0
+    - yamllint@1.37.0
     - [email protected]
-    - [email protected].372
-    - [email protected].0
+    - [email protected].396
+    - [email protected].1
     - [email protected]
-    - [email protected].1
-    - ruff@0.9.7
+    - [email protected].3
+    - ruff@0.11.2
     - [email protected]
-    - [email protected].12
+    - [email protected].20
 actions:
   enabled:
     - trunk-upgrade-available

README.md

@@ -11,12 +11,13 @@ Simple python template I am experimenting with around a set of overlapping conce
 uv run fastapi dev main.py
 uv run pytest
 
+atlas schema inspect -u "sqlite://file?cache=shared&mode=memory" --format "{{ sql . }}"
 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"
+atlas schema apply --url "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --to "file://migrate.sql" --dev-url "docker://postgres/17" --schema "issue_analysis" --dry-run
 ```
 
 ```mermaid

app/core/factory.py

@@ -3,22 +3,19 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
-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 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 config import Settings
+from app.resource_adapters.persistence.sqlmodel.database import get_engine
 
 
 def create_app(settings: Settings, lifespan_handler=None) -> FastAPI:
     if lifespan_handler is None:
 
         @asynccontextmanager
         async def default_lifespan(app: FastAPI):
-
-            # Initialize database if using SQLModel
-            get_engine()
-
+            get_engine(settings)
             app.state.running = True
 
             yield
@@ -34,10 +31,10 @@ async def default_lifespan(app: FastAPI):
     )
 
     # Register global exception handler
-    app.add_exception_handler(AppException, app_exception_handler)
+    # app.add_exception_handler(AppException, app_exception_handler)
 
     # Register routes
-    app.include_router(api_router, prefix="/v1")
+    # app.include_router(api_router, prefix="/v1")
 
     # Configure CORS
     app.add_middleware(

app/domain/aggregate_root.py

@@ -18,7 +18,14 @@ class BaseEvent(Protocol):
     timestamp: datetime
 
 
-class AggregateRoot(SQLModel, abc.ABC):
+class AggregateRoot(SQLModel, abc.ABC, table=False):
+    """
+    Base class for all aggregate roots in the domain.
+
+    The metadata for this class is injected at runtime using FastAPI's dependency injection.
+    Before using this class for database operations, ensure that set_metadata has been called.
+    """
+
     version: int = 0
 
     def __init__(self, **data):

app/interface_adapters/__init__.py

@@ -3,4 +3,4 @@
 from app.interface_adapters.endpoints import issues
 
 api_router = APIRouter()
-api_router.include_router(issues.router, tags=["issues"])
+api_router.include_router(issues.router, tags=["issues"])

app/resource_adapters/persistence/sqlmodel/database.py

@@ -2,22 +2,49 @@
 
 from fastapi import Depends
 from loguru import logger
-from sqlalchemy import inspect, schema, text
 from sqlalchemy.engine import Engine
 from sqlalchemy.pool import StaticPool
-from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel import Session, SQLModel, create_engine, MetaData
+from sqlalchemy import text, schema
 
 from config import Settings, get_settings
 
 _engine: Engine | None = None
+_metadata: MetaData | None = None
+
+
+def get_metadata(
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> MetaData:
+    """Dependency that provides SQLModel metadata with the correct schema configuration."""
+    global _metadata
+
+    if _metadata is None:
+        _metadata = MetaData(schema=settings.get_table_schema)
+
+        # Optional: Add naming convention for constraints
+        # _metadata.naming_convention = {
+        #     "ix": "ix_%(column_0_label)s",
+        #     "uq": "uq_%(table_name)s_%(column_0_name)s",
+        #     "ck": "ck_%(table_name)s_%(constraint_name)s",
+        #     "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+        #     "pk": "pk_%(table_name)s"
+        # }
+
+    return _metadata
+
+
+MetadataDep = Annotated[MetaData, Depends(get_metadata)]
 
 
 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}}
+            execution_options={
+                "schema_translate_map": {None: settings.get_table_schema}
+            }
         )
         yield session
 
@@ -35,18 +62,16 @@ def get_engine(_settings: Settings | None = None) -> Engine:
     if _settings is None:
         _settings = get_settings()
 
-    database_url = _settings.database_url
-
     # Configure engine based on database type
     engine_args = {"echo": True}
 
     # Add SQLite-specific settings for in-memory database
-    if database_url == "sqlite://":
+    if _settings.database_url.startswith("sqlite"):
         engine_args.update(
             {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
         )
 
-    _engine = create_engine(database_url, **engine_args)
+    _engine = create_engine(_settings.database_url, **engine_args)
 
     # Enable WAL mode if configured
     if _settings.sqlite_wal_mode:
@@ -56,42 +81,29 @@ def get_engine(_settings: Settings | None = None) -> Engine:
             # conn.execute(text("PRAGMA synchronous=OFF"))
             logger.info("SQLite WAL mode enabled")
 
-    # Initialize database if using SQLModel
-    if (
-        _settings.database_type == "sqlmodel"
-        and _settings.database_schema
-        and not _settings.database_url.startswith("sqlite")
-    ):
-        SQLModel.metadata.schema = _settings.database_schema
+    # Initialize schema if using SQLModel, with a schema if not sqlite
+    if _settings.database_type == "sqlmodel":
+        SQLModel.metadata.schema = _settings.get_table_schema
         with _engine.connect() as conn:
             conn.execution_options = {
-                "schema_translate_map": {None: _settings.database_schema}
+                "schema_translate_map": {None: _settings.get_table_schema}
             }
-            if not conn.dialect.has_schema(conn, _settings.database_schema):
+            if not _settings.database_url.startswith(
+                "sqlite"
+            ) and not conn.dialect.has_schema(conn, _settings.get_table_schema):
                 logger.warning(
-                    f"Schema '{_settings.database_schema}' not found in database. Creating..."
+                    f"Schema '{_settings.get_table_schema}' not found in database. Creating..."
                 )
-                conn.execute(schema.CreateSchema(_settings.database_schema))
+                conn.execute(schema.CreateSchema(_settings.get_table_schema))
                 conn.commit()
 
-    if _settings.database_type == "sqlmodel" and not _settings.migrate_database:
-        logger.info("Creating database tables...")
-
-        # 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
-        )
-        if not existing_tables:
-            # if settings.database_schema:
-            #     # Create schema if it doesn't exist
-            #     if not inspector.has_schema(settings.database_schema):
-            #         logger.info(f"Creating schema '{settings.database_schema}'")
-            #         with _engine.begin() as conn:
-            #             conn.execute(text(f'CREATE SCHEMA IF NOT EXISTS {settings.database_schema}'))
-            SQLModel.metadata.create_all(_engine)
-            logger.info("Database tables created successfully")
-        else:
-            logger.info("Database tables already exist, skipping creation")
+            if not _settings.migrate_database:
+                SQLModel.metadata.create_all(conn)
+                conn.commit()
+                logger.info("Database tables created successfully")
+            else:
+                logger.info(
+                    "Database tables already exist or migration is configured, skipping creation"
+                )
 
     return _engine

app/tests/conftest.py

@@ -25,10 +25,19 @@
 
 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")
+    # Extract database path from the database URL
+    db_url = settings.database_url
+    if db_url.startswith("sqlite:///") and not db_url.endswith(":memory:"):
+        # Remove the sqlite:/// prefix to get the file path
+        db_path = db_url.replace("sqlite:///", "")
+        # Remove ./ prefix if present
+        if db_path.startswith("./"):
+            db_path = db_path[2:]
+
+        # Remove test database if it exists
+        if os.path.exists(db_path):
+            logger.info(f"Removing test database: {db_path}")
+            os.remove(db_path)
 
 
 @asynccontextmanager

app/tests/test_config.py

@@ -2,6 +2,4 @@
 
 
 def test_testing_database_url():
-    """Verify that the database URL is set to sqlite for testing"""
-    assert settings.database_url == "sqlite:///./test.db"
     assert settings.current_env == "testing"  # Verify we're in testing environment

config.py

@@ -1,7 +1,6 @@
-from functools import lru_cache
 from typing import List, Literal
 
-from pydantic import AnyHttpUrl, field_validator
+from pydantic import AnyHttpUrl
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -15,6 +14,13 @@ class Settings(BaseSettings):
     database_type: str = "sqlmodel"
     current_env: Literal["testing", "default"]
 
+    @property
+    def get_table_schema(self) -> str | None:
+        """Return table_schema if database_url does not start with sqlite, otherwise return None."""
+        if self.database_url.startswith("sqlite"):
+            return None
+        return self.database_schema
+
     # CORS settings
     backend_cors_origins: List[str] = ["http://localhost:8000", "http://localhost:3000"]
     cors_allow_credentials: bool = False
@@ -36,19 +42,19 @@ class Settings(BaseSettings):
         extra="ignore",
     )
 
-    @field_validator("database_schema")
+    """ @field_validator("database_schema")
     def validate_schema(cls, v: str | None, values) -> str | None:
         # Only set database_schema if database_url doesn't start with sqlite
         if values.data.get("database_url", "").startswith("sqlite"):
             return None
-        return v
+        return v """
 
     @property
     def backend_cors_origins_list(self) -> List[AnyHttpUrl]:
         return [AnyHttpUrl(origin) for origin in self.backend_cors_origins]
 
 
 # Singleton pattern to ensure only one settings instance
-@lru_cache()
+# @lru_cache()
 def get_settings() -> "Settings":
     return Settings()