SQLModel API Reference¶

metaxy.ext.sqlmodel ¶

Classes¶

metaxy.ext.sqlmodel.SQLModelPluginConfig ¶

Bases: PluginConfig

Configuration for SQLModel integration.

This plugin enhances SQLModel-based features with automatic table name inference and optional primary key injection.

metaxy.ext.sqlmodel.BaseSQLModelFeature `pydantic-model` ¶

Bases: SQLModel, BaseFeature

Base class for Metaxy features that are also SQLModel tables.

Example

from metaxy.integrations.sqlmodel import BaseSQLModelFeature
from metaxy import FeatureSpec, FeatureKey, FieldSpec, FieldKey
from sqlmodel import Field

class VideoFeature(
    BaseSQLModelFeature,
    table=True,
    spec=FeatureSpec(
        key=FeatureKey(["video"]),
        id_columns=["uid"],
        fields=[
            FieldSpec(
                key=FieldKey(["video_file"]),
                code_version="1",
            ),
        ],
    ),
):

    uid: str = Field(primary_key=True)
    path: str
    duration: float

    # Now you can use both Metaxy and SQLModel features:
    # - VideoFeature.feature_version() -> Metaxy versioning
    # - session.exec(select(VideoFeature)) -> SQLModel queries

Show JSON schema:

{
  "description": "Base class for `Metaxy` features that are also `SQLModel` tables.\n\n!!! example\n\n    ```py\n    from metaxy.integrations.sqlmodel import BaseSQLModelFeature\n    from metaxy import FeatureSpec, FeatureKey, FieldSpec, FieldKey\n    from sqlmodel import Field\n\n    class VideoFeature(\n        BaseSQLModelFeature,\n        table=True,\n        spec=FeatureSpec(\n            key=FeatureKey([\"video\"]),\n            id_columns=[\"uid\"],\n            fields=[\n                FieldSpec(\n                    key=FieldKey([\"video_file\"]),\n                    code_version=\"1\",\n                ),\n            ],\n        ),\n    ):\n\n        uid: str = Field(primary_key=True)\n        path: str\n        duration: float\n\n        # Now you can use both Metaxy and SQLModel features:\n        # - VideoFeature.feature_version() -> Metaxy versioning\n        # - session.exec(select(VideoFeature)) -> SQLModel queries\n    ```",
  "properties": {
    "metaxy_provenance_by_field": {
      "additionalProperties": {
        "type": "string"
      },
      "default": null,
      "description": "Field-level provenance hashes (maps field names to hashes)",
      "title": "Metaxy Provenance By Field",
      "type": "object"
    },
    "metaxy_provenance": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hash of metaxy_provenance_by_field",
      "title": "Metaxy Provenance"
    },
    "metaxy_feature_version": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hash of the feature definition (dependencies + fields + code_versions)",
      "title": "Metaxy Feature Version"
    },
    "metaxy_snapshot_version": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hash of the entire feature graph snapshot",
      "title": "Metaxy Snapshot Version"
    },
    "metaxy_data_version_by_field": {
      "anyOf": [
        {
          "additionalProperties": {
            "type": "string"
          },
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Field-level data version hashes (maps field names to version hashes)",
      "title": "Metaxy Data Version By Field"
    },
    "metaxy_data_version": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hash of metaxy_data_version_by_field",
      "title": "Metaxy Data Version"
    },
    "metaxy_created_at": {
      "anyOf": [
        {
          "format": "date-time",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Timestamp when the metadata row was created (UTC)",
      "title": "Metaxy Created At"
    },
    "metaxy_materialization_id": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "External orchestration run ID (e.g., Dagster Run ID)",
      "title": "Metaxy Materialization Id"
    },
    "metaxy_feature_spec_version": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Hash of the complete feature specification.",
      "title": "Metaxy Feature Spec Version"
    }
  },
  "title": "BaseSQLModelFeature",
  "type": "object"
}

Config:

default: {'frozen': False}

Fields:

metaxy_provenance (str | None)
metaxy_provenance_by_field (dict[str, str])
metaxy_feature_version (str | None)
metaxy_feature_spec_version (str | None)
metaxy_snapshot_version (str | None)
metaxy_data_version (str | None)
metaxy_data_version_by_field (dict[str, str] | None)
metaxy_created_at (AwareDatetime | None)
metaxy_materialization_id (str | None)

Validators:

_validate_id_columns_exist

Attributes¶

metaxy_provenance `pydantic-field` ¶

metaxy_provenance: str | None = None

Hash of metaxy_provenance_by_field

metaxy_provenance_by_field `pydantic-field` ¶

metaxy_provenance_by_field: dict[str, str] = None

Field-level provenance hashes (maps field names to hashes)

metaxy_feature_version `pydantic-field` ¶

metaxy_feature_version: str | None = None

Hash of the feature definition (dependencies + fields + code_versions)

metaxy_feature_spec_version `pydantic-field` ¶

metaxy_feature_spec_version: str | None = None

Hash of the complete feature specification.

metaxy_snapshot_version `pydantic-field` ¶

metaxy_snapshot_version: str | None = None

Hash of the entire feature graph snapshot

metaxy_data_version `pydantic-field` ¶

metaxy_data_version: str | None = None

Hash of metaxy_data_version_by_field

metaxy_data_version_by_field `pydantic-field` ¶

metaxy_data_version_by_field: dict[str, str] | None = None

Field-level data version hashes (maps field names to version hashes)

metaxy_created_at `pydantic-field` ¶

metaxy_created_at: AwareDatetime | None = None

Timestamp when the metadata row was created (UTC)

metaxy_materialization_id `pydantic-field` ¶

metaxy_materialization_id: str | None = None

External orchestration run ID (e.g., Dagster Run ID)

Functions¶

table_name `classmethod` ¶

table_name() -> str

Get SQL-like table name for this feature.

Converts feature key to SQL-compatible table name by joining parts with double underscores, consistent with IbisMetadataStore.

Returns:

str –

Table name string (e.g., "my_namespace__my_feature")

Example

class VideoFeature(Feature, spec=FeatureSpec(
    key=FeatureKey(["video", "processing"]),
    ...
)):
    pass
VideoFeature.table_name()
# 'video__processing'

Source code in src/metaxy/models/feature.py

@classmethod
def table_name(cls) -> str:
    """Get SQL-like table name for this feature.

    Converts feature key to SQL-compatible table name by joining
    parts with double underscores, consistent with IbisMetadataStore.

    Returns:
        Table name string (e.g., "my_namespace__my_feature")

    Example:
        ```py
        class VideoFeature(Feature, spec=FeatureSpec(
            key=FeatureKey(["video", "processing"]),
            ...
        )):
            pass
        VideoFeature.table_name()
        # 'video__processing'
        ```
    """
    return cls.spec().table_name()

feature_version `classmethod` ¶

feature_version() -> str

Get hash of feature specification.

Returns a hash representing the feature's complete configuration: - Feature key - Field definitions and code versions - Dependencies (feature-level and field-level)

This hash changes when you modify: - Field code versions - Dependencies - Field definitions

Used to distinguish current vs historical metafield provenance hashes. Stored in the 'metaxy_feature_version' column of metadata DataFrames.

Returns:

str –

SHA256 hex digest (like git short hashes)

Example

class MyFeature(Feature, spec=FeatureSpec(
    key=FeatureKey(["my", "feature"]),
    fields=[FieldSpec(key=FieldKey(["default"]), code_version="1")],
)):
    pass
MyFeature.feature_version()
# 'a3f8b2c1...'

Source code in src/metaxy/models/feature.py

@classmethod
def feature_version(cls) -> str:
    """Get hash of feature specification.

    Returns a hash representing the feature's complete configuration:
    - Feature key
    - Field definitions and code versions
    - Dependencies (feature-level and field-level)

    This hash changes when you modify:
    - Field code versions
    - Dependencies
    - Field definitions

    Used to distinguish current vs historical metafield provenance hashes.
    Stored in the 'metaxy_feature_version' column of metadata DataFrames.

    Returns:
        SHA256 hex digest (like git short hashes)

    Example:
        ```py
        class MyFeature(Feature, spec=FeatureSpec(
            key=FeatureKey(["my", "feature"]),
            fields=[FieldSpec(key=FieldKey(["default"]), code_version="1")],
        )):
            pass
        MyFeature.feature_version()
        # 'a3f8b2c1...'
        ```
    """
    return cls.graph.get_feature_version(cls.spec().key)

feature_spec_version `classmethod` ¶

feature_spec_version() -> str

Get hash of the complete feature specification.

Returns a hash representing ALL specification properties including: - Feature key - Dependencies - Fields - Code versions - Any future metadata, tags, or other properties

Unlike feature_version which only hashes computational properties (for migration triggering), feature_spec_version captures the entire specification for complete reproducibility and audit purposes.

Stored in the 'metaxy_feature_spec_version' column of metadata DataFrames.

Returns:

str –

SHA256 hex digest of the complete specification

Example

class MyFeature(Feature, spec=FeatureSpec(
    key=FeatureKey(["my", "feature"]),
    fields=[FieldSpec(key=FieldKey(["default"]), code_version="1")],
)):
    pass
MyFeature.feature_spec_version()
# 'def456...'  # Different from feature_version

Source code in src/metaxy/models/feature.py

@classmethod
def feature_spec_version(cls) -> str:
    """Get hash of the complete feature specification.

    Returns a hash representing ALL specification properties including:
    - Feature key
    - Dependencies
    - Fields
    - Code versions
    - Any future metadata, tags, or other properties

    Unlike feature_version which only hashes computational properties
    (for migration triggering), feature_spec_version captures the entire specification
    for complete reproducibility and audit purposes.

    Stored in the 'metaxy_feature_spec_version' column of metadata DataFrames.

    Returns:
        SHA256 hex digest of the complete specification

    Example:
        ```py
        class MyFeature(Feature, spec=FeatureSpec(
            key=FeatureKey(["my", "feature"]),
            fields=[FieldSpec(key=FieldKey(["default"]), code_version="1")],
        )):
            pass
        MyFeature.feature_spec_version()
        # 'def456...'  # Different from feature_version
        ```
    """
    return cls.spec().feature_spec_version

full_definition_version `classmethod` ¶

full_definition_version() -> str

Get hash of the complete feature definition including Pydantic schema.

This method computes a hash of the entire feature class definition, including: - Pydantic model schema - Project name

Used in the metaxy_full_definition_version column of system tables.

Returns:

str –

SHA256 hex digest of the complete definition

Source code in src/metaxy/models/feature.py

@classmethod
def full_definition_version(cls) -> str:
    """Get hash of the complete feature definition including Pydantic schema.

    This method computes a hash of the entire feature class definition, including:
    - Pydantic model schema
    - Project name

    Used in the `metaxy_full_definition_version` column of system tables.

    Returns:
        SHA256 hex digest of the complete definition
    """
    import json

    hasher = hashlib.sha256()

    # Hash the Pydantic schema (includes field types, descriptions, validators, etc.)
    schema = cls.model_json_schema()
    schema_json = json.dumps(schema, sort_keys=True)
    hasher.update(schema_json.encode())

    # Hash the feature specification
    hasher.update(cls.feature_spec_version().encode())

    # Hash the project name
    hasher.update(cls.project.encode())

    return truncate_hash(hasher.hexdigest())

provenance_by_field `classmethod` ¶

provenance_by_field() -> dict[str, str]

Get the code-level field provenance for this feature.

This returns a static hash based on code versions and dependencies, not sample-level field provenance computed from upstream data.

Returns:

dict[str, str] –

Dictionary mapping field keys to their provenance hashes.

Source code in src/metaxy/models/feature.py

@classmethod
def provenance_by_field(cls) -> dict[str, str]:
    """Get the code-level field provenance for this feature.

    This returns a static hash based on code versions and dependencies,
    not sample-level field provenance computed from upstream data.

    Returns:
        Dictionary mapping field keys to their provenance hashes.
    """
    return cls.graph.get_feature_version_by_field(cls.spec().key)

load_input `classmethod` ¶

load_input(joiner: Any, upstream_refs: dict[str, LazyFrame[Any]]) -> tuple[LazyFrame[Any], dict[str, str]]

Join upstream feature metadata.

Override for custom join logic (1:many, different keys, filtering, etc.).

Parameters:

joiner (Any) –

UpstreamJoiner from MetadataStore
upstream_refs (dict[str, LazyFrame[Any]]) –

Upstream feature metadata references (lazy where possible)

Returns:

LazyFrame[Any] –

(joined_upstream, upstream_column_mapping)
dict[str, str] –
- joined_upstream: All upstream data joined together
tuple[LazyFrame[Any], dict[str, str]] –
- upstream_column_mapping: Maps upstream_key -> column name

Source code in src/metaxy/models/feature.py

@classmethod
def load_input(
    cls,
    joiner: Any,
    upstream_refs: dict[str, "nw.LazyFrame[Any]"],
) -> tuple["nw.LazyFrame[Any]", dict[str, str]]:
    """Join upstream feature metadata.

    Override for custom join logic (1:many, different keys, filtering, etc.).

    Args:
        joiner: UpstreamJoiner from MetadataStore
        upstream_refs: Upstream feature metadata references (lazy where possible)

    Returns:
        (joined_upstream, upstream_column_mapping)
        - joined_upstream: All upstream data joined together
        - upstream_column_mapping: Maps upstream_key -> column name
    """
    from metaxy.models.feature_spec import FeatureDep

    # Extract columns and renames from deps
    upstream_columns: dict[str, tuple[str, ...] | None] = {}
    upstream_renames: dict[str, dict[str, str] | None] = {}

    deps = cls.spec().deps
    if deps:
        for dep in deps:
            if isinstance(dep, FeatureDep):
                dep_key_str = dep.feature.to_string()
                upstream_columns[dep_key_str] = dep.columns
                upstream_renames[dep_key_str] = dep.rename

    return joiner.join_upstream(
        upstream_refs=upstream_refs,
        feature_spec=cls.spec(),
        feature_plan=cls.graph.get_feature_plan(cls.spec().key),
        upstream_columns=upstream_columns,
        upstream_renames=upstream_renames,
    )

resolve_data_version_diff `classmethod` ¶

resolve_data_version_diff(diff_resolver: Any, target_provenance: LazyFrame[Any], current_metadata: LazyFrame[Any] | None, *, lazy: bool = False) -> Increment | LazyIncrement

Resolve differences between target and current field provenance.

Override for custom diff logic (ignore certain fields, custom rules, etc.).

Parameters:

diff_resolver (Any) –

MetadataDiffResolver from MetadataStore
target_provenance (LazyFrame[Any]) –

Calculated target field provenance (Narwhals LazyFrame)
current_metadata (LazyFrame[Any] | None) –

Current metadata for this feature (Narwhals LazyFrame, or None). Should be pre-filtered by feature_version at the store level.
lazy (bool, default: False ) –

If True, return LazyIncrement. If False, return Increment.

Returns:

Increment | LazyIncrement –

Increment (eager) or LazyIncrement (lazy) with added, changed, removed

Example (default):

class MyFeature(Feature, spec=...):
    pass  # Uses diff resolver's default implementation

Example (ignore certain field changes):

class MyFeature(Feature, spec=...):
    @classmethod
    def resolve_data_version_diff(cls, diff_resolver, target_provenance, current_metadata, **kwargs):
        # Get standard diff
        result = diff_resolver.find_changes(target_provenance, current_metadata, cls.spec().id_columns)

        # Custom: Only consider 'frames' field changes, ignore 'audio'
        # Users can filter/modify the increment here

        return result  # Return modified Increment

Source code in src/metaxy/models/feature.py

@classmethod
def resolve_data_version_diff(
    cls,
    diff_resolver: Any,
    target_provenance: "nw.LazyFrame[Any]",
    current_metadata: "nw.LazyFrame[Any] | None",
    *,
    lazy: bool = False,
) -> "Increment | LazyIncrement":
    """Resolve differences between target and current field provenance.

    Override for custom diff logic (ignore certain fields, custom rules, etc.).

    Args:
        diff_resolver: MetadataDiffResolver from MetadataStore
        target_provenance: Calculated target field provenance (Narwhals LazyFrame)
        current_metadata: Current metadata for this feature (Narwhals LazyFrame, or None).
            Should be pre-filtered by feature_version at the store level.
        lazy: If True, return LazyIncrement. If False, return Increment.

    Returns:
        Increment (eager) or LazyIncrement (lazy) with added, changed, removed

    Example (default):
        ```py
        class MyFeature(Feature, spec=...):
            pass  # Uses diff resolver's default implementation
        ```

    Example (ignore certain field changes):
        ```py
        class MyFeature(Feature, spec=...):
            @classmethod
            def resolve_data_version_diff(cls, diff_resolver, target_provenance, current_metadata, **kwargs):
                # Get standard diff
                result = diff_resolver.find_changes(target_provenance, current_metadata, cls.spec().id_columns)

                # Custom: Only consider 'frames' field changes, ignore 'audio'
                # Users can filter/modify the increment here

                return result  # Return modified Increment
        ```
    """
    # Diff resolver always returns LazyIncrement - materialize if needed
    lazy_result = diff_resolver.find_changes(
        target_provenance=target_provenance,
        current_metadata=current_metadata,
        id_columns=cls.spec().id_columns,  # Pass ID columns from feature spec
    )

    # Materialize to Increment if lazy=False
    if not lazy:
        from metaxy.versioning.types import Increment

        return Increment(
            added=lazy_result.added.collect(),
            changed=lazy_result.changed.collect(),
            removed=lazy_result.removed.collect(),
        )

    return lazy_result

metaxy.ext.sqlmodel.SQLModelFeatureMeta ¶

Bases: MetaxyMeta, SQLModelMetaclass

Functions¶

new ¶

__new__(cls_name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], *, spec: FeatureSpecWithIDColumns | None = None, inject_primary_key: bool | None = None, inject_index: bool | None = None, **kwargs: Any) -> type[Any]

Create a new SQLModel + Metaxy Feature class.

Parameters:

cls_name (str) –

Name of the class being created
bases (tuple[type[Any], ...]) –

Base classes
namespace (dict[str, Any]) –

Class namespace (attributes and methods)
spec (FeatureSpecWithIDColumns | None, default: None ) –

Metaxy FeatureSpec (required for concrete features)
inject_primary_key (bool | None, default: None ) –

If True, automatically create composite primary key including id_columns + (metaxy_created_at, metaxy_data_version).
inject_index (bool | None, default: None ) –

If True, automatically create composite index including id_columns + (metaxy_created_at, metaxy_data_version).
**kwargs (Any, default: {} ) –

Additional keyword arguments (e.g., table=True for SQLModel)

Returns:

type[Any] –

New class that is both a SQLModel table and a Metaxy feature

Source code in src/metaxy/ext/sqlmodel/plugin.py

def __new__(
    cls,
    cls_name: str,
    bases: tuple[type[Any], ...],
    namespace: dict[str, Any],
    *,
    spec: FeatureSpecWithIDColumns | None = None,
    inject_primary_key: bool | None = None,
    inject_index: bool | None = None,
    **kwargs: Any,
) -> type[Any]:
    """Create a new SQLModel + Metaxy Feature class.

    Args:
        cls_name: Name of the class being created
        bases: Base classes
        namespace: Class namespace (attributes and methods)
        spec: Metaxy FeatureSpec (required for concrete features)
        inject_primary_key: If True, automatically create composite primary key
            including id_columns + (metaxy_created_at, metaxy_data_version).
        inject_index: If True, automatically create composite index
            including id_columns + (metaxy_created_at, metaxy_data_version).
        **kwargs: Additional keyword arguments (e.g., table=True for SQLModel)

    Returns:
        New class that is both a SQLModel table and a Metaxy feature
    """
    # Override frozen config for SQLModel - instances need to be mutable for ORM
    if "model_config" not in namespace:
        from pydantic import ConfigDict

        namespace["model_config"] = ConfigDict(frozen=False)

    # Check plugin config for defaults
    config = MetaxyConfig.get()
    sqlmodel_config = config.get_plugin("sqlmodel", SQLModelPluginConfig)
    if inject_primary_key is None:
        inject_primary_key = sqlmodel_config.inject_primary_key
    if inject_index is None:
        inject_index = sqlmodel_config.inject_index

    # If this is a concrete table (table=True) with a spec
    if kwargs.get("table") and spec is not None:
        # Forbid custom __tablename__ since it won't work with metadata store's get_table_name()
        if "__tablename__" in namespace:
            raise ValueError(
                f"Cannot define custom __tablename__ in {cls_name}. "
                "The table name is automatically derived from the feature key. "
                "If you need a different table name, adjust the feature key instead."
            )

        # Prevent user-defined fields from shadowing system-managed columns
        conflicts = {
            attr_name
            for attr_name in namespace
            if attr_name in RESERVED_SQLMODEL_FIELD_NAMES
        }

        # Also guard against explicit sa_column_kwargs targeting system columns
        for attr_name, attr_value in namespace.items():
            sa_column_kwargs = getattr(attr_value, "sa_column_kwargs", None)
            if isinstance(sa_column_kwargs, dict):
                column_name = sa_column_kwargs.get("name")
                if column_name in ALL_SYSTEM_COLUMNS:
                    conflicts.add(attr_name)

        if conflicts:
            reserved = ", ".join(sorted(ALL_SYSTEM_COLUMNS))
            conflict_list = ", ".join(sorted(conflicts))
            raise ValueError(
                "Cannot define SQLModel field(s) "
                f"{conflict_list} because they map to reserved Metaxy system columns. "
                f"Reserved columns: {reserved}"
            )

        # Automatically set __tablename__ from the feature key
        namespace["__tablename__"] = spec.key.table_name

        # Inject table args (info metadata + optional constraints)
        cls._inject_table_args(
            namespace, spec, cls_name, inject_primary_key, inject_index
        )

    # Call super().__new__ which follows MRO: MetaxyMeta -> SQLModelMetaclass -> ...
    # MetaxyMeta will consume the spec parameter and pass remaining kwargs to SQLModelMetaclass
    new_class = super().__new__(
        cls, cls_name, bases, namespace, spec=spec, **kwargs
    )

    return new_class

Functions¶

metaxy.ext.sqlmodel.filter_feature_sqlmodel_metadata ¶

filter_feature_sqlmodel_metadata(store: IbisMetadataStore, source_metadata: MetaData, project: str | None = None, filter_by_project: bool = True, inject_primary_key: bool | None = None, inject_index: bool | None = None) -> tuple[str, MetaData]

Get SQLAlchemy URL and filtered SQLModel feature metadata for a metadata store.

This function transforms SQLModel table names to include the store's table_prefix, ensuring that table names in the metadata match what's expected in the database.

You can pass SQLModel.metadata directly - this function will transform table names by adding the store's table_prefix. The returned metadata will have prefixed table names that match the actual database tables.

This function must be called after init_metaxy() to ensure features are loaded.

Parameters:

store (IbisMetadataStore) –

IbisMetadataStore instance (provides table_prefix and sqlalchemy_url)
source_metadata (MetaData) –

Source SQLAlchemy MetaData to filter (typically SQLModel.metadata). Tables are looked up in this metadata by their unprefixed names.
project (str | None, default: None ) –

Project name to filter by. If None, uses MetaxyConfig.get().project
filter_by_project (bool, default: True ) –

If True, only include features for the specified project.
inject_primary_key (bool | None, default: None ) –

If True, inject composite primary key constraints. If False, do not inject. If None, uses config default.
inject_index (bool | None, default: None ) –

If True, inject composite index. If False, do not inject. If None, uses config default.

Returns:

tuple[str, MetaData] –

Tuple of (sqlalchemy_url, filtered_metadata)

Raises:

ValueError –

If store's sqlalchemy_url is empty

Example:

```py
from sqlmodel import SQLModel
from metaxy.ext.sqlmodel import filter_feature_sqlmodel_metadata
from metaxy import init_metaxy
from metaxy.config import MetaxyConfig

# Load features first
init_metaxy()

# Get store instance
config = MetaxyConfig.get()
store = config.get_store("my_store")

# Filter SQLModel metadata with prefix transformation
url, metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)

# Use with Alembic env.py
from alembic import context
url, target_metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)
context.configure(url=url, target_metadata=target_metadata)
```

Source code in src/metaxy/ext/sqlmodel/plugin.py

def filter_feature_sqlmodel_metadata(
    store: "IbisMetadataStore",
    source_metadata: "MetaData",
    project: str | None = None,
    filter_by_project: bool = True,
    inject_primary_key: bool | None = None,
    inject_index: bool | None = None,
) -> tuple[str, "MetaData"]:
    """Get SQLAlchemy URL and filtered SQLModel feature metadata for a metadata store.

    This function transforms SQLModel table names to include the store's table_prefix,
    ensuring that table names in the metadata match what's expected in the database.

    You can pass `SQLModel.metadata` directly - this function will transform table names
    by adding the store's `table_prefix`. The returned metadata will have prefixed table
    names that match the actual database tables.

    This function must be called after init_metaxy() to ensure features are loaded.

    Args:
        store: IbisMetadataStore instance (provides table_prefix and sqlalchemy_url)
        source_metadata: Source SQLAlchemy MetaData to filter (typically SQLModel.metadata).
                        Tables are looked up in this metadata by their unprefixed names.
        project: Project name to filter by. If None, uses MetaxyConfig.get().project
        filter_by_project: If True, only include features for the specified project.
        inject_primary_key: If True, inject composite primary key constraints.
                           If False, do not inject. If None, uses config default.
        inject_index: If True, inject composite index.
                     If False, do not inject. If None, uses config default.

    Returns:
        Tuple of (sqlalchemy_url, filtered_metadata)

    Raises:
        ValueError: If store's sqlalchemy_url is empty

    Example:

        ```py
        from sqlmodel import SQLModel
        from metaxy.ext.sqlmodel import filter_feature_sqlmodel_metadata
        from metaxy import init_metaxy
        from metaxy.config import MetaxyConfig

        # Load features first
        init_metaxy()

        # Get store instance
        config = MetaxyConfig.get()
        store = config.get_store("my_store")

        # Filter SQLModel metadata with prefix transformation
        url, metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)

        # Use with Alembic env.py
        from alembic import context
        url, target_metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)
        context.configure(url=url, target_metadata=target_metadata)
        ```
    """

    from sqlalchemy import MetaData

    config = MetaxyConfig.get()

    if project is None:
        project = config.project

    # Check plugin config for defaults
    sqlmodel_config = config.get_plugin("sqlmodel", SQLModelPluginConfig)
    if inject_primary_key is None:
        inject_primary_key = sqlmodel_config.inject_primary_key
    if inject_index is None:
        inject_index = sqlmodel_config.inject_index

    # Get SQLAlchemy URL from store
    if not store.sqlalchemy_url:
        raise ValueError("IbisMetadataStore has an empty `sqlalchemy_url`.")
    url = store.sqlalchemy_url

    # Create new metadata with transformed table names
    filtered_metadata = MetaData()

    # Get the FeatureGraph to look up feature classes by key
    from metaxy.models.feature import FeatureGraph

    feature_graph = FeatureGraph.get_active()

    # Iterate over tables in source metadata
    for table_name, original_table in source_metadata.tables.items():
        # Check if this table has Metaxy feature metadata
        if metaxy_system_info := original_table.info.get("metaxy-system"):
            metaxy_info = MetaxyTableInfo.model_validate(metaxy_system_info)
            feature_key = metaxy_info.feature_key
        else:
            continue
        # Look up the feature class from the FeatureGraph
        feature_cls = feature_graph.features_by_key.get(feature_key)
        if feature_cls is None:
            # Skip tables for features that aren't registered
            continue

        # Filter by project if requested
        if filter_by_project:
            feature_project = getattr(feature_cls, "project", None)
            if feature_project != project:
                continue

        # Compute prefixed name using store's table_prefix
        prefixed_name = store.get_table_name(feature_key)

        # Copy table to new metadata with prefixed name
        new_table = original_table.to_metadata(filtered_metadata, name=prefixed_name)

        # Inject constraints if requested
        if inject_primary_key or inject_index:
            from metaxy.ext.sqlalchemy.plugin import _inject_constraints

            spec = feature_cls.spec()
            _inject_constraints(
                table=new_table,
                spec=spec,
                inject_primary_key=inject_primary_key,
                inject_index=inject_index,
            )

    return url, filtered_metadata

SQLModel API Reference¶

metaxy.ext.sqlmodel ¶

Classes¶

metaxy.ext.sqlmodel.SQLModelPluginConfig ¶

metaxy.ext.sqlmodel.BaseSQLModelFeature pydantic-model ¶

Attributes¶

metaxy_provenance pydantic-field ¶

metaxy_provenance_by_field pydantic-field ¶

metaxy_feature_version pydantic-field ¶

metaxy_feature_spec_version pydantic-field ¶

metaxy_snapshot_version pydantic-field ¶

metaxy_data_version pydantic-field ¶

metaxy_data_version_by_field pydantic-field ¶

metaxy_created_at pydantic-field ¶

metaxy_materialization_id pydantic-field ¶

Functions¶

table_name classmethod ¶

feature_version classmethod ¶

feature_spec_version classmethod ¶

full_definition_version classmethod ¶

provenance_by_field classmethod ¶

load_input classmethod ¶

resolve_data_version_diff classmethod ¶

metaxy.ext.sqlmodel.SQLModelFeatureMeta ¶

Functions¶

__new__ ¶

Functions¶

metaxy.ext.sqlmodel.filter_feature_sqlmodel_metadata ¶

metaxy.ext.sqlmodel.BaseSQLModelFeature `pydantic-model` ¶

metaxy_provenance `pydantic-field` ¶

metaxy_provenance_by_field `pydantic-field` ¶

metaxy_feature_version `pydantic-field` ¶

metaxy_feature_spec_version `pydantic-field` ¶

metaxy_snapshot_version `pydantic-field` ¶

metaxy_data_version `pydantic-field` ¶

metaxy_data_version_by_field `pydantic-field` ¶

metaxy_created_at `pydantic-field` ¶

metaxy_materialization_id `pydantic-field` ¶

table_name `classmethod` ¶

feature_version `classmethod` ¶

feature_spec_version `classmethod` ¶

full_definition_version `classmethod` ¶

provenance_by_field `classmethod` ¶

load_input `classmethod` ¶

resolve_data_version_diff `classmethod` ¶

new ¶