Source¶

@click.command(name="bundle")
@option.default_key
@option.overwrite
@option.resource_generator
@option.env
@option.conf_source
@option.pipeline
@option.params
@click.pass_obj
def command(
    metadata: ProjectMetadata,
    default_key: str,
    overwrite: bool,
    resource_generator: str,
    env: str,
    conf_source: str,
    pipeline: str | None,
    params: str | None,
):
    """Databricks Asset Bundle commands"""
    # If the configuration directory does not exist, Kedro will not load any configuration
    local_config_dir = metadata.project_path / conf_source / env
    if not local_config_dir.exists():
        log.warning(f"Creating {local_config_dir.relative_to(metadata.project_path)}")
        local_config_dir.mkdir(parents=True)

    with KedroSession.create(project_path=metadata.project_path, env=env) as session:
        overrides = _load_kedro_env_config(session=session)
        if not overrides:
            raise NoOverridesError(Path(conf_source))
        elif "resources" not in overrides:
            raise NoResourcesKeyError(env)

        ResourceGenerator = RESOURCE_GENERATOR_RESOLVER.resolve(resource_generator)

        g = ResourceGenerator(
            session=session,
            metadata=metadata,
            conf_source=conf_source,
            params=params,
        )

        all_resources = {"jobs": g.generate_jobs(pipeline)}
        overridden_resources = {}
        for resource_type, resource_override_items in overrides["resources"].items():
            overridden_resources[resource_type] = {}
            resource_items = all_resources.get(resource_type, {})
            overrider = RESOURCE_OVERRIDER_RESOLVER.resolve(resource_type)()
            all_keys = set(resource_items.keys()).union(
                set(resource_override_items.keys())
            )
            for key in all_keys:
                if key == default_key or key.startswith(config.regex_prefix):
                    continue
                resource = resource_items.get(key, {})
                overridden_resources[resource_type][key] = overrider.override(
                    resource_key=key,
                    resource=resource,
                    overrides=copy.deepcopy(resource_override_items),
                    default_key=default_key,
                )

        _save_resources(
            metadata=metadata,
            env=env,
            resources=overridden_resources,
            overwrite=overwrite,
        )

@click.command(name="deploy")
@option.env
@option.bundle
@option.default_key
@option.resource_generator
@option.conf_source
@option.pipeline
@option.params
@option.databricks_args
@option.runtime_params
@click.pass_context
def command(
    ctx: click.Context,
    env: str,
    bundle: bool,
    default_key: str,
    resource_generator: str,
    conf_source: str,
    pipeline: str | None,
    params: str | None,
    runtime_params: str | None,
    databricks_args: tuple[str, ...],
):
    """Deploy the Databricks Asset Bundle.

    This function deploys the Databricks Asset Bundle in the current project
    directory. It also creates a Databricks configuration file and a
    Databricks target configuration file.
    """
    if runtime_params:
        warnings.warn(
            "'--runtime-params' has been renamed to '--params' to be consistent with `kedro databricks bundle`"
        )
        params = runtime_params
    metadata = ctx.obj
    if not isinstance(metadata, ProjectMetadata):
        raise TypeError("Project metadata is not available in the context.")
    if bundle:
        ctx.invoke(
            bundle_command,
            env=env,
            default_key=default_key,
            conf_source=conf_source,
            resource_generator=resource_generator,
            pipeline=pipeline,
            params=params,
            overwrite=True,
        )
    dbcli = DatabricksCli(metadata, env=env, additional_args=list(databricks_args))
    dbcli.deploy()
    log.info(f"Deployed Databricks Asset Bundle in {metadata.project_path}")
    dbcli.upload()
    log.info(f"Uploaded project data to Databricks from {metadata.project_path}")
    dbcli.summary()

@click.command(name="destroy")
@option.env
@option.databricks_args
@click.pass_obj
def command(metadata: ProjectMetadata, env: str, databricks_args: tuple[str, ...]):
    """Databricks Asset Bundle Destroy commands"""
    dbcli = DatabricksCli(
        metadata=metadata,
        env=env,
        additional_args=list(databricks_args),
    )
    dbcli.destroy()

@click.command(name="init")
@option.catalog
@option.schema
@option.default_key
@option.env
@option.conf_source
@option.resource_generator
@option.regex_prefix
@option.overwrite
@option.databricks_args
@click.pass_obj
def command(
    metadata: ProjectMetadata,
    catalog: str,
    schema: str,
    default_key: str,
    env: str,
    conf_source: str,
    resource_generator: str,
    regex_prefix: str,
    overwrite: bool,
    databricks_args: tuple[str, ...],
):
    """Initialize a Kedro project for Databricks Asset Bundles."""
    log.info("Initializing Databricks Asset Bundle...")
    dbcli = DatabricksCli(metadata, additional_args=list(databricks_args))
    assets_dir, template_params = _prepare_template(metadata)
    config_path = metadata.project_path / "databricks.yml"
    if config_path.exists() and not overwrite:
        raise FileExistsError(f"`databricks.yml` already exist at {config_path}")
    elif config_path.exists() and overwrite:
        config_path.unlink(missing_ok=True)
    validated_conf = dbcli.init(assets_dir, template_params)
    log.info(f"Initialized Databricks Asset Bundle in {metadata.project_path}")
    _create_target_configs(
        metadata,
        default_key=default_key,
        default_catalog=catalog,
        default_schema=schema,
        validated_conf=validated_conf,
    )
    _update_gitignore(metadata)
    _update_pyproject(
        metadata,
        Config(
            init_catalog=catalog,
            init_schema=schema,
            default_env=env,
            conf_source=conf_source,
            workflow_default_key=default_key,
            workflow_generator=resource_generator,
            regex_prefix=regex_prefix,
        ),
    )
    hooks_path = metadata.project_path / "src" / metadata.package_name / "hooks.py"
    if hooks_path.exists():
        _transform_spark_hook(hooks_path.as_posix())
    if require_databricks_run_script():  # pragma: no cover - Might be removed in future
        log.warning(
            "Kedro version less than 0.19.8 requires a script to run tasks on Databricks. "
        )
        _write_databricks_run_script(metadata)
    log.info(
        f"Successfully initialized Databricks Asset Bundle in {metadata.project_path}"
    )

@click.command(name="run")
@option.pipeline_arg
@option.env
@option.databricks_args
@click.pass_obj
def command(
    metadata: ProjectMetadata,
    env: str,
    pipeline: str,
    databricks_args: tuple[str, ...],
):
    """Databricks Asset Bundle Run commands"""
    # If the first argument starts with '--', it means no pipeline was provided
    # This is to handle the case where the user wants to pass only options to the CLI
    # e.g. `databricks bundle run -- --profile prod` will run the default pipeline with the prod profile
    if pipeline.startswith("--"):
        databricks_args = (pipeline,) + databricks_args
        pipeline = ""

    if not pipeline:
        pipeline = metadata.package_name

    dbcli = DatabricksCli(
        metadata=metadata,
        env=env,
        additional_args=list(databricks_args),
    )
    dbcli.run(pipeline)
    log.info(
        f"Successfully triggered Databricks job for pipeline '{pipeline}' in project {metadata.project_path}"
    )

def get_arg_value(args: list[str], arg_name: str) -> str | None:
    """Get the value of an argument from a list of arguments.

    Args:
        args (List[str]): list of arguments
        arg_name (str): name of the argument to get the value for

    Returns:
        str | None: value of the argument or None if not found
    """
    for i, arg in enumerate(args):
        if "=" in arg:
            _arg, value = arg.split("=", 1)
            if _arg == arg_name:
                return value
        elif arg == arg_name:
            return args[i + 1]

def get_entry_point(project_name: str) -> str:
    """Get the entry point for a project.

    Args:
        project_name (str): name of the project

    Returns:
        str: entry point for the project
    """
    entrypoint = project_name.strip().lower()
    entrypoint = re.sub(r" +", " ", entrypoint)
    entrypoint = re.sub(r"[^a-zA-Z_]", "-", entrypoint)
    entrypoint = re.sub(r"(-+)$", "", entrypoint)
    entrypoint = re.sub(r"^(-+)", "", entrypoint)
    return entrypoint

def get_regex_values(
    lookup_key: str,
    values: dict[str, Any],
    regex_prefix: str = "re:",
) -> dict[str, Any]:
    """Get values matching a regex

    Args:
        lookup_key (str): key to match regexes to
        values (dict[str, Any]): The overrides to apply.
        regex_prefix (str): prefix that identifies a key as a regex

    Returns:
        dict[str, Any]: values where lookup key matches the regex
    """
    matched_values = {}
    for key, value in values.items():
        if key.startswith(regex_prefix):
            pattern = key[len(regex_prefix) :]
            if re.match(pattern, lookup_key):
                return value
    return matched_values

def get_value_from_dotpath(ddict, dotpath) -> Any:
    """Get a value from a dictionary using a dotpath.

    Args:
        ddict (Any): dictionary to get the value from
        dotpath (str): dotpath to get the value for

    Returns:
        Any: value from the dictionary or None if not found
    """
    if not isinstance(ddict, dict):
        return None
    keys = dotpath.split(".")
    key = keys.pop(0)
    current_level = ddict.get(key)
    if current_level is None:
        return None
    elif len(keys) > 0:
        return get_value_from_dotpath(current_level, ".".join(keys))
    else:
        return current_level

def remove_nulls(value: dict[str, Any] | list[Any]) -> dict[str, Any] | list[Any]:
    """Remove None values from a dictionary or list.

    Args:
        value (Dict[Any, Any] | List[Dict[Any, Any]]): dictionary or list to remove None values from

    Returns:
        Dict[Any, Any] | List[Dict[Any, Any]]: dictionary or list with None values removed
    """
    non_null = copy.deepcopy(value)
    if isinstance(non_null, dict):
        _remove_nulls_from_dict(non_null)
    elif isinstance(non_null, list):
        _remove_nulls_from_list(non_null)
    return non_null

def require_databricks_run_script(_version=KEDRO_VERSION) -> bool:
    """Check if the current Kedro version is less than 0.19.8.

    Kedro 0.19.8 introduced a new `run_script` method that is required for
    running tasks on Databricks. This method is not available in earlier
    versions of Kedro. This function checks if the current Kedro version is
    less than 0.19.8.

    Returns:
        bool: whether the current Kedro version is less than 0.19.8
    """
    return _version < Version("0.19.8")