config

Project configuration loaded from project.yaml.

AzureStateConfig

Bases: BaseModel

Azure Blob Storage connection settings.

Parameters:

Name	Type	Description	Default
account_name		Azure Storage Account name.	required
container_name		Name of the blob container to use.	required

Source code in src/cloe_delta_table_manager/cli/config.py

class AzureStateConfig(BaseModel):
    """Azure Blob Storage connection settings.

    Args:
        account_name: Azure Storage Account name.
        container_name: Name of the blob container to use.

    """

    model_config = ConfigDict(extra="forbid")

    account_name: str
    container_name: str

ProjectConfig

Bases: BaseModel

Project-level configuration resolved from project.yaml.

Parameters:

Name	Type	Description	Default
state_backend		Storage backend for state files. One of local or azure.	required
azure		Azure Blob Storage settings. Required when state_backend is azure.	required
log_level		Python logging level. Defaults to INFO.	required
target_platform		Deployment platform. Currently only azure_databricks.	required
targets		Named deployment targets. Keys are arbitrary target names (e.g. dev, prod); values hold the per-target settings.	required
migrations		Migration paths and state file locations. Required.	required
platform_options		Platform-specific render-time options for the target_platform. For Azure Databricks this currently exposes table_properties (applied to new tables on creation only).	required

Source code in src/cloe_delta_table_manager/cli/config.py

class ProjectConfig(BaseModel):
    """Project-level configuration resolved from ``project.yaml``.

    Args:
        state_backend: Storage backend for state files. One of ``local`` or ``azure``.
        azure: Azure Blob Storage settings. Required when *state_backend* is ``azure``.
        log_level: Python logging level. Defaults to ``INFO``.
        target_platform: Deployment platform. Currently only ``azure_databricks``.
        targets: Named deployment targets. Keys are arbitrary target names (e.g.
            ``dev``, ``prod``); values hold the per-target settings.
        migrations: Migration paths and state file locations. Required.
        platform_options: Platform-specific render-time options for the
            ``target_platform``. For Azure Databricks this currently exposes
            ``table_properties`` (applied to new tables on creation only).

    """

    model_config = ConfigDict(extra="forbid")

    state_backend: Literal["local", "azure"] = "local"
    azure: AzureStateConfig | None = None
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    target_platform: Literal["azure_databricks"]
    targets: dict[str, TargetConfig] = {}
    migrations: MigrationOrchestratorConfig
    platform_options: AzureDatabricksOptions = Field(default_factory=AzureDatabricksOptions)

    # Holds a TemporaryDirectory when rendering to a temp location so it
    # stays alive for the lifetime of this config object and is cleaned up
    # automatically when the object is garbage-collected.
    _render_temp_dir: tempfile.TemporaryDirectory | None = PrivateAttr(default=None)

    @model_validator(mode="after")
    def _require_azure_config(self) -> ProjectConfig:
        if self.state_backend == "azure" and self.azure is None:
            raise ValueError("'azure' config block is required when state_backend is 'azure'")
        return self

    @classmethod
    def from_yaml(cls, path: str | Path = "project.yaml") -> ProjectConfig:
        """Load and validate configuration from a YAML file.

        Relative paths in the ``migrations`` block are resolved relative to
        the directory containing the YAML file, not the current working directory.

        Args:
            path: Path to the YAML configuration file.

        Returns:
            A validated ProjectConfig instance.

        Raises:
            FileNotFoundError: If the file does not exist.

        """
        file_path = Path(path).resolve()
        if not file_path.exists():
            raise FileNotFoundError(f"Config file not found: {file_path}")
        with file_path.open() as f:
            data = yaml.safe_load(f) or {}
        base = file_path.parent
        _path_fields = (
            "database_sql_folder",
            "migration_root_path",
        )
        migrations = data.get("migrations", {})
        for field in _path_fields:
            if field in migrations:
                migrations[field] = str((base / migrations[field]).resolve())
        for target_data in data.get("targets", {}).values():
            if isinstance(target_data, dict) and "state_folder" in target_data:
                target_data["state_folder"] = str((base / target_data["state_folder"]).resolve())
        return cls.model_validate(data)

    def build_state_client(self) -> StateClient:
        """Return the configured state client.

        Returns:
            A StateClient based on *state_backend*.

        """
        if self.state_backend == "azure":
            from cloe_delta_table_manager.state.clients.azure import AzureBlobStateClient  # noqa: PLC0415

            assert self.azure is not None
            return AzureBlobStateClient(
                account_name=self.azure.account_name,
                container_name=self.azure.container_name,
            )
        return LocalStateClient()

    def prepare_for_target(self, target_name: str, render_output_path: str | None = None) -> ProjectConfig:
        """Render template files for the given target and return an updated config.

        Renders ``database_sql_folder`` and ``migration_root_path`` using the
        target's variables and database name mappings. When *render_output_path*
        is provided the rendered files are written there (persisted between runs,
        useful for debugging). When it is ``None`` a temporary directory is used
        and cleaned up automatically when the returned config object is
        garbage-collected.

        If *target_name* is unknown, returns *self* unchanged.

        Args:
            target_name: Key into ``targets`` identifying the deployment target.
            render_output_path: Optional directory to write rendered files into.

        Returns:
            A ``ProjectConfig`` with updated migration paths pointing to the
            rendered output, or *self* if the target is not found.

        """
        if target_name not in self.targets:
            return self

        target = self.targets[target_name]
        variables = {**target.databases, **target.variables}

        temp_dir: tempfile.TemporaryDirectory | None = None
        if render_output_path is None:
            temp_dir = tempfile.TemporaryDirectory(prefix="dtm-")
            output_root = Path(temp_dir.name)
        else:
            output_root = Path(render_output_path)

        sql_output = output_root / Path(self.migrations.database_sql_folder).name
        migrations_output = output_root / Path(self.migrations.migration_root_path).name

        render_folder(source=self.migrations.database_sql_folder, params=variables, output=sql_output)
        render_folder(source=self.migrations.migration_root_path, params=variables, output=migrations_output)

        state_updates: dict[str, str] = {}
        if target.state_folder is not None:
            state_path = Path(target.state_folder)
            state_updates = {
                "database_statefile_path": str(state_path / "database_state.json"),
                "solution_statefile_path": str(state_path / "solution_state.json"),
                "pre_deployment_migration_state_path": str(state_path / "pre_deployment_state.json"),
                "post_deployment_migration_state_path": str(state_path / "post_deployment_state.json"),
                "auto_migration_state_path": str(state_path / "auto_migration_state.json"),
            }

        updated_migrations = self.migrations.model_copy(
            update={
                "database_sql_folder": str(sql_output),
                "migration_root_path": str(migrations_output),
                **state_updates,
            }
        )
        result = self.model_copy(update={"migrations": updated_migrations})
        result._render_temp_dir = temp_dir
        return result

    def build_state(self, keyword: StateKeyword, target_name: str | None = None) -> DatabaseState:
        """Return a DatabaseState for the given keyword.

        Args:
            keyword: One of ``solution``, ``statefile``, or ``database``.
            target_name: Name of the target to connect to. Required when
                *keyword* is ``database``.

        Returns:
            The corresponding ``DatabaseState``.

        Raises:
            NotImplementedError: If ``keyword`` is ``database`` (not yet implemented).

        """
        from cloe_delta_table_manager.state.state import DatabaseState  # noqa: PLC0415

        if keyword == "solution":
            from cloe_delta_table_manager.sql_parser.process import process_sql  # noqa: PLC0415

            mapping = process_sql(self.migrations.database_sql_folder, recursive=True)
            return DatabaseState(databases=mapping.to_databases().databases)
        if keyword == "statefile":
            return self.build_state_client().read_state(self.migrations.database_statefile_path, DatabaseState)
        if keyword == "database":
            from cloe_dbx_crawler.crawler import DatabricksCrawler  # noqa: PLC0415

            connector = self.build_databricks_connector(target_name)
            assert target_name is not None
            target = self.targets[target_name]
            catalog_names = list(target.databases.values())
            catalog_filter = "^(" + "|".join(re.escape(name) for name in catalog_names) + ")$"

            crawler = DatabricksCrawler(
                client=connector.get_workspace_client(),
                ignore_tables=False,
                ignore_columns=False,
                catalog_filter=catalog_filter,
                schema_filter="^(?!default$|information_schema$).*$",
            )
            crawler.crawl(write_to_disk=False)
            return DatabaseState(databases=crawler.databases.databases)

        raise NotImplementedError(
            f"The keyword {keyword} is not yet supported. Please choose from 'solution' or 'statefile'."
        )

    def refresh_database_state(self, target_name: str) -> DatabaseState:
        """Crawl the live database and persist the result as the database statefile.

        Args:
            target_name: Key into ``targets`` identifying the deployment target.

        Returns:
            The crawled ``DatabaseState``.

        """
        if not self.migrations.database_statefile_path:
            raise ValueError(
                "No database_statefile_path configured. Add a 'state_folder' entry under the target in project.yaml."
            )
        state = self.build_state("database", target_name)
        self.build_state_client().write_state(self.migrations.database_statefile_path, state)
        return state

    def build_databricks_connector(self, target_name: str | None) -> DatabricksConnector:
        """Return a DatabricksConnector for the given target.

        Args:
            target_name: Key into ``targets`` identifying the deployment target.

        Returns:
            A DatabricksConnector instance.

        Raises:
            ValueError: If *target_name* is not found in ``targets``.
            NotImplementedError: If the target platform is not supported.

        """
        if target_name is None or target_name not in self.targets:
            available = list(self.targets)
            raise ValueError(f"Target '{target_name}' not found. Available targets: {available}")
        workspace_url = self.targets[target_name].workspace_url
        if self.target_platform == "azure_databricks":
            from cloe_dbx_connector import AzureDatabricksConfig, DatabricksConnector  # noqa: PLC0415

            return DatabricksConnector(AzureDatabricksConfig(workspace_url=workspace_url))  # ty: ignore[missing-argument]
        raise NotImplementedError(f"Target platform {self.target_platform} is not supported yet.")

build_databricks_connector(target_name)

Return a DatabricksConnector for the given target.

Parameters:

Name	Type	Description	Default
target_name	str | None	Key into targets identifying the deployment target.	required

Returns:

Type	Description
DatabricksConnector	A DatabricksConnector instance.

Raises:

Type	Description
ValueError	If target_name is not found in targets.
NotImplementedError	If the target platform is not supported.

Source code in src/cloe_delta_table_manager/cli/config.py

def build_databricks_connector(self, target_name: str | None) -> DatabricksConnector:
    """Return a DatabricksConnector for the given target.

    Args:
        target_name: Key into ``targets`` identifying the deployment target.

    Returns:
        A DatabricksConnector instance.

    Raises:
        ValueError: If *target_name* is not found in ``targets``.
        NotImplementedError: If the target platform is not supported.

    """
    if target_name is None or target_name not in self.targets:
        available = list(self.targets)
        raise ValueError(f"Target '{target_name}' not found. Available targets: {available}")
    workspace_url = self.targets[target_name].workspace_url
    if self.target_platform == "azure_databricks":
        from cloe_dbx_connector import AzureDatabricksConfig, DatabricksConnector  # noqa: PLC0415

        return DatabricksConnector(AzureDatabricksConfig(workspace_url=workspace_url))  # ty: ignore[missing-argument]
    raise NotImplementedError(f"Target platform {self.target_platform} is not supported yet.")

build_state(keyword, target_name=None)

Return a DatabaseState for the given keyword.

Parameters:

Name	Type	Description	Default
keyword	StateKeyword	One of solution, statefile, or database.	required
target_name	str | None	Name of the target to connect to. Required when keyword is database.	None

Returns:

Type	Description
DatabaseState	The corresponding DatabaseState.

Raises:

Type	Description
NotImplementedError	If keyword is database (not yet implemented).

Source code in src/cloe_delta_table_manager/cli/config.py

def build_state(self, keyword: StateKeyword, target_name: str | None = None) -> DatabaseState:
    """Return a DatabaseState for the given keyword.

    Args:
        keyword: One of ``solution``, ``statefile``, or ``database``.
        target_name: Name of the target to connect to. Required when
            *keyword* is ``database``.

    Returns:
        The corresponding ``DatabaseState``.

    Raises:
        NotImplementedError: If ``keyword`` is ``database`` (not yet implemented).

    """
    from cloe_delta_table_manager.state.state import DatabaseState  # noqa: PLC0415

    if keyword == "solution":
        from cloe_delta_table_manager.sql_parser.process import process_sql  # noqa: PLC0415

        mapping = process_sql(self.migrations.database_sql_folder, recursive=True)
        return DatabaseState(databases=mapping.to_databases().databases)
    if keyword == "statefile":
        return self.build_state_client().read_state(self.migrations.database_statefile_path, DatabaseState)
    if keyword == "database":
        from cloe_dbx_crawler.crawler import DatabricksCrawler  # noqa: PLC0415

        connector = self.build_databricks_connector(target_name)
        assert target_name is not None
        target = self.targets[target_name]
        catalog_names = list(target.databases.values())
        catalog_filter = "^(" + "|".join(re.escape(name) for name in catalog_names) + ")$"

        crawler = DatabricksCrawler(
            client=connector.get_workspace_client(),
            ignore_tables=False,
            ignore_columns=False,
            catalog_filter=catalog_filter,
            schema_filter="^(?!default$|information_schema$).*$",
        )
        crawler.crawl(write_to_disk=False)
        return DatabaseState(databases=crawler.databases.databases)

    raise NotImplementedError(
        f"The keyword {keyword} is not yet supported. Please choose from 'solution' or 'statefile'."
    )

build_state_client()

Return the configured state client.

Returns:

Type	Description
StateClient	A StateClient based on state_backend.

Source code in src/cloe_delta_table_manager/cli/config.py

def build_state_client(self) -> StateClient:
    """Return the configured state client.

    Returns:
        A StateClient based on *state_backend*.

    """
    if self.state_backend == "azure":
        from cloe_delta_table_manager.state.clients.azure import AzureBlobStateClient  # noqa: PLC0415

        assert self.azure is not None
        return AzureBlobStateClient(
            account_name=self.azure.account_name,
            container_name=self.azure.container_name,
        )
    return LocalStateClient()

from_yaml(path='project.yaml') classmethod

Load and validate configuration from a YAML file.

Relative paths in the migrations block are resolved relative to the directory containing the YAML file, not the current working directory.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the YAML configuration file.	'project.yaml'

Returns:

Type	Description
ProjectConfig	A validated ProjectConfig instance.

Raises:

Type	Description
FileNotFoundError	If the file does not exist.

Source code in src/cloe_delta_table_manager/cli/config.py

@classmethod
def from_yaml(cls, path: str | Path = "project.yaml") -> ProjectConfig:
    """Load and validate configuration from a YAML file.

    Relative paths in the ``migrations`` block are resolved relative to
    the directory containing the YAML file, not the current working directory.

    Args:
        path: Path to the YAML configuration file.

    Returns:
        A validated ProjectConfig instance.

    Raises:
        FileNotFoundError: If the file does not exist.

    """
    file_path = Path(path).resolve()
    if not file_path.exists():
        raise FileNotFoundError(f"Config file not found: {file_path}")
    with file_path.open() as f:
        data = yaml.safe_load(f) or {}
    base = file_path.parent
    _path_fields = (
        "database_sql_folder",
        "migration_root_path",
    )
    migrations = data.get("migrations", {})
    for field in _path_fields:
        if field in migrations:
            migrations[field] = str((base / migrations[field]).resolve())
    for target_data in data.get("targets", {}).values():
        if isinstance(target_data, dict) and "state_folder" in target_data:
            target_data["state_folder"] = str((base / target_data["state_folder"]).resolve())
    return cls.model_validate(data)

prepare_for_target(target_name, render_output_path=None)

Render template files for the given target and return an updated config.

Renders database_sql_folder and migration_root_path using the target's variables and database name mappings. When render_output_path is provided the rendered files are written there (persisted between runs, useful for debugging). When it is None a temporary directory is used and cleaned up automatically when the returned config object is garbage-collected.

If target_name is unknown, returns self unchanged.

Parameters:

Name	Type	Description	Default
target_name	str	Key into targets identifying the deployment target.	required
render_output_path	str | None	Optional directory to write rendered files into.	None

Returns:

Type	Description
ProjectConfig	A ProjectConfig with updated migration paths pointing to the
ProjectConfig	rendered output, or self if the target is not found.

Source code in src/cloe_delta_table_manager/cli/config.py

def prepare_for_target(self, target_name: str, render_output_path: str | None = None) -> ProjectConfig:
    """Render template files for the given target and return an updated config.

    Renders ``database_sql_folder`` and ``migration_root_path`` using the
    target's variables and database name mappings. When *render_output_path*
    is provided the rendered files are written there (persisted between runs,
    useful for debugging). When it is ``None`` a temporary directory is used
    and cleaned up automatically when the returned config object is
    garbage-collected.

    If *target_name* is unknown, returns *self* unchanged.

    Args:
        target_name: Key into ``targets`` identifying the deployment target.
        render_output_path: Optional directory to write rendered files into.

    Returns:
        A ``ProjectConfig`` with updated migration paths pointing to the
        rendered output, or *self* if the target is not found.

    """
    if target_name not in self.targets:
        return self

    target = self.targets[target_name]
    variables = {**target.databases, **target.variables}

    temp_dir: tempfile.TemporaryDirectory | None = None
    if render_output_path is None:
        temp_dir = tempfile.TemporaryDirectory(prefix="dtm-")
        output_root = Path(temp_dir.name)
    else:
        output_root = Path(render_output_path)

    sql_output = output_root / Path(self.migrations.database_sql_folder).name
    migrations_output = output_root / Path(self.migrations.migration_root_path).name

    render_folder(source=self.migrations.database_sql_folder, params=variables, output=sql_output)
    render_folder(source=self.migrations.migration_root_path, params=variables, output=migrations_output)

    state_updates: dict[str, str] = {}
    if target.state_folder is not None:
        state_path = Path(target.state_folder)
        state_updates = {
            "database_statefile_path": str(state_path / "database_state.json"),
            "solution_statefile_path": str(state_path / "solution_state.json"),
            "pre_deployment_migration_state_path": str(state_path / "pre_deployment_state.json"),
            "post_deployment_migration_state_path": str(state_path / "post_deployment_state.json"),
            "auto_migration_state_path": str(state_path / "auto_migration_state.json"),
        }

    updated_migrations = self.migrations.model_copy(
        update={
            "database_sql_folder": str(sql_output),
            "migration_root_path": str(migrations_output),
            **state_updates,
        }
    )
    result = self.model_copy(update={"migrations": updated_migrations})
    result._render_temp_dir = temp_dir
    return result

refresh_database_state(target_name)

Crawl the live database and persist the result as the database statefile.

Parameters:

Name	Type	Description	Default
target_name	str	Key into targets identifying the deployment target.	required

Returns:

Type	Description
DatabaseState	The crawled DatabaseState.

Source code in src/cloe_delta_table_manager/cli/config.py

def refresh_database_state(self, target_name: str) -> DatabaseState:
    """Crawl the live database and persist the result as the database statefile.

    Args:
        target_name: Key into ``targets`` identifying the deployment target.

    Returns:
        The crawled ``DatabaseState``.

    """
    if not self.migrations.database_statefile_path:
        raise ValueError(
            "No database_statefile_path configured. Add a 'state_folder' entry under the target in project.yaml."
        )
    state = self.build_state("database", target_name)
    self.build_state_client().write_state(self.migrations.database_statefile_path, state)
    return state

TargetConfig

Bases: BaseModel

Configuration for a single named deployment target (e.g. dev, prod).

Parameters:

Name	Type	Description	Default
workspace_url		Databricks workspace URL for this target.	required
databases		Mapping from logical database names (as used in SQL files) to their environment-specific catalog/database names.	required
variables		Arbitrary key-value pairs available as substitution variables during deployment for this target.	required

Source code in src/cloe_delta_table_manager/cli/config.py

class TargetConfig(BaseModel):
    """Configuration for a single named deployment target (e.g. dev, prod).

    Args:
        workspace_url: Databricks workspace URL for this target.
        databases: Mapping from logical database names (as used in SQL files) to
            their environment-specific catalog/database names.
        variables: Arbitrary key-value pairs available as substitution variables
            during deployment for this target.

    """

    model_config = ConfigDict(extra="forbid")

    workspace_url: str
    state_folder: str | None = None
    databases: dict[str, str] = {}
    variables: dict[str, str] = {}