Configuration¶
This is the Python SDK for Metaxy's configuration. See config file reference to learn how to configure Metaxy via TOML files.
metaxy.MetaxyConfig
pydantic-model
¶
MetaxyConfig(
__pydantic_self__,
_case_sensitive: bool | None = None,
_nested_model_default_partial_update: bool
| None = None,
_env_prefix: str | None = None,
_env_file: DotenvType | None = ENV_FILE_SENTINEL,
_env_file_encoding: str | None = None,
_env_ignore_empty: bool | None = None,
_env_nested_delimiter: str | None = None,
_env_nested_max_split: int | None = None,
_env_parse_none_str: str | None = None,
_env_parse_enums: bool | None = None,
_cli_prog_name: str | None = None,
_cli_parse_args: bool
| list[str]
| tuple[str, ...]
| None = None,
_cli_settings_source: CliSettingsSource[Any]
| None = None,
_cli_parse_none_str: str | None = None,
_cli_hide_none_type: bool | None = None,
_cli_avoid_json: bool | None = None,
_cli_enforce_required: bool | None = None,
_cli_use_class_docs_for_groups: bool | None = None,
_cli_exit_on_error: bool | None = None,
_cli_prefix: str | None = None,
_cli_flag_prefix_char: str | None = None,
_cli_implicit_flags: bool | None = None,
_cli_ignore_unknown_args: bool | None = None,
_cli_kebab_case: bool
| Literal["all", "no_enums"]
| None = None,
_cli_shortcuts: Mapping[str, str | list[str]]
| None = None,
_secrets_dir: PathType | None = None,
**values: Any,
)
Bases: BaseSettings
Main Metaxy configuration.
Loads from (in order of precedence):
-
Init arguments
-
Environment variables (METAXY_*)
-
Config file (
metaxy.tomlor[tool.metaxy]inpyproject.toml)
Environment variables can be templated with ${MY_VAR:-default} syntax.
Show JSON schema:
{
"$defs": {
"FeatureKey": {
"description": "Feature key as a sequence of string parts.\n\nHashable for use as dict keys in registries.\nParts cannot contain forward slashes (/) or double underscores (__).\n\nExample:\n\n ```py\n FeatureKey(\"a/b/c\") # String format\n # FeatureKey(parts=['a', 'b', 'c'])\n\n FeatureKey([\"a\", \"b\", \"c\"]) # List format\n # FeatureKey(parts=['a', 'b', 'c'])\n\n FeatureKey(FeatureKey([\"a\", \"b\", \"c\"])) # FeatureKey copy\n # FeatureKey(parts=['a', 'b', 'c'])\n ```",
"items": {
"type": "string"
},
"title": "FeatureKey",
"type": "array"
},
"FeatureSelection": {
"additionalProperties": false,
"description": "Selects a set of features from a metadata store.\n\nFields can be combined \u2014 both `projects` and `keys` may be set simultaneously\nto select all features from those projects *plus* the individually listed keys.\n\nSet `all=True` to select every feature in the store.\n\nSupports set operators `|`, `&`, and `-` which merge the underlying fields.\n\nExamples:\n >>> mx.FeatureSelection(projects=[\"my-project\"])\n FeatureSelection(projects=['my-project'], keys=None, all=None)\n\n >>> mx.FeatureSelection(keys=[\"raw/video\", mx.FeatureKey(\"ml/embeddings\")])\n FeatureSelection(projects=None, keys=[raw/video, ml/embeddings], all=None)\n\n >>> mx.FeatureSelection(all=True)\n FeatureSelection(projects=None, keys=None, all=True)\n\n >>> mx.FeatureSelection(projects=[\"a\"]) | mx.FeatureSelection(projects=[\"b\"])\n FeatureSelection(projects=['a', 'b'], keys=None, all=None)\n\n >>> mx.FeatureSelection(keys=[\"a/b\", \"c/d\"]) & mx.FeatureSelection(keys=[\"c/d\"])\n FeatureSelection(projects=None, keys=[c/d], all=None)\n\n >>> mx.FeatureSelection(projects=[\"a\", \"b\", \"c\"]) - mx.FeatureSelection(projects=[\"b\"])\n FeatureSelection(projects=['a', 'c'], keys=None, all=None)",
"properties": {
"projects": {
"anyOf": [
{
"items": {
"type": "string"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"title": "Projects"
},
"keys": {
"anyOf": [
{
"items": {
"$ref": "#/$defs/FeatureKey"
},
"type": "array"
},
{
"type": "null"
}
],
"default": null,
"title": "Keys"
},
"all": {
"anyOf": [
{
"type": "boolean"
},
{
"type": "null"
}
],
"default": null,
"title": "All"
}
},
"title": "FeatureSelection",
"type": "object"
},
"PluginConfig": {
"additionalProperties": true,
"description": "Configuration for Metaxy plugins",
"properties": {
"enable": {
"default": false,
"description": "Whether to enable the plugin.",
"title": "Enable",
"type": "boolean"
}
},
"title": "PluginConfig",
"type": "object"
},
"StoreConfig": {
"additionalProperties": false,
"description": "Configuration options for metadata stores.",
"properties": {
"type": {
"description": "Full import path to metadata store class (e.g., `\"metaxy.ext.metadata_stores.duckdb.DuckDBMetadataStore\"`)",
"title": "Type",
"type": "string"
},
"config": {
"additionalProperties": true,
"description": "Store-specific configuration parameters (constructor kwargs). Includes `fallback_stores`, database connection parameters, etc.",
"title": "Config",
"type": "object"
}
},
"required": [
"type"
],
"title": "StoreConfig",
"type": "object"
}
},
"additionalProperties": false,
"description": "Main Metaxy configuration.\n\nLoads from (in order of precedence):\n\n1. Init arguments\n\n2. Environment variables (METAXY_*)\n\n3. Config file (`metaxy.toml` or `[tool.metaxy]` in `pyproject.toml` )\n\nEnvironment variables can be templated with `${MY_VAR:-default}` syntax.\n\nExample: Accessing current configuration\n <!-- skip next -->\n ```py\n config = MetaxyConfig.load()\n ```\n\nExample: Getting a configured metadata store\n ```py\n store = config.get_store(\"prod\")\n ```",
"properties": {
"store": {
"default": "dev",
"description": "Default metadata store to use",
"title": "Store",
"type": "string"
},
"stores": {
"additionalProperties": {
"$ref": "#/$defs/StoreConfig"
},
"description": "Named store configurations",
"title": "Stores",
"type": "object"
},
"migrations_dir": {
"default": ".metaxy/migrations",
"description": "Directory where migration files are stored",
"title": "Migrations Dir",
"type": "string"
},
"entrypoints": {
"description": "List of Python module paths to load for feature discovery",
"items": {
"type": "string"
},
"title": "Entrypoints",
"type": "array"
},
"theme": {
"default": "default",
"description": "Graph rendering theme for CLI visualization",
"title": "Theme",
"type": "string"
},
"ext": {
"additionalProperties": {
"$ref": "#/$defs/PluginConfig"
},
"description": "Configuration for Metaxy integrations with third-party tools",
"title": "Ext",
"type": "object"
},
"hash_truncation_length": {
"default": 8,
"description": "Truncate hash values to this length.",
"minimum": 8,
"title": "Hash Truncation Length",
"type": "integer"
},
"auto_create_tables": {
"default": false,
"description": "Auto-create tables when opening stores. It is not advised to enable this setting in production.",
"title": "Auto Create Tables",
"type": "boolean"
},
"project": {
"anyOf": [
{
"type": "string"
},
{
"type": "null"
}
],
"default": null,
"description": "[Project](/guide/concepts/projects.md) name. Used to scope operations to enable multiple independent projects in a shared metadata store. Does not modify feature keys or table names. Project names must be valid alphanumeric strings with dashes, underscores, and cannot contain forward slashes (`/`) or double underscores (`__`)",
"title": "Project"
},
"locked": {
"anyOf": [
{
"type": "boolean"
},
{
"type": "null"
}
],
"default": null,
"description": "Whether to raise an error if an external feature doesn't have a matching feature version when [syncing external features][metaxy.sync_external_features] from the metadata store.",
"title": "Locked"
},
"sync": {
"default": true,
"description": "Whether to automatically [sync external feature definitions][metaxy.sync_external_features] from the metadata during some operations. It's recommended to keep this enabled as it ensures versioning correctness for external feature definitions with a negligible performance impact.",
"title": "Sync",
"type": "boolean"
},
"extra_features": {
"description": "Extra features to load from the metadata store when calling [`sync_external_features`][metaxy.sync_external_features]. Each entry is a [`FeatureSelection`][metaxy.FeatureSelection]. All entries are combined together. Learn more [here](/guide/concepts/definitions/external-features.md/#loading-extra-features).",
"items": {
"$ref": "#/$defs/FeatureSelection"
},
"title": "Extra Features",
"type": "array"
},
"metaxy_lock_path": {
"default": "metaxy.lock",
"description": "Relative or absolute path to the lock file, resolved from the config file's location.",
"title": "Metaxy Lock Path",
"type": "string"
}
},
"title": "MetaxyConfig",
"type": "object"
}
Config:
env_prefix:METAXY_env_nested_delimiter:__frozen:True
Source code in .venv/lib/python3.10/site-packages/pydantic_settings/main.py
def __init__(
__pydantic_self__,
_case_sensitive: bool | None = None,
_nested_model_default_partial_update: bool | None = None,
_env_prefix: str | None = None,
_env_file: DotenvType | None = ENV_FILE_SENTINEL,
_env_file_encoding: str | None = None,
_env_ignore_empty: bool | None = None,
_env_nested_delimiter: str | None = None,
_env_nested_max_split: int | None = None,
_env_parse_none_str: str | None = None,
_env_parse_enums: bool | None = None,
_cli_prog_name: str | None = None,
_cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,
_cli_settings_source: CliSettingsSource[Any] | None = None,
_cli_parse_none_str: str | None = None,
_cli_hide_none_type: bool | None = None,
_cli_avoid_json: bool | None = None,
_cli_enforce_required: bool | None = None,
_cli_use_class_docs_for_groups: bool | None = None,
_cli_exit_on_error: bool | None = None,
_cli_prefix: str | None = None,
_cli_flag_prefix_char: str | None = None,
_cli_implicit_flags: bool | None = None,
_cli_ignore_unknown_args: bool | None = None,
_cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,
_cli_shortcuts: Mapping[str, str | list[str]] | None = None,
_secrets_dir: PathType | None = None,
**values: Any,
) -> None:
super().__init__(
**__pydantic_self__._settings_build_values(
values,
_case_sensitive=_case_sensitive,
_nested_model_default_partial_update=_nested_model_default_partial_update,
_env_prefix=_env_prefix,
_env_file=_env_file,
_env_file_encoding=_env_file_encoding,
_env_ignore_empty=_env_ignore_empty,
_env_nested_delimiter=_env_nested_delimiter,
_env_nested_max_split=_env_nested_max_split,
_env_parse_none_str=_env_parse_none_str,
_env_parse_enums=_env_parse_enums,
_cli_prog_name=_cli_prog_name,
_cli_parse_args=_cli_parse_args,
_cli_settings_source=_cli_settings_source,
_cli_parse_none_str=_cli_parse_none_str,
_cli_hide_none_type=_cli_hide_none_type,
_cli_avoid_json=_cli_avoid_json,
_cli_enforce_required=_cli_enforce_required,
_cli_use_class_docs_for_groups=_cli_use_class_docs_for_groups,
_cli_exit_on_error=_cli_exit_on_error,
_cli_prefix=_cli_prefix,
_cli_flag_prefix_char=_cli_flag_prefix_char,
_cli_implicit_flags=_cli_implicit_flags,
_cli_ignore_unknown_args=_cli_ignore_unknown_args,
_cli_kebab_case=_cli_kebab_case,
_cli_shortcuts=_cli_shortcuts,
_secrets_dir=_secrets_dir,
)
)
Attributes¶
metaxy.MetaxyConfig.stores
pydantic-field
¶
stores: dict[str, StoreConfig]
Named store configurations
metaxy.MetaxyConfig.migrations_dir
pydantic-field
¶
migrations_dir: str = '.metaxy/migrations'
Directory where migration files are stored
metaxy.MetaxyConfig.entrypoints
pydantic-field
¶
List of Python module paths to load for feature discovery
metaxy.MetaxyConfig.theme
pydantic-field
¶
theme: str = 'default'
Graph rendering theme for CLI visualization
metaxy.MetaxyConfig.ext
pydantic-field
¶
Configuration for Metaxy integrations with third-party tools
metaxy.MetaxyConfig.hash_truncation_length
pydantic-field
¶
hash_truncation_length: int = 8
Truncate hash values to this length.
metaxy.MetaxyConfig.auto_create_tables
pydantic-field
¶
auto_create_tables: bool = False
Auto-create tables when opening stores. It is not advised to enable this setting in production.
metaxy.MetaxyConfig.project
pydantic-field
¶
project: str | None = None
Project name. Used to scope operations to enable multiple independent projects in a shared metadata store. Does not modify feature keys or table names. Project names must be valid alphanumeric strings with dashes, underscores, and cannot contain forward slashes (/) or double underscores (__)
metaxy.MetaxyConfig.locked
pydantic-field
¶
locked: bool | None = None
Whether to raise an error if an external feature doesn't have a matching feature version when syncing external features from the metadata store.
metaxy.MetaxyConfig.sync
pydantic-field
¶
sync: bool = True
Whether to automatically sync external feature definitions from the metadata during some operations. It's recommended to keep this enabled as it ensures versioning correctness for external feature definitions with a negligible performance impact.
metaxy.MetaxyConfig.extra_features
pydantic-field
¶
extra_features: list[FeatureSelection]
Extra features to load from the metadata store when calling sync_external_features. Each entry is a FeatureSelection. All entries are combined together. Learn more here.
metaxy.MetaxyConfig.metaxy_lock_path
pydantic-field
¶
metaxy_lock_path: str = 'metaxy.lock'
Relative or absolute path to the lock file, resolved from the config file's location.
metaxy.MetaxyConfig.config_file
property
¶
config_file: Path | None
The config file path used to load this configuration.
Returns None if the config was created directly (not via MetaxyConfig.load).
metaxy.MetaxyConfig.lock_file
property
¶
lock_file: Path | None
The resolved lock file path.
Returns the absolute path if metaxy_lock_path is absolute, otherwise
resolves it relative to the config file's directory.
Returns None if the path is relative and no config file is set.
metaxy.MetaxyConfig.plugins
property
¶
Returns all enabled plugin names from ext configuration.
Functions¶
metaxy.MetaxyConfig.validate_project
pydantic-validator
¶
Validate project name follows naming rules.
Source code in src/metaxy/config.py
@field_validator("project")
@classmethod
def validate_project(cls, v: str | None) -> str | None:
"""Validate project name follows naming rules."""
if v is None:
return None
if not v:
raise ValueError("project name cannot be empty")
if "/" in v:
raise ValueError(
f"project name '{v}' cannot contain forward slashes (/). "
f"Forward slashes are reserved for FeatureKey separation"
)
if "__" in v:
raise ValueError(
f"project name '{v}' cannot contain double underscores (__). "
f"Double underscores are reserved for table name generation"
)
import re
if not re.match(r"^[a-zA-Z0-9_-]+$", v):
raise ValueError(f"project name '{v}' must contain only alphanumeric characters, underscores, and hyphens")
return v
metaxy.MetaxyConfig.get_plugin
classmethod
¶
Get the plugin config from the global Metaxy config.
Unlike get(), this method does not warn when the global config is not
initialized. This is intentional because plugins may call this at import
time to read their configuration, and returning default plugin config
is always safe.
Source code in src/metaxy/config.py
@classmethod
def get_plugin(cls, name: str, plugin_cls: type[PluginConfigT]) -> PluginConfigT:
"""Get the plugin config from the global Metaxy config.
Unlike `get()`, this method does not warn when the global config is not
initialized. This is intentional because plugins may call this at import
time to read their configuration, and returning default plugin config
is always safe.
"""
ext = cls.get(_allow_default_config=True).ext
if name in ext:
existing = ext[name]
if isinstance(existing, plugin_cls):
# Already the correct type
plugin = existing
else:
# Convert from generic PluginConfig or dict to specific plugin class
plugin = plugin_cls.model_validate(existing.model_dump())
else:
# Return default config if plugin not configured
plugin = plugin_cls()
return plugin
metaxy.MetaxyConfig.settings_customise_sources
classmethod
¶
settings_customise_sources(
settings_cls: type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]
Customize settings sources: init β env β TOML.
Priority (first wins): 1. Init arguments 2. Environment variables 3. TOML file
Source code in src/metaxy/config.py
@classmethod
def settings_customise_sources(
cls,
settings_cls: type[BaseSettings],
init_settings: PydanticBaseSettingsSource,
env_settings: PydanticBaseSettingsSource,
dotenv_settings: PydanticBaseSettingsSource,
file_secret_settings: PydanticBaseSettingsSource,
) -> tuple[PydanticBaseSettingsSource, ...]:
"""Customize settings sources: init β env β TOML.
Priority (first wins):
1. Init arguments
2. Environment variables
3. TOML file
"""
toml_settings = TomlConfigSettingsSource(settings_cls)
return (init_settings, env_settings, toml_settings)
metaxy.MetaxyConfig.get
classmethod
¶
get(
*,
load: bool = False,
_allow_default_config: bool = False,
) -> MetaxyConfig
Get the current Metaxy configuration.
Parameters:
-
load(bool, default:False) βIf True and config is not set, calls
MetaxyConfig.load()to load configuration from file. Useful for plugins that need config but don't want to require manual initialization. -
_allow_default_config(bool, default:False) βInternal parameter. When True, returns default config without warning if global config is not set. Used by methods like
get_pluginthat may be called at import time.
Source code in src/metaxy/config.py
@classmethod
def get(cls, *, load: bool = False, _allow_default_config: bool = False) -> "MetaxyConfig":
"""Get the current Metaxy configuration.
Args:
load: If True and config is not set, calls `MetaxyConfig.load()` to
load configuration from file. Useful for plugins that need config
but don't want to require manual initialization.
_allow_default_config: Internal parameter. When True, returns default
config without warning if global config is not set. Used by methods
like `get_plugin` that may be called at import time.
"""
cfg = _config_override.get() or _global_config
if cfg is None:
if load:
return cls.load()
if not _allow_default_config:
warnings.warn(
UserWarning(
"Global Metaxy configuration not initialized. It can be set with MetaxyConfig.set(config) typically after loading it from a toml file. Returning default configuration (with environment variables and other pydantic settings sources resolved)."
),
stacklevel=2,
)
return cls()
else:
return cfg
metaxy.MetaxyConfig.set
classmethod
¶
metaxy.MetaxyConfig.reset
classmethod
¶
metaxy.MetaxyConfig.use
¶
use() -> Iterator[Self]
Use this configuration temporarily, restoring previous config on exit.
Example
Source code in src/metaxy/config.py
@contextmanager
def use(self) -> Iterator[Self]:
"""Use this configuration temporarily, restoring previous config on exit.
Example:
```py
test_config = MetaxyConfig(project="test")
with test_config.use():
# Code here uses test config
assert MetaxyConfig.get().project == "test"
# Previous config restored
```
"""
token = _config_override.set(self)
try:
yield self
finally:
_config_override.reset(token)
metaxy.MetaxyConfig.load
classmethod
¶
load(
config_file: str | Path | None = None,
*,
search_parents: bool = True,
auto_discovery_start: Path | None = None,
) -> MetaxyConfig
Load config with auto-discovery and parent directory search.
Parameters:
-
config_file(str | Path | None, default:None) βOptional config file path.
Tip
METAXY_CONFIGenvironment variable can be used to set this parameter -
search_parents(bool, default:True) βSearch parent directories for config file
-
auto_discovery_start(Path | None, default:None) βDirectory to start search from. Defaults to current working directory.
Returns:
-
MetaxyConfigβLoaded config (TOML + env vars merged)
Example
# Auto-discover with parent search
config = MetaxyConfig.load()
# Explicit file
config = MetaxyConfig.load("custom.toml")
# Auto-discover without parent search
config = MetaxyConfig.load(search_parents=False)
# Auto-discover from a specific directory
config = MetaxyConfig.load(auto_discovery_start=Path("/path/to/project"))
Source code in src/metaxy/config.py
@classmethod
def load(
cls,
config_file: str | Path | None = None,
*,
search_parents: bool = True,
auto_discovery_start: Path | None = None,
) -> "MetaxyConfig":
"""Load config with auto-discovery and parent directory search.
Args:
config_file: Optional config file path.
!!! tip
`METAXY_CONFIG` environment variable can be used to set this parameter
search_parents: Search parent directories for config file
auto_discovery_start: Directory to start search from.
Defaults to current working directory.
Returns:
Loaded config (TOML + env vars merged)
Example:
<!-- skip next -->
```py
# Auto-discover with parent search
config = MetaxyConfig.load()
# Explicit file
config = MetaxyConfig.load("custom.toml")
# Auto-discover without parent search
config = MetaxyConfig.load(search_parents=False)
# Auto-discover from a specific directory
config = MetaxyConfig.load(auto_discovery_start=Path("/path/to/project"))
```
"""
# Search for config file if not explicitly provided
if config_from_env := os.getenv("METAXY_CONFIG"):
config_file = Path(config_from_env)
if config_file is None and search_parents:
config_file = cls._discover_config_with_parents(auto_discovery_start)
# For explicit file, temporarily patch the TomlConfigSettingsSource
# to use that file, then use normal instantiation
# This ensures env vars still work
if config_file:
# Create a custom settings source class for this file
toml_path = Path(config_file)
class CustomTomlSource(TomlConfigSettingsSource):
def __init__(self, settings_cls: type[BaseSettings]):
# Skip auto-discovery, use explicit file
super(TomlConfigSettingsSource, self).__init__(settings_cls)
self.toml_file = toml_path
self.toml_data = self._load_toml()
# Customize sources to use custom TOML file
original_method = cls.settings_customise_sources
@classmethod
def custom_sources(
cls_inner,
settings_cls,
init_settings,
env_settings,
dotenv_settings,
file_secret_settings,
):
toml_settings = CustomTomlSource(settings_cls)
return (init_settings, env_settings, toml_settings)
# Temporarily replace method
cls.settings_customise_sources = custom_sources # ty: ignore[invalid-assignment]
try:
config = cls()
finally:
cls.settings_customise_sources = original_method # ty: ignore[invalid-assignment]
# Store the resolved config file path
config._config_file = toml_path.resolve()
else:
# Use default sources (auto-discovery + env vars)
config = cls()
# No config file used
config._config_file = None
cls.set(config)
# Load plugins after config is set (plugins may access MetaxyConfig.get())
config._load_plugins()
return config
metaxy.MetaxyConfig.get_store
¶
get_store(
name: str | None = None,
*,
expected_type: Literal[None] = None,
**kwargs: Any,
) -> MetadataStore
Instantiate metadata store by name.
Parameters:
-
name(str | None, default:None) βStore name (uses config.store if None)
-
expected_type(type[StoreTypeT] | None, default:None) βExpected type of the store. If the actual store type does not match the expected type, a
TypeErroris raised. -
**kwargs(Any, default:{}) βAdditional keyword arguments to pass to the store constructor.
Returns:
-
MetadataStore | StoreTypeTβInstantiated metadata store
Raises:
-
ValueErrorβIf store name not found in config, or if fallback stores have different hash algorithms than the parent store
-
ImportErrorβIf store class cannot be imported
-
TypeErrorβIf the actual store type does not match the expected type
Source code in src/metaxy/config.py
def get_store(
self,
name: str | None = None,
*,
expected_type: type[StoreTypeT] | None = None,
**kwargs: Any,
) -> "MetadataStore | StoreTypeT":
"""Instantiate metadata store by name.
Args:
name: Store name (uses config.store if None)
expected_type: Expected type of the store.
If the actual store type does not match the expected type, a `TypeError` is raised.
**kwargs: Additional keyword arguments to pass to the store constructor.
Returns:
Instantiated metadata store
Raises:
ValueError: If store name not found in config, or if fallback stores
have different hash algorithms than the parent store
ImportError: If store class cannot be imported
TypeError: If the actual store type does not match the expected type
Example:
```py
store = config.get_store("prod")
# Use default store
store = config.get_store()
```
"""
from metaxy.versioning.types import HashAlgorithm
if len(self.stores) == 0:
raise InvalidConfigError.from_config(
self,
"No Metaxy stores available. They should be configured in metaxy.toml|pyproject.toml or via environment variables.",
)
name = name or self.store
if name not in self.stores:
raise InvalidConfigError.from_config(
self,
f"Store '{name}' not found in config. Available stores: {list(self.stores.keys())}",
)
store_config = self.stores[name]
# Get store class (lazily imported on first access)
try:
store_class = store_config.type_cls
except Exception as e:
raise InvalidConfigError.from_config(
self,
f"Failed to import store class '{store_config.type}' for store '{name}': {e}",
) from e
if expected_type is not None and not issubclass(store_class, expected_type):
raise InvalidConfigError.from_config(
self,
f"Store '{name}' is not of type '{expected_type.__name__}'",
)
# Extract configuration and prepare for typed config model
config_copy = store_config.config.copy()
# Get hash_algorithm from config (if specified) and convert to enum
configured_hash_algorithm = config_copy.get("hash_algorithm")
if configured_hash_algorithm is not None:
# Convert string to enum if needed
if isinstance(configured_hash_algorithm, str):
configured_hash_algorithm = HashAlgorithm(configured_hash_algorithm)
config_copy["hash_algorithm"] = configured_hash_algorithm
else:
# Don't set a default here - let the store choose its own default
configured_hash_algorithm = None
# Get the store's config model class and create typed config
config_model_cls = store_class.config_model()
# Get auto_create_tables from global config only if the config model supports it
if (
"auto_create_tables" not in config_copy
and self.auto_create_tables is not None
and "auto_create_tables" in config_model_cls.model_fields
):
# Use global setting from MetaxyConfig if not specified per-store
config_copy["auto_create_tables"] = self.auto_create_tables
# Separate kwargs into config fields and extra constructor args
config_fields = set(config_model_cls.model_fields.keys())
extra_kwargs = {}
for key, value in kwargs.items():
if key in config_fields:
config_copy[key] = value
else:
extra_kwargs[key] = value
try:
typed_config = config_model_cls.model_validate(config_copy)
except Exception as e:
raise InvalidConfigError.from_config(
self,
f"Failed to validate config for store '{name}': {e}",
) from e
# Instantiate using from_config() - fallback stores are resolved via MetaxyConfig.get()
# Use self.use() to ensure this config is available for fallback resolution
try:
with self.use():
store = store_class.from_config(typed_config, name=name, **extra_kwargs)
except InvalidConfigError:
# Don't re-wrap InvalidConfigError (e.g., from nested fallback store resolution)
raise
except Exception as e:
raise InvalidConfigError.from_config(
self,
f"Failed to instantiate store '{name}' ({store_class.__name__}): {e}",
) from e
# Verify the store actually uses the hash algorithm we configured
# (in case a store subclass overrides the default or ignores the parameter)
# Only check if we explicitly configured a hash algorithm
if configured_hash_algorithm is not None and store.hash_algorithm != configured_hash_algorithm:
raise InvalidConfigError.from_config(
self,
f"Store '{name}' ({store_class.__name__}) was configured with "
f"hash_algorithm='{configured_hash_algorithm.value}' but is using "
f"'{store.hash_algorithm.value}'. The store class may have overridden "
f"the hash algorithm. All stores must use the same hash algorithm.",
)
if expected_type is not None and not isinstance(store, expected_type):
raise InvalidConfigError.from_config(
self,
f"Store '{name}' is not of type '{expected_type.__name__}'",
)
return store
metaxy.StoreConfig
pydantic-model
¶
StoreConfig(
__pydantic_self__,
_case_sensitive: bool | None = None,
_nested_model_default_partial_update: bool
| None = None,
_env_prefix: str | None = None,
_env_file: DotenvType | None = ENV_FILE_SENTINEL,
_env_file_encoding: str | None = None,
_env_ignore_empty: bool | None = None,
_env_nested_delimiter: str | None = None,
_env_nested_max_split: int | None = None,
_env_parse_none_str: str | None = None,
_env_parse_enums: bool | None = None,
_cli_prog_name: str | None = None,
_cli_parse_args: bool
| list[str]
| tuple[str, ...]
| None = None,
_cli_settings_source: CliSettingsSource[Any]
| None = None,
_cli_parse_none_str: str | None = None,
_cli_hide_none_type: bool | None = None,
_cli_avoid_json: bool | None = None,
_cli_enforce_required: bool | None = None,
_cli_use_class_docs_for_groups: bool | None = None,
_cli_exit_on_error: bool | None = None,
_cli_prefix: str | None = None,
_cli_flag_prefix_char: str | None = None,
_cli_implicit_flags: bool | None = None,
_cli_ignore_unknown_args: bool | None = None,
_cli_kebab_case: bool
| Literal["all", "no_enums"]
| None = None,
_cli_shortcuts: Mapping[str, str | list[str]]
| None = None,
_secrets_dir: PathType | None = None,
**values: Any,
)
Bases: BaseSettings
Configuration options for metadata stores.
Show JSON schema:
{
"additionalProperties": false,
"description": "Configuration options for metadata stores.",
"properties": {
"type": {
"description": "Full import path to metadata store class (e.g., `\"metaxy.ext.metadata_stores.duckdb.DuckDBMetadataStore\"`)",
"title": "Type",
"type": "string"
},
"config": {
"additionalProperties": true,
"description": "Store-specific configuration parameters (constructor kwargs). Includes `fallback_stores`, database connection parameters, etc.",
"title": "Config",
"type": "object"
}
},
"required": [
"type"
],
"title": "StoreConfig",
"type": "object"
}
Config:
extra:forbidfrozen:True
Source code in .venv/lib/python3.10/site-packages/pydantic_settings/main.py
def __init__(
__pydantic_self__,
_case_sensitive: bool | None = None,
_nested_model_default_partial_update: bool | None = None,
_env_prefix: str | None = None,
_env_file: DotenvType | None = ENV_FILE_SENTINEL,
_env_file_encoding: str | None = None,
_env_ignore_empty: bool | None = None,
_env_nested_delimiter: str | None = None,
_env_nested_max_split: int | None = None,
_env_parse_none_str: str | None = None,
_env_parse_enums: bool | None = None,
_cli_prog_name: str | None = None,
_cli_parse_args: bool | list[str] | tuple[str, ...] | None = None,
_cli_settings_source: CliSettingsSource[Any] | None = None,
_cli_parse_none_str: str | None = None,
_cli_hide_none_type: bool | None = None,
_cli_avoid_json: bool | None = None,
_cli_enforce_required: bool | None = None,
_cli_use_class_docs_for_groups: bool | None = None,
_cli_exit_on_error: bool | None = None,
_cli_prefix: str | None = None,
_cli_flag_prefix_char: str | None = None,
_cli_implicit_flags: bool | None = None,
_cli_ignore_unknown_args: bool | None = None,
_cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None,
_cli_shortcuts: Mapping[str, str | list[str]] | None = None,
_secrets_dir: PathType | None = None,
**values: Any,
) -> None:
super().__init__(
**__pydantic_self__._settings_build_values(
values,
_case_sensitive=_case_sensitive,
_nested_model_default_partial_update=_nested_model_default_partial_update,
_env_prefix=_env_prefix,
_env_file=_env_file,
_env_file_encoding=_env_file_encoding,
_env_ignore_empty=_env_ignore_empty,
_env_nested_delimiter=_env_nested_delimiter,
_env_nested_max_split=_env_nested_max_split,
_env_parse_none_str=_env_parse_none_str,
_env_parse_enums=_env_parse_enums,
_cli_prog_name=_cli_prog_name,
_cli_parse_args=_cli_parse_args,
_cli_settings_source=_cli_settings_source,
_cli_parse_none_str=_cli_parse_none_str,
_cli_hide_none_type=_cli_hide_none_type,
_cli_avoid_json=_cli_avoid_json,
_cli_enforce_required=_cli_enforce_required,
_cli_use_class_docs_for_groups=_cli_use_class_docs_for_groups,
_cli_exit_on_error=_cli_exit_on_error,
_cli_prefix=_cli_prefix,
_cli_flag_prefix_char=_cli_flag_prefix_char,
_cli_implicit_flags=_cli_implicit_flags,
_cli_ignore_unknown_args=_cli_ignore_unknown_args,
_cli_kebab_case=_cli_kebab_case,
_cli_shortcuts=_cli_shortcuts,
_secrets_dir=_secrets_dir,
)
)
Attributes¶
metaxy.StoreConfig.type
pydantic-field
¶
type: str
Full import path to metadata store class (e.g., "metaxy.ext.metadata_stores.duckdb.DuckDBMetadataStore")
metaxy.StoreConfig.config
pydantic-field
¶
Store-specific configuration parameters (constructor kwargs). Includes fallback_stores, database connection parameters, etc.
metaxy.StoreConfig.type_cls
cached
property
¶
Get the store class, importing lazily on first access.
Returns:
-
MetadataStoreTβThe metadata store class
Raises:
-
ImportErrorβIf the store class cannot be imported
Functions¶
metaxy.config.InvalidConfigError
¶
Bases: Exception
Raised when Metaxy configuration is invalid.
This error includes helpful context about where the configuration was loaded from and how environment variables can affect configuration.
Source code in src/metaxy/config.py
def __init__(
self,
message: str,
*,
config_file: Path | None = None,
):
self.config_file = config_file
self.base_message = message
# Build the full error message with context
parts = [message]
if config_file:
parts.append(f"Config file: {config_file}")
parts.append("Note: METAXY_* environment variables can override config file settings ")
super().__init__("\n".join(parts))
Functions¶
metaxy.config.InvalidConfigError.from_config
classmethod
¶
from_config(
config: MetaxyConfig, message: str
) -> InvalidConfigError
Create an InvalidConfigError from a MetaxyConfig instance.
Parameters:
-
config(MetaxyConfig) βThe MetaxyConfig instance that has the invalid configuration.
-
message(str) βThe error message describing what's wrong.
Returns:
-
InvalidConfigErrorβAn InvalidConfigError with context from the config.
Source code in src/metaxy/config.py
@classmethod
def from_config(cls, config: "MetaxyConfig", message: str) -> "InvalidConfigError":
"""Create an InvalidConfigError from a MetaxyConfig instance.
Args:
config: The MetaxyConfig instance that has the invalid configuration.
message: The error message describing what's wrong.
Returns:
An InvalidConfigError with context from the config.
"""
return cls(message, config_file=config._config_file)