SQLModel¶
Experimental
This functionality is experimental.
SQLModel combines SQLAlchemy and Pydantic into a single ORM. If you want your Metaxy feature definitions to double as ORM models, enable the SQLModel integration. This exposes user-defined feature tables directly to SQLAlchemy.
It is the primary way to use Metaxy with database-backed metadata stores.
Database Migrations
For database migration management with Alembic, see the SQLAlchemy integration guide.
Installation¶
The SQLModel integration requires the sqlmodel package:
and has to be enabled explicitly:
Usage¶
The SQLModel integration provides BaseSQLModelFeature which combines the functionality of a Metaxy feature and an SQLModel table.
import metaxy as mx
import metaxy.ext.sqlmodel as mxsql
from sqlmodel import Field
class VideoFeature(
mxsql.BaseSQLModelFeature,
table=True,
spec=mx.FeatureSpec(
key=FeatureKey(["video"]),
id_columns=["video_id"],
fields=[
"frames",
"duration",
],
),
):
# User-defined metadata columns
video_id: str
path: str
duration: float
Do Not Use Server-Generated IDs
ID columns should not be server-generated because they are typically used to determine data locations such as object storage keys, so they have to be defined before metadata is inserted into the database
Automatic Table Naming
When __tablename__ is not specified, it is automatically generated from the feature key. For FeatureKey(["video", "processing"]), it becomes "video__processing". This behavior can be disabled in the plugin configuration.
Database Migrations¶
When using SQLModel features with Alembic or other migration tools, use filter_feature_sqlmodel_metadata() to transform table names and filter metadata.
Table Name Transformation
Pass SQLModel.metadata to filter_feature_sqlmodel_metadata() and it 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.
from sqlmodel import SQLModel
from metaxy.ext.sqlmodel import filter_feature_sqlmodel_metadata
from metaxy.config import MetaxyConfig
from metaxy import init
config = init()
store = config.get_store()
# Transform SQLModel metadata with table_prefix
url, target_metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)
# Use with Alembic env.py
from alembic import context
context.configure(url=url, target_metadata=target_metadata)
The filter_feature_sqlmodel_metadata() function:
- Transforms table names by adding the store's
table_prefix - Filters tables by project (configurable)
- Returns the SQLAlchemy URL for the store
- Optionally injects primary key and index constraints
See the SQLAlchemy integration guide for complete Alembic setup examples.
Separate Alembic Version Tables
When managing both system tables and feature tables with Alembic, use separate version tables to avoid conflicts. See the Multi-Store Setup section for configuration details.
pydantic-model
¶
Bases: SQLModel, BaseFeature
Base class for Metaxy features that are also SQLModel tables.
Example
from metaxy.integrations.sqlmodel import BaseSQLModelFeature
from sqlmodel import Field
class VideoFeature(
BaseSQLModelFeature,
table=True,
spec=mx.FeatureSpec(
key=mx.FeatureKey(["video"]),
id_columns=["uid"],
fields=[
mx.FieldSpec(
key=mx.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:
{
"additionalProperties": false,
"description": "Base class for `Metaxy` features that are also `SQLModel` tables.\n\n!!! example\n\n <!-- skip next -->\n ```py\n from metaxy.integrations.sqlmodel import BaseSQLModelFeature\n from sqlmodel import Field\n\n\n class VideoFeature(\n BaseSQLModelFeature,\n table=True,\n spec=mx.FeatureSpec(\n key=mx.FeatureKey([\"video\"]),\n id_columns=[\"uid\"],\n fields=[\n mx.FieldSpec(\n key=mx.FieldKey([\"video_file\"]),\n code_version=\"1\",\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_project_version": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Hash of the entire feature graph snapshot",
"title": "Metaxy Project 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_updated_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Timestamp when the metadata row was last updated (UTC)",
"title": "Metaxy Updated At"
},
"metaxy_deleted_at": {
"anyOf": [
{
"format": "date-time",
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "Soft delete timestamp (UTC); null means active row",
"title": "Metaxy Deleted At"
}
},
"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_project_version(str | None) -
metaxy_data_version(str | None) -
metaxy_data_version_by_field(dict[str, str] | None) -
metaxy_created_at(AwareDatetime | None) -
metaxy_updated_at(AwareDatetime | None) -
metaxy_materialization_id(str | None) -
metaxy_deleted_at(AwareDatetime | None)
Validators:
-
_validate_id_columns_exist
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,
protocol: str | None = None,
port: int | 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() 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.
-
protocol(str | None, default:None) –Optional protocol to replace the existing one in the URL. Useful when Ibis uses a different protocol than SQLAlchemy requires.
-
port(int | None, default:None) –Optional port to replace the existing one in the URL. Useful when the SQLAlchemy driver uses a different port than Ibis.
Returns:
Raises:
-
ValueError–If store's sqlalchemy_url is empty
Note
For ClickHouse, the sqlalchemy_url property already returns the native
protocol with port 9000, so you typically don't need to override these.
Example: Basic Usage
<!-- skip next -->
```py
from sqlmodel import SQLModel
from metaxy.ext.sqlmodel import filter_feature_sqlmodel_metadata
from alembic import context
# Load features first
mx.init()
# Get store instance
config = mx.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
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
@public
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,
protocol: str | None = None,
port: int | 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() 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.
protocol: Optional protocol to replace the existing one in the URL.
Useful when Ibis uses a different protocol than SQLAlchemy requires.
port: Optional port to replace the existing one in the URL.
Useful when the SQLAlchemy driver uses a different port than Ibis.
Returns:
Tuple of (sqlalchemy_url, filtered_metadata)
Raises:
ValueError: If store's sqlalchemy_url is empty
Note:
For ClickHouse, the `sqlalchemy_url` property already returns the native
protocol with port 9000, so you typically don't need to override these.
Example: Basic Usage
<!-- skip next -->
```py
from sqlmodel import SQLModel
from metaxy.ext.sqlmodel import filter_feature_sqlmodel_metadata
from alembic import context
# Load features first
mx.init()
# Get store instance
config = mx.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
url, target_metadata = filter_feature_sqlmodel_metadata(store, SQLModel.metadata)
context.configure(url=url, target_metadata=target_metadata)
```
"""
from sqlalchemy import MetaData
from metaxy.ext.sqlalchemy.plugin import _get_store_sqlalchemy_url
config = MetaxyConfig.get(load=True)
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
url = _get_store_sqlalchemy_url(store, protocol=protocol, port=port)
# 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 definition from the FeatureGraph
definition = feature_graph.feature_definitions_by_key.get(feature_key)
if definition is None:
# Skip tables for features that aren't registered
continue
# Filter by project if requested
if filter_by_project:
if definition.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
_inject_constraints(
table=new_table,
spec=definition.spec,
inject_primary_key=inject_primary_key,
inject_index=inject_index,
)
return url, filtered_metadata
Configuration¶
enable¶
Whether to enable the plugin.
Type: bool | Default: False
inject_primary_key¶
Automatically inject composite primary key constraints on SQLModel tables. The key is composed of ID columns, metaxy_created_at, and metaxy_data_version.
Type: bool | Default: False
inject_index¶
Automatically inject composite index on SQLModel tables. The index covers ID columns, metaxy_created_at, and metaxy_data_version.
Type: bool | Default: False