Merge pull request #46 from getmarkus/codex_mob

async operations

Author: getmarkus

.env.example

@@ -5,19 +5,15 @@ APP_PROJECT_NAME=python-template
 # sqlite:///:memory: follows normal exception handling rules
 # APP_DATABASE_URL_TEMPLATE=sqlite:///:memory:
 # APP_DATABASE_URL_TEMPLATE=sqlite:///./issues.db
-APP_DATABASE_URL_TEMPLATE=postgresql+psycopg://app_user:change_this_password@localhost:5432/app_database
+# APP_DATABASE_URL_TEMPLATE=postgresql+psycopg://app_user:change_this_password@localhost:5432/app_database
+# Or, for an async driver using asyncpg:
+APP_DATABASE_URL_TEMPLATE=postgresql+asyncpg://app_user:change_this_password@localhost:5432/app_database
 APP_DATABASE_SCHEMA=issue_analysis
-APP_MIGRATE_DATABASE=false
+APP_CREATE_TABLES=true
 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
+# APP_CURRENT_ENV can be 'default' or 'testing'
+APP_CURRENT_ENV=default
 
 # CORS Settings
 APP_BACKEND_CORS_ORIGINS=["http://localhost:8000","http://localhost:3000"]

AGENTS.md

@@ -0,0 +1,99 @@
+# Design and Principles
+
+- Simple python template I am experimenting with around a set of overlapping concepts with a Fastapi implementation, primarily:
+  - Vertical Slice
+  - With elements of:
+    - Domain Driven Design (DDD)
+    - Clean Architecture
+    - Ports & Adapters
+    - CQRS
+
+## Pair Programming Rules
+
+- do not make git commits
+- pair program with the human developer with <https://github.com/remotemobprogramming/mob> framework
+- human developer will the only one who executes `mob done`
+- agent will execute `mob start` when coding starts
+- agent will execute `mob next` when coding is complete and ready for review
+
+## Core System Components
+
+- The main application logic is in app.
+- Feature flags and configuration settings are in config.py.
+
+## Code styles
+
+- Use PEP 8 and pythonic code
+- Use type hints
+- Functions: Use `snake_case`
+- Variables: Use `snake_case`
+- Constants: Use `UPPER_CASE`
+- Classes: Use `CamelCase`
+- Use `uv` for dependency management
+- Use `pyenv` for python version management
+- Use `loguru` for logging
+- Use `pydantic` for data validation
+- Use `sqlmodel` for database interactions
+- Use `pydantic-settings` for configuration
+- Use `python-dotenv` for environment variables
+- Use `pytest` for testing
+- Use `pytest-cov` for coverage
+
+## Data & Storage
+
+- Support for Sqlite and Postgres via sqlmodel.
+- Use `atlas` for database migrations
+- Use `migrations` directory for migration files
+- Use Bytebase and Supabase style guides
+- Identifier Style: snake_case, lowercase, descriptive
+- Table Names: snake_case, plural, no prefixes, descriptive
+- Column Names: snake_case, singular, descriptive, avoid generic, with `_id`
+- Data Types: Prefer specific types, ENUM for small sets, avoid CHAR(n)
+- Query Formatting: Lowercase keywords, whitespace, clear indentation
+- Functions: Lowercase, snake_case
+- Constraints: Explicit, descriptive names
+
+## Testing & Debugging
+
+- Unit, component and integration tests in app/tests
+- Create unit and component tests first before writing the actual code
+- Follow code implementation with integration tests
+
+## Workflow and developer interaction
+
+- TBD
+
+## Bash commands
+
+- Running api server:
+
+```bash
+uv run fastapi dev main.py
+```
+
+- Running tests:
+
+```bash
+uv run pytest
+```
+
+- Running database and migrations:
+
+```bash
+docker compose up
+docker compose down -v
+```
+
+- Various migration commands depending on database type:
+
+```bash
+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 inspect -u "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --schema "issue_analysis" --format "{{ 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://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --to "file://./migrations/migrate.sql" --dev-url "docker://postgres/17" --dry-run
+atlas schema apply --url "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --to "file://./migrations/migrate.sql" --dev-url "docker://postgres/17"
+```

README.md

@@ -11,6 +11,9 @@ Simple python template I am experimenting with around a set of overlapping conce
 uv run fastapi dev main.py
 uv run pytest
 
+docker compose up
+docker compose down -v
+
 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 inspect -u "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --schema "issue_analysis" --format "{{ sql . }}"
@@ -21,7 +24,3 @@ atlas schema apply --url "sqlite://issues.db" --to "file://migrate.sql" --dev-ur
 atlas schema apply --url "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --to "file://./migrations/migrate.sql" --dev-url "docker://postgres/17" --dry-run
 atlas schema apply --url "postgres://app_user:change_this_password@localhost:5432/app_database?sslmode=disable" --to "file://./migrations/migrate.sql" --dev-url "docker://postgres/17"
 ```
-
-```mermaid
-graph TD;
-```

app/core/database.py

@@ -1,16 +1,21 @@
-from typing import Annotated, Generator
+from typing import Annotated, Generator, AsyncGenerator
 
 from fastapi import Depends
 from loguru import logger
 from sqlalchemy.engine import Engine
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from sqlmodel.ext.asyncio.session import AsyncSession
+from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
 from sqlmodel import Session, SQLModel, create_engine, MetaData
 from sqlalchemy import text, schema
 
 from config import Settings, get_settings
 
 _engine: Engine | None = None
+_async_engine: AsyncEngine | None = None
 _metadata: MetaData | None = None
+_async_sessionmaker: sessionmaker[AsyncSession] | None = None
 
 
 def get_metadata(
@@ -62,48 +67,134 @@ def get_engine(_settings: Settings | None = None) -> Engine:
     if _settings is None:
         _settings = get_settings()
 
-    # Configure engine based on database type
-    engine_args = {"echo": True}
+    # Get the base URL template and credentials
+    url = _settings.database_url_template
 
-    # Add SQLite-specific settings for in-memory database
-    if _settings.database_url.startswith("sqlite"):
+    try:
+        from sqlalchemy.engine import make_url
+
+        # Just for validation purposes
+        make_url(url)
+    except ImportError:
+        logger.warning("Could not import sqlalchemy.engine.make_url")
+
+    engine_args: dict = {"echo": True}
+    if url.startswith("sqlite"):
         engine_args.update(
             {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
         )
 
-    _engine = create_engine(_settings.database_url, **engine_args)
+    _engine = create_engine(url, **engine_args)
 
-    # Enable WAL mode if configured
-    if _settings.sqlite_wal_mode:
+    if _settings.sqlite_wal_mode and url.startswith("sqlite"):
         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 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.get_table_schema}
-            } """
-            if not _settings.database_url.startswith(
-                "sqlite"
-            ) and not conn.dialect.has_schema(conn, _settings.get_table_schema):
-                logger.warning(
-                    f"Schema '{_settings.get_table_schema}' not found in database. Creating..."
-                )
-                conn.execute(schema.CreateSchema(_settings.get_table_schema))
-                conn.commit()
-
-            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"
-                )
+    # Set schema for SQLModel metadata
+    SQLModel.metadata.schema = _settings.get_table_schema
+    with _engine.connect() as conn:
+        if not url.startswith("sqlite") and not conn.dialect.has_schema(
+            conn,
+            _settings.get_table_schema,  # type: ignore[arg-type]
+        ):
+            logger.warning(
+                f"Schema '{_settings.get_table_schema}' not found in database. Creating..."
+            )
+            conn.execute(schema.CreateSchema(_settings.get_table_schema))
+            conn.commit()
+
+        if _settings.create_tables:
+            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
+
+
+def get_async_engine(_settings: Settings | None = None) -> AsyncEngine:
+    """Get or create async SQLModel engine instance."""
+    global _async_engine, _async_sessionmaker
+
+    if _async_engine is not None:
+        return _async_engine
+
+    if _settings is None:
+        _settings = get_settings()
+
+    url = _settings.database_url
+    engine_args: dict = {"echo": True}
+    
+    # SQLite: use the aiosqlite driver for async support
+    if url.startswith("sqlite") and "+aiosqlite" not in url:
+        url = url.replace("sqlite://", "sqlite+aiosqlite://", 1)
+        engine_args.update(
+            {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
+        )
+    else:
+        # PostgreSQL: ensure an asyncpg driver if scheme is postgres/postgresql without a +driver suffix
+        try:
+            scheme, rest = url.split("://", 1)
+        except ValueError:
+            scheme = url
+            rest = ""
+            
+        # Handle SSL mode for asyncpg
+        connect_args = {}
+        if "?" in rest:
+            base_url, query_string = rest.split("?", 1)
+            params = {}
+            for param in query_string.split("&"):
+                if "=" in param:
+                    key, value = param.split("=", 1)
+                    params[key] = value
+            
+            # Remove sslmode from URL and add as connect_args for asyncpg
+            if "sslmode" in params:
+                sslmode = params.pop("sslmode")
+                if sslmode == "disable":
+                    connect_args["ssl"] = False
+                elif sslmode in ("require", "verify-ca", "verify-full"):
+                    connect_args["ssl"] = True
+                
+                # Rebuild the URL without sslmode
+                new_query = "&".join([f"{k}={v}" for k, v in params.items()])
+                if new_query:
+                    rest = f"{base_url}?{new_query}"
+                else:
+                    rest = base_url
+        
+        if connect_args:
+            engine_args["connect_args"] = connect_args
+            
+        if scheme in ("postgres", "postgresql") and "+" not in scheme:
+            url = f"postgresql+asyncpg://{rest}"
+
+    engine_args: dict = {"echo": True}
+    if url.startswith("sqlite"):
+        engine_args.update(
+            {"connect_args": {"check_same_thread": False}, "poolclass": StaticPool}
+        )
+
+    _async_engine = create_async_engine(url, **engine_args)
+    _async_sessionmaker = sessionmaker(
+        _async_engine, class_=AsyncSession, expire_on_commit=False
+    )
+    return _async_engine
+
+
+async def get_async_session(
+    settings: Annotated[Settings, Depends(get_settings)],
+) -> AsyncGenerator[AsyncSession, None]:
+    # Initialize the async engine if it doesn't exist yet
+    get_async_engine(settings)
+    assert _async_sessionmaker is not None
+    async with _async_sessionmaker() as session:
+        yield session
+
+
+AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]

app/core/factory.py

@@ -7,7 +7,7 @@
 from app.core.exceptions import AppException
 from app.core.middleware import app_exception_handler
 from config import Settings
-from app.core.database import get_engine
+from app.core.database import get_async_engine, get_engine
 
 # Create API router and include feature routers
 api_router = APIRouter()
@@ -19,7 +19,18 @@ def create_app(settings: Settings, lifespan_handler=None) -> FastAPI:
 
         @asynccontextmanager
         async def default_lifespan(app: FastAPI):
-            get_engine(settings)
+            # Choose database engine based on configuration
+            if settings.is_async_database:
+                # Use async engine for async database URLs
+                get_async_engine(settings)
+            else:
+                # For sync database URLs, we need to initialize the engine outside the async context
+                # This is done here for simplicity, but in production you might want to
+                # use a background task or separate initialization step
+                import asyncio
+                loop = asyncio.get_event_loop()
+                await loop.run_in_executor(None, lambda: get_engine(settings))
+                
             app.state.running = True
 
             yield