How to Build a Custom Database Adapter using the Hybrid MX Pattern

This guide walks you through building a custom database adapter (such as framework-mx-snowflake) by inheriting from the standard SQLAlchemy adapters in framework-m-standard.

The Hybrid MX Pattern maximizes code reuse by inheriting standard CRUD, repository, and transaction management logic, only overriding the dialect-specific parts (like data type mappings, sequence generation, and sync engine DDL statements).

---

1. Package Structure

To create a hybrid adapter, structure your package as follows:

framework-mx-snowflake/
├── pyproject.toml
└── src/
    └── framework_mx_snowflake/
        ├── __init__.py
        ├── repository.py
        ├── schema_mapper.py
        └── sync_engine.py

---

2. Inheriting and Overriding the Schema Mapper

Different databases have specific column constraints and type representations. To customize these, subclass the standard SchemaMapper and override _get_column_type:

# src/framework_mx_snowflake/schema_mapper.py
from typing import Any
from pydantic.fields import FieldInfo
from sqlalchemy import Numeric
from sqlalchemy.types import TypeEngine
from framework_m_standard.adapters.db.schema_mapper import SchemaMapper


class SnowflakeSchemaMapper(SchemaMapper):
    """Custom SchemaMapper that resolves Snowflake-specific database types."""

    def _get_column_type(
        self,
        _field_name: str,
        python_type: type,
        field_info: FieldInfo,
    ) -> type[TypeEngine[Any]] | TypeEngine[Any]:
        # Customize type mapping if needed (e.g. mapping bool to Numeric(1) for certain dialects)
        if python_type is bool:
            return Numeric(precision=1)

        # Fallback to standard SQLAlchemy mappings for other types
        return super()._get_column_type(_field_name, python_type, field_info)

---

3. Customizing Naming Sequences

To retrieve values from a custom sequence (which might use a custom dialect syntax), subclass the DocumentNamingService and override the sequence retrieval hook:

# src/framework_mx_snowflake/naming.py
from typing import Any
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession
from framework_m_standard.adapters.db.naming import DocumentNamingService


class SnowflakeNamingService(DocumentNamingService[Any]):
    """Custom document naming service utilizing Snowflake sequences."""

    async def _fetch_sequence_value(
        self, session: AsyncSession, sequence_name: str
    ) -> int:
        # Override standard nextval() logic with Snowflake-specific syntax
        stmt = select(text(f"SELECT {sequence_name}.NEXTVAL"))
        result = await session.execute(stmt)
        val = result.scalar()
        if val is None:
            raise ValueError(f"Sequence {sequence_name} returned None")
        return int(val)

---

4. Customizing Schema Migration DDL

The DeclarativeSyncEngine handles runtime database DDL synchronization. Subclass it to configure custom DDL locks, schemas, and table alteration commands for your database:

# src/framework_mx_snowflake/sync_engine.py
from typing import Any
from sqlalchemy.engine import Engine
from framework_m_standard.adapters.db.declarative_sync_engine import (
    DeclarativeSyncEngine,
)


class SnowflakeDeclarativeSyncEngine(DeclarativeSyncEngine):
    """Synchronizer that executes Snowflake-compatible DDL modifications."""

    def _apply_lock_timeout(
        self, conn: Any, dialect_name: str, max_lock_time: int | None
    ) -> None:
        # Override to set custom lock timeouts if needed
        pass

    def _create_schema(
        self, conn: Any, dialect_name: str, schema_name: str | None
    ) -> None:
        # Custom schema creation DDL
        if schema_name:
            conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_name}"))

    def _get_add_column_sql(
        self,
        engine: Engine,
        table_name: str,
        column_name: str,
        col_type: str,
        nullable: str,
        default: str,
    ) -> str:
        # Generate custom ALTER TABLE ADD syntax
        return (
            f"ALTER TABLE {table_name} "
            f"ADD COLUMN {column_name} {col_type} {nullable}{default}"
        )

5. Registering the Custom Adapter

To hook your custom classes into the framework's dynamic database resolver, register them in your package's pyproject.toml project entry points using the dialect name (e.g. snowflake) as the entry point key:

# pyproject.toml
[project.entry-points."framework_m.adapters.schema_mapper"]
snowflake = "framework_mx_snowflake.schema_mapper:SnowflakeSchemaMapper"

[project.entry-points."framework_m.adapters.naming_service"]
snowflake = "framework_mx_snowflake.naming:SnowflakeNamingService"

[project.entry-points."framework_m.adapters.declarative_sync_engine"]
snowflake = "framework_mx_snowflake.sync_engine:SnowflakeDeclarativeSyncEngine"

When your package (framework-mx-snowflake) is installed in the target environment, the framework automatically detects these entry points and binds your custom classes dynamically at runtime whenever it connects to a database using the snowflake dialect, ensuring compatibility even in dynamic multi-db tenant architectures.