SQLAlchemy API Reference¶
SQLAlchemy integration for metaxy.
This module provides SQLAlchemy table definitions and helpers for metaxy. These can be used with migration tools like Alembic.
The main functions return tuples of (sqlalchemy_url, metadata) for easy integration with migration tools:
get_system_slqa_metadata: Get URL and system table metadata for a storefilter_feature_sqla_metadata: Get URL and feature table metadata for a store
Classes¶
Bases: PluginConfig
Configuration for SQLAlchemy integration.
This plugin provides helpers for working with SQLAlchemy metadata and table definitions.
Functions¶
filter_feature_sqla_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 feature table metadata for a metadata store.
This function filters the source metadata to include only feature tables belonging to the specified project, and returns the connection URL for the store.
This function must be called after init_metaxy() to ensure features are loaded.
Parameters:
-
store(IbisMetadataStore) –IbisMetadataStore instance
-
source_metadata(MetaData) –Source SQLAlchemy MetaData to filter.
-
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. If False, include all features.
-
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
-
ImportError–If source_metadata is None and SQLModel is not installed
Note
Metadata stores do their best at providing the correct sqlalchemy_url, so you typically don't need to modify the output of this function.
Example: Basic Usage
```py
from metaxy.ext.sqlalchemy import filter_feature_sqla_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")
# With custom metadata
from sqlalchemy import MetaData
my_metadata = MetaData()
# ... define tables in my_metadata ...
# apply the filter function
url, metadata = filter_feature_sqla_metadata(store, source_metadata=my_metadata)
```
Example: With SQLModel
```py
# With SQLModel
from sqlmodel import SQLModel
url, metadata = filter_feature_sqla_metadata(store, SQLModel.metadata)
```
Source code in src/metaxy/ext/sqlalchemy/plugin.py
def filter_feature_sqla_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 feature table metadata for a metadata store.
This function filters the source metadata to include only feature tables
belonging to the specified project, and returns the connection URL for the store.
This function must be called after init_metaxy() to ensure features are loaded.
Args:
store: IbisMetadataStore instance
source_metadata: Source SQLAlchemy MetaData to filter.
project: Project name to filter by. If None, uses MetaxyConfig.get().project
filter_by_project: If True, only include features for the specified project.
If False, include all features.
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
ImportError: If source_metadata is None and SQLModel is not installed
Note:
Metadata stores do their best at providing the correct `sqlalchemy_url`, so you typically don't need to modify the output of this function.
Example: Basic Usage
```py
from metaxy.ext.sqlalchemy import filter_feature_sqla_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")
# With custom metadata
from sqlalchemy import MetaData
my_metadata = MetaData()
# ... define tables in my_metadata ...
# apply the filter function
url, metadata = filter_feature_sqla_metadata(store, source_metadata=my_metadata)
```
Example: With SQLModel
```py
# With SQLModel
from sqlmodel import SQLModel
url, metadata = filter_feature_sqla_metadata(store, SQLModel.metadata)
```
"""
url = _get_store_sqlalchemy_url(store, protocol=protocol, port=port)
metadata = _get_features_metadata(
source_metadata=source_metadata,
store=store,
project=project,
filter_by_project=filter_by_project,
inject_primary_key=inject_primary_key,
inject_index=inject_index,
)
return url, metadata
get_system_slqa_metadata(store: IbisMetadataStore, protocol: str | None = None, port: int | None = None) -> tuple[str, MetaData]
Get SQLAlchemy URL and Metaxy system tables metadata for a metadata store.
This function retrieves both the connection URL and system table metadata
for a store, with the store's table_prefix automatically applied to table names.
Parameters:
-
store(IbisMetadataStore) –IbisMetadataStore instance
-
protocol(str | None, default:None) –Optional protocol (drivername) 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
Metadata stores do their best at providing the correct sqlalchemy_url, so you typically don't need to modify the output of this function.
Source code in src/metaxy/ext/sqlalchemy/plugin.py
def get_system_slqa_metadata(
store: IbisMetadataStore,
protocol: str | None = None,
port: int | None = None,
) -> tuple[str, MetaData]:
"""Get SQLAlchemy URL and Metaxy system tables metadata for a metadata store.
This function retrieves both the connection URL and system table metadata
for a store, with the store's `table_prefix` automatically applied to table names.
Args:
store: IbisMetadataStore instance
protocol: Optional protocol (drivername) 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, system_metadata)
Raises:
ValueError: If store's sqlalchemy_url is empty
Note:
Metadata stores do their best at providing the correct `sqlalchemy_url`, so you typically don't need to modify the output of this function.
"""
url = _get_store_sqlalchemy_url(store, protocol=protocol, port=port)
metadata = _get_system_metadata(table_prefix=store._table_prefix)
return url, metadata