kedro.inspection¶

def get_project_snapshot(  # noqa: PLR0913
    project_path: str | Path | None = None,
    env: str | None = None,
    conf_source: str | None = None,
    metadata: ProjectMetadata | None = None,
    runtime_params: dict[str, Any] | None = None,
    group_by: Literal["namespace", "none"] | None = "namespace",
) -> ProjectSnapshot:
    """Return a read-only snapshot of the Kedro project.

    This is the primary entry point for the inspection API. It initialises
    the project, loads configuration and pipelines, and assembles a
    ``ProjectSnapshot`` without executing any pipeline nodes or writing data.

    At least one of *project_path* or *metadata* must be supplied.  When both
    are given and point to different directories, *metadata.project_path* takes
    precedence and a ``UserWarning`` is emitted.

    Args:
        project_path: Path to the project root directory (the directory that
            contains ``pyproject.toml``). Optional when *metadata* is provided.
        env: Optional run environment override (e.g. ``"staging"``). When
            ``None``, the project's default run environment is used.
        conf_source: Optional path to the configuration directory. When
            ``None``, defaults to ``<project_path>/<settings.CONF_SOURCE>``.
        metadata: Optional pre-computed ``ProjectMetadata`` returned by a prior
            ``bootstrap_project`` call. When provided, ``bootstrap_project`` is
            skipped. Pass this when calling ``get_project_snapshot`` repeatedly
            in the same process to avoid redundant initialisation.

            For example, in a long-running server the project is bootstrapped
            once at startup and the result is forwarded to every snapshot call::

                # --- server startup ---
                metadata = bootstrap_project(project_path)

                # --- per-request snapshot (bootstrap_project not called again) ---
                snapshot = get_project_snapshot(metadata=metadata)

            The same pattern applies to programmatic callers that need multiple
            snapshots across different environments::

                from kedro.framework.startup import bootstrap_project
                from kedro.inspection import get_project_snapshot

                metadata = bootstrap_project(project_path)
                snapshot_default = get_project_snapshot(metadata=metadata)
                snapshot_staging = get_project_snapshot(env="staging", metadata=metadata)

        runtime_params: Optional dictionary of runtime parameters forwarded to
            the config loader for ``${runtime_params:...}`` interpolation. This
            mirrors ``kedro run --params`` / ``KedroSession.create(runtime_params=...)``
            for configuration resolution::

                snapshot = get_project_snapshot(
                    metadata=metadata,
                    runtime_params={"version": "02"},
                )

        group_by: Strategy used to build the node groups in each pipeline's
            `groups`, passed to `Pipeline.group_nodes_by`. Accepted values are
            `"namespace"` (the default), which groups nodes by their top-level
            namespace, and `"none"` or `None`, which makes every node a group
            of its own. Matching is case-insensitive, and the value used is
            recorded in `PipelineSnapshot.group_by`.

    Returns:
        A fully populated ``ProjectSnapshot``.

    Raises:
        ValueError: If `group_by` is not a supported strategy. Raised before
            the project is bootstrapped or any configuration is loaded.
    """
    return _build_project_snapshot(
        project_path=project_path,
        env=env,
        conf_source=conf_source,
        metadata=metadata,
        runtime_params=runtime_params,
        group_by=group_by,
    )

def validate_deployment_grouping(
    pipeline: Pipeline,
    catalog: CatalogProtocol,
    groups: list[GroupedNodes] | None = None,
    group_by: str | None = "namespace",
) -> DeploymentGroupingResult:
    """Check that a node grouping can run with each group as a separate task.

    Deployment platforms such as Airflow or Databricks run each group from
    `Pipeline.group_nodes_by` as its own task, usually in its own container, so
    groups do not share memory or local disk. This function reports the
    problems that only appear once that happens:

    - `invalid_grouping` (error): only for `groups` passed in. A node is in no
      group, in more than one group or not in the pipeline, a group is empty, or
      a group name is repeated. The remaining checks are skipped.
    - `group_cycle` (error): groups depend on each other in a loop, usually
      because a namespace is interrupted by a node outside it.
    - `ephemeral_boundary` (error): a dataset produced in one group and used in
      another is only kept in memory.
    - `local_boundary` (warning): a dataset passed between groups is saved to
      local disk. Databricks shared paths such as `/dbfs/` and `/Volumes/` are
      not reported.
    - `single_node_groups` (info): groups that contain one node.

    Datasets are resolved the way the catalog resolves them at run time,
    including dataset factories and catch-all patterns, but from configuration
    only: no data is loaded and no dataset class is imported. The function
    never raises for issues it finds; call `raise_if_failed` on the result to
    raise.

    Args:
        pipeline: The pipeline to deploy.
        catalog: The catalog the deployed pipeline will use.
        groups: Groups to check. Defaults to `pipeline.group_nodes_by(group_by)`.
        group_by: Grouping strategy passed to `Pipeline.group_nodes_by` when
            `groups` is not given.

    Returns:
        A `DeploymentGroupingResult`, truthy when no errors were found.

    Example:
    ```python
        from kedro.inspection import validate_deployment_grouping

        result = validate_deployment_grouping(pipeline, catalog)
        if not result:
            for issue in result.errors:
                print(issue.message)
    ```
    """
    if groups is None:
        groups = pipeline.group_nodes_by(group_by)
    group_names = tuple(g.name for g in groups)

    structural = _check_structure(pipeline, groups)
    if structural:
        return DeploymentGroupingResult(groups=group_names, issues=tuple(structural))

    node_to_group = {node_name: g.name for g in groups for node_name in g.nodes}
    boundaries = _find_dataset_boundaries(pipeline, node_to_group)
    issues = [
        *_check_cycles(group_names, boundaries),
        *_check_dataset_boundaries(catalog, boundaries),
        *_check_single_node_groups(groups),
    ]
    issues.sort(key=lambda issue: _SEVERITY_ORDER[issue.severity])
    return DeploymentGroupingResult(groups=group_names, issues=tuple(issues))