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 ¶

Bases: BaseSettings

Main Metaxy configuration.

Loads from (in order of precedence):

Init arguments
Environment variables (METAXY_*)
Config file (metaxy.toml or [tool.metaxy] in pyproject.toml )

Accessing current configuration

config = MetaxyConfig.load()

Getting a configured metadata store

store = config.get_store("prod")

The default store is "dev"; METAXY_STORE can be used to override it.

Attributes¶

metaxy.MetaxyConfig.plugins `property` ¶

plugins: list[str]

Returns all enabled plugin names from ext configuration.

Functions¶

metaxy.MetaxyConfig.validate_project `classmethod` ¶

validate_project(v: str) -> str

Validate project name follows naming rules.

Source code in src/metaxy/config.py

@field_validator("project")
@classmethod
def validate_project(cls, v: str) -> str:
    """Validate project name follows naming rules."""
    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_plugin(name: str, plugin_cls: type[PluginConfigT]) -> PluginConfigT

Get the plugin config from the global Metaxy config.

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."""
    ext = cls.get().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.validate_hash_truncation_length `classmethod` ¶

validate_hash_truncation_length(v: int | None) -> int | None

Validate hash truncation length is at least 8 if set.

Source code in src/metaxy/config.py

@field_validator("hash_truncation_length")
@classmethod
def validate_hash_truncation_length(cls, v: int | None) -> int | None:
    """Validate hash truncation length is at least 8 if set."""
    if v is not None and v < 8:
        raise ValueError(
            f"hash_truncation_length must be at least 8 characters, got {v}"
        )
    return v

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() -> MetaxyConfig

Get the current Metaxy configuration.

Source code in src/metaxy/config.py

@classmethod
def get(cls) -> "MetaxyConfig":
    """Get the current Metaxy configuration."""
    cfg = _metaxy_config.get()
    if cfg is None:
        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, project='default')."
            )
        )
        return cls(project="default")
    else:
        return cfg

metaxy.MetaxyConfig.set `classmethod` ¶

set(config: Self | None) -> None

Set the current Metaxy configuration.

Source code in src/metaxy/config.py

@classmethod
def set(cls, config: Self | None) -> None:
    """Set the current Metaxy configuration."""
    _metaxy_config.set(config)

metaxy.MetaxyConfig.is_set `classmethod` ¶

is_set() -> bool

Check if the current Metaxy configuration is set.

Source code in src/metaxy/config.py

@classmethod
def is_set(cls) -> bool:
    """Check if the current Metaxy configuration is set."""
    return _metaxy_config.get() is not None

metaxy.MetaxyConfig.reset `classmethod` ¶

reset() -> None

Reset the current Metaxy configuration to None.

Source code in src/metaxy/config.py

@classmethod
def reset(cls) -> None:
    """Reset the current Metaxy configuration to None."""
    _metaxy_config.set(None)

metaxy.MetaxyConfig.use ¶

use() -> Iterator[Self]

Use this configuration temporarily, restoring previous config on exit.

Example

config = MetaxyConfig(project="test")
with config.use():
    # Code here uses test config
    assert MetaxyConfig.get().project == "test"
# Previous config restored

Source code in src/metaxy/config.py

@contextmanager
def use(self) -> Iterator[Self]:
    """Use this configuration temporarily, restoring previous config on exit.

    Example:
        ```py
        config = MetaxyConfig(project="test")
        with config.use():
            # Code here uses test config
            assert MetaxyConfig.get().project == "test"
        # Previous config restored
        ```
    """
    previous = _metaxy_config.get()
    _metaxy_config.set(self)
    try:
        yield self
    finally:
        _metaxy_config.set(previous)

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_CONFIG environment 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:
        ```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  # type: ignore[misc]
        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  # type: ignore[assignment]
        config = cls()
        cls.settings_customise_sources = original_method  # type: ignore[method-assign]
    else:
        # Use default sources (auto-discovery + env vars)
        config = cls()

    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

get_store(name: str | None = None, *, expected_type: type[StoreTypeT], **kwargs: Any) -> StoreTypeT

get_store(name: str | None = None, *, expected_type: type[StoreTypeT] | None = None, **kwargs: Any) -> MetadataStore | StoreTypeT

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 TypeError is 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

Example

config = MetaxyConfig.load()
store = config.get_store("prod")

# Use default store
store = config.get_store()

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
        config = MetaxyConfig.load()
        store = config.get_store("prod")

        # Use default store
        store = config.get_store()
        ```
    """
    from metaxy.versioning.types import HashAlgorithm

    if len(self.stores) == 0:
        raise ValueError(
            "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 ValueError(
            f"Store '{name}' not found in config. "
            f"Available stores: {list(self.stores.keys())}"
        )

    store_config = self.stores[name]

    # Get store class (already imported by Pydantic's ImportString)
    store_class = store_config.type

    if expected_type is not None and not issubclass(store_class, expected_type):
        raise TypeError(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

    typed_config = config_model_cls.model_validate(config_copy)

    # Instantiate using from_config() - fallback stores are resolved via MetaxyConfig.get()
    # Use self.use() to ensure this config is available for fallback resolution
    with self.use():
        store = store_class.from_config(typed_config, **extra_kwargs)

    # 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 ValueError(
            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 TypeError(f"Store '{name}' is not of type '{expected_type.__name__}'")

    return store

metaxy.StoreConfig ¶

Bases: BaseSettings

Configuration for a single metadata store.

Example

config = StoreConfig(
    type="metaxy_delta.DeltaMetadataStore",
    config={
        "root_path": "s3://bucket/metadata",
        "region": "us-west-2",
        "fallback_stores": ["prod"],
    }
)

Configuration¶

metaxy.MetaxyConfig ¶

Attributes¶

metaxy.MetaxyConfig.plugins property ¶

Functions¶

metaxy.MetaxyConfig.validate_project classmethod ¶

metaxy.MetaxyConfig.get_plugin classmethod ¶

metaxy.MetaxyConfig.validate_hash_truncation_length classmethod ¶

metaxy.MetaxyConfig.settings_customise_sources classmethod ¶

metaxy.MetaxyConfig.get classmethod ¶

metaxy.MetaxyConfig.set classmethod ¶

metaxy.MetaxyConfig.is_set classmethod ¶

metaxy.MetaxyConfig.reset classmethod ¶

metaxy.MetaxyConfig.use ¶

metaxy.MetaxyConfig.load classmethod ¶

metaxy.MetaxyConfig.get_store ¶

metaxy.StoreConfig ¶

metaxy.MetaxyConfig.plugins `property` ¶

metaxy.MetaxyConfig.validate_project `classmethod` ¶

metaxy.MetaxyConfig.get_plugin `classmethod` ¶

metaxy.MetaxyConfig.validate_hash_truncation_length `classmethod` ¶

metaxy.MetaxyConfig.settings_customise_sources `classmethod` ¶

metaxy.MetaxyConfig.get `classmethod` ¶

metaxy.MetaxyConfig.set `classmethod` ¶

metaxy.MetaxyConfig.is_set `classmethod` ¶

metaxy.MetaxyConfig.reset `classmethod` ¶

metaxy.MetaxyConfig.load `classmethod` ¶