SchemaMapperProtocol

Protocol defining the interface for schema mapping.

SchemaMappers convert Pydantic DocType models to storage-specific schemas.
This allows MX packages to provide adapters for different databases.

Methods:
    create_table: Map single DocType to storage schema
    create_tables: Map DocType with child tables to multiple schemas
    detect_schema_changes: Detect differences between two model versions

Example - Standard SQLAlchemy implementation:
    class SchemaMapper:
        def create_table(self, model: type[BaseDocType], metadata: MetaData) -> Table:
            # Returns SQLAlchemy Table
            return Table(...)

        def create_tables(self, model: type[BaseDocType], metadata: MetaData) -> list[Table]:
            # Returns parent + child tables
            return [parent_table, child_table_1, ...]

        def detect_schema_changes(
            self, old_model: type[BaseDocType], new_model: type[BaseDocType]
        ) -> dict[str, Any]:
            # Returns diff
            return {"added_fields": [...], "removed_fields": [...]}

Example - MX MongoDB implementation:
    class MongoSchemaMapper:
        def create_table(self, model: type[BaseDocType], metadata: Any) -> dict[str, Any]:
            # Returns MongoDB collection schema
            return {
                "collection": model.__name__.lower(),
                "validator": {...},  # JSON Schema validator
                "indexes": [...]
            }

        def create_tables(self, model: type[BaseDocType], metadata: Any) -> list[dict[str, Any]]:
            # MongoDB: parent and embedded docs in same collection
            return [self.create_table(model, metadata)]

        def detect_schema_changes(
            self, old_model: type[BaseDocType], new_model: type[BaseDocType]
        ) -> dict[str, Any]:
            # Returns MongoDB-specific changes
            return {
                "added_fields": [...],
                "removed_fields": [...],
                "index_changes": [...]
            }

Source: schema_mapper.py

Methods

create_table

def create_table(self, model: type[BaseDocTypeProtocol], metadata: Any) -> Any

Create storage schema for a single DocType.

    Args:
        model: The DocType class to map
        metadata: Storage-specific metadata object
                 (SQLAlchemy MetaData, MongoDB client, etc.)

    Returns:
        Storage-specific schema representation
        (SQLAlchemy Table, MongoDB schema dict, etc.)

    Example - SQLAlchemy:
        mapper = SchemaMapper()
        metadata = MetaData()
        table = mapper.create_table(Todo, metadata)
        # Returns: Table('todo', metadata, Column('id', UUID), ...)

    Example - MongoDB:
        mapper = MongoSchemaMapper()
        client = MongoClient()
        schema = mapper.create_table(Todo, client)
        # Returns: {"collection": "todo", "validator": {...}}

create_tables

def create_tables(self, model: type[BaseDocTypeProtocol], metadata: Any
) -> list[Any]

Create storage schemas for DocType with child tables.

    For DocTypes with child table fields (list[ChildDocType]), this
    creates schemas for both parent and children.

    Args:
        model: The DocType class (may have child table fields)
        metadata: Storage-specific metadata object

    Returns:
        List of storage schemas (parent first, then children)

    Example - SQLAlchemy (with child tables):
        class Invoice(DocType):
            items: list[InvoiceItem]  # Child table

        mapper = SchemaMapper()
        tables = mapper.create_tables(Invoice, metadata)
        # Returns: [Table('invoice', ...), Table('invoiceitem', ...)]

    Example - MongoDB (embedded documents):
        # MongoDB stores children as embedded arrays
        mapper = MongoSchemaMapper()
        schemas = mapper.create_tables(Invoice, client)
        # Returns: [{"collection": "invoice", "items": [embedded]}]

detect_schema_changes

def detect_schema_changes(self,
    old_model: type[BaseDocTypeProtocol],
    new_model: type[BaseDocTypeProtocol],
) -> dict[str, Any]

Detect schema changes between two model versions.

    Used for migrations: compares old and new model definitions
    to generate ALTER statements or migration scripts.

    Args:
        old_model: The previous version of the DocType
        new_model: The new version of the DocType

    Returns:
        Dictionary describing changes (structure is implementation-specific)

    Example - SQLAlchemy:
        changes = mapper.detect_schema_changes(OldUser, NewUser)
        # Returns: {
        #     "added_fields": ["department", "employee_id"],
        #     "removed_fields": ["old_field"],
        #     "type_changes": [("age", "String", "Integer")]
        # }

    Example - MongoDB:
        changes = mapper.detect_schema_changes(OldUser, NewUser)
        # Returns: {
        #     "added_fields": ["department"],
        #     "index_changes": [{"add": "department_idx"}]
        # }