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) -> 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.
Returns:
Raises:
-
ValueError–If store's sqlalchemy_url is empty
-
ImportError–If source_metadata is None and SQLModel is not installed
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,
) -> 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.
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
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)
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) -> 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
Returns:
Raises:
-
ValueError–If store's sqlalchemy_url is empty
Source code in src/metaxy/ext/sqlalchemy/plugin.py
def get_system_slqa_metadata(
store: IbisMetadataStore,
) -> 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
Returns:
Tuple of (sqlalchemy_url, system_metadata)
Raises:
ValueError: If store's sqlalchemy_url is empty
"""
url = _get_store_sqlalchemy_url(store)
metadata = _get_system_metadata(table_prefix=store._table_prefix)
return url, metadata