Dependencies

With resolve_dependencies=True, the default, plan_all_items() and deploy_all_items() read what the selected items refer to, and deploy each item after what it needs, even in a run that deploys only what changed.

References read

Item	Needs	Read from
Report	SemanticModel	`definition.pbir`: the folder in `datasetReference.byPath`, or the `initial catalog` of `byConnection`
Notebook	Lakehouse	The default lakehouse: by logical ID, as Fabric writes a lakehouse of the same workspace, or else by `default_lakehouse_name`
Notebook	Environment	The attached environment, by logical ID
Notebook	Notebook	`%run <notebook>` (`%run -b` runs a script of the notebook's resources, not a notebook)

A reference to something outside the source, such as a semantic model of another workspace, is left out: it is neither deployed nor blocking.

What the plan does with them

Say Orders.Notebook changed, and its default lakehouse is Bronze.Lakehouse, which did not.

The workspace has the lakehouse: it is checked, not deployed.

NOOP     Bronze.Lakehouse  DEPENDENCY_REQUIRED: Required by Orders.Notebook (notebook default lakehouse); already in the workspace.
UPDATE   Orders.Notebook  SOURCE_CHANGED

The workspace lacks it, as when someone deleted it by hand: it is created first.

CREATE   Bronze.Lakehouse  DEPENDENCY_REQUIRED: Required by Orders.Notebook (notebook default lakehouse) and missing from the workspace.
UPDATE   Orders.Notebook  SOURCE_CHANGED

It cannot be created, because the run deploys only item_types=["Notebook"] or because the source does not define it: the notebook is blocked.

BLOCKED  Orders.Notebook  SOURCE_CHANGED: Needs Bronze.Lakehouse, which is missing from the workspace and not among the item types of this run.

An item is blocked too when a reference of its definition is broken, such as a report whose byPath folder is missing or holds no semantic model, or when it is part of a dependency cycle, such as two notebooks that %run each other. What needs a blocked item is blocked in turn. A blocked item is reported as failed, so the deployment state does not move.

Reports and their semantic model

In Git, a report's definition.pbir points to its semantic model by the path of its folder (byPath). The Fabric API accepts no path: only a connection to the model's ID in the workspace, which exists once the model is deployed. So when it deploys such a report, the engine sends the reference as a connection to the ID of that model in the target workspace, the one created earlier in the same run or the one already there. The file keeps its path.

In Git:     "byPath": {"path": "../Sales.SemanticModel"}
Sent:       "byConnection": {"connectionString": "semanticmodelid=<ID of Sales in the workspace>"}

A report whose model the workspace lacks fails with the reason, and nothing is sent. A report that already points to a model by connection is sent as it is.

When an item fails

The plan lists in each action's needs the items of the plan it needs. When one of them fails to deploy, the item is not sent without it: it is reported as skipped, and so is what needs a skipped item.

report = deploy_all_items("Sales-PRD", "stg/workspace", ...)
[(r.display_name, r.action, r.error) for r in report.results]
# [('Bronze', 'failed', 'Create failed with 400: ...'),
#  ('Orders', 'skipped', 'Needs Bronze.Lakehouse, which failed.')]

The rest of the run goes on, and the deployment state does not move, so the next run tries both again.

Data pipelines

A data pipeline refers to notebooks, pipelines, dataflows and other items by ID, which only the workspace can resolve. After staging, every ID that points to the target workspace, nested activities included, is checked against it. An ID the workspace lacks, or a value that is no ID, such as a placeholder left unreplaced, gives a warning in the plan and in the log:

UPDATE   Load.DataPipeline  SOURCE_CHANGED: Warning: Activity 'Run Orders' refers to notebook '#{orders_notebook_id}#', which is not an ID (a placeholder left unreplaced?).

The pipeline is still deployed: an item created in the same run gets its ID only when it is created. References to other workspaces are not checked.

Deletions

The same references keep an item deleted from Git in the workspace while an item that stays in the source refers to it, and so does the item's ID in the workspace when another item's files hold it: see Deletions. That check runs whatever resolve_dependencies says.

Turning it off

resolve_dependencies=False deploys the selected items in DEPLOY_ORDER without reading their references, as before: no item needs another, so a failure skips nothing.

Reference

References between local items, read from their definitions.

LocalCatalog lists the items under a folder with the identity the planner uses (item type and display name), their folder and their logical ID. scan_references reads the definitions of the selected items, and of the items they need, and turns each reference into a Dependency on a local item, or into a broken reference when the definition points to a local item that is not there. A reference to something outside the source, such as a semantic model of another workspace, is left out: it is neither deployed nor blocking.

References read:

Report → SemanticModel: datasetReference in definition.pbir, by path, or by connection through the initial catalog of the connection string.
Notebook → Lakehouse: the default lakehouse in the notebook metadata, by logical ID (as Fabric writes it for a lakehouse of the same workspace) or else by default_lakehouse_name.
Notebook → Environment: the attached environment, by logical ID.
Notebook → Notebook: %run <notebook>; %run -b runs a script of the notebook's resources, not a notebook.

pipeline_references reads what a data pipeline refers to by ID (notebooks, pipelines, dataflows, lakehouses and other items): those IDs are the workspace's own, so they are checked against it rather than resolved to local items.

`CatalogItem` `dataclass`

A local item that references can point to.

Attributes:

Name	Type	Description
`key`	`ItemKey`	Its item type, from the folder suffix, and its display name, from `.platform`.
`path`	`str`	Its folder.
`logical_id`	`str \| None`	`config.logicalId` from `.platform`.

Source code in src/pyfabricops/helpers/dependencies.py

@dataclass(frozen=True)
class CatalogItem:
    """
    A local item that references can point to.

    Attributes:
        key (ItemKey): Its item type, from the folder suffix, and its
            display name, from ``.platform``.
        path (str): Its folder.
        logical_id (str | None): ``config.logicalId`` from ``.platform``.
    """

    key: ItemKey
    path: str
    logical_id: str | None = None

`IdReference` `dataclass`

A reference by ID from a data pipeline to an item of a workspace.

Attributes:

Name	Type	Description
`kind`	`str`	What is referred to: `"notebook"`, `"pipeline"`, `"dataflow"`, or `"item"` for any other, such as the lakehouse a copy writes to.
`item_id`	`str`	The ID, which may be a placeholder left unreplaced.
`workspace_id`	`str \| None`	The workspace of the item, or None when the definition does not say, which means the pipeline's own.
`where`	`str`	Where the reference is, such as `"Activity 'Load sales'"`.

Source code in src/pyfabricops/helpers/dependencies.py

@dataclass(frozen=True)
class IdReference:
    """
    A reference by ID from a data pipeline to an item of a workspace.

    Attributes:
        kind (str): What is referred to: ``"notebook"``, ``"pipeline"``,
            ``"dataflow"``, or ``"item"`` for any other, such as the
            lakehouse a copy writes to.
        item_id (str): The ID, which may be a placeholder left unreplaced.
        workspace_id (str | None): The workspace of the item, or None when
            the definition does not say, which means the pipeline's own.
        where (str): Where the reference is, such as
            ``"Activity 'Load sales'"``.
    """

    kind: str
    item_id: str
    workspace_id: str | None
    where: str

`LocalCatalog`

The local items that references can point to.

Parameters:

Name	Type	Description	Default
`items`	`Iterable[CatalogItem]`	The items. When two share a type and display name, or a folder, the first one is kept.	required

Source code in src/pyfabricops/helpers/dependencies.py

class LocalCatalog:
    """
    The local items that references can point to.

    Args:
        items (Iterable[CatalogItem]): The items. When two share a type and
            display name, or a folder, the first one is kept.
    """

    def __init__(self, items: Iterable[CatalogItem]) -> None:
        self._by_key: dict[ItemKey, CatalogItem] = {}
        self._by_folder: dict[str, CatalogItem] = {}
        self._by_logical_id: dict[str, CatalogItem] = {}
        for item in items:
            self._by_key.setdefault(item.key, item)
            self._by_folder.setdefault(_folder_id(item.path), item)
            if item.logical_id is not None:
                self._by_logical_id.setdefault(item.logical_id, item)

    @classmethod
    def read(cls, path: str, item_types: Sequence[str]) -> LocalCatalog:
        """
        List the items of the given types under a folder.

        An item whose ``.platform`` cannot be read is left out; the planner
        reports it when it is selected.

        Args:
            path (str): The folder of the items.
            item_types (Sequence[str]): The item types to list.

        Returns:
            LocalCatalog: The items found.
        """
        items: list[CatalogItem] = []
        for item_type in item_types:
            for item_path in sorted(
                str(p) for p in list_paths_of_type(path, item_type)
            ):
                identity = _read_platform(item_path)
                if identity is not None:
                    display_name, logical_id = identity
                    items.append(
                        CatalogItem(
                            (item_type, display_name), item_path, logical_id
                        )
                    )
        return cls(items)

    def __iter__(self) -> Iterator[CatalogItem]:
        """Iterate over the items, one per type and display name."""
        return iter(self._by_key.values())

    def get(self, key: ItemKey) -> CatalogItem | None:
        """
        Return the item with a type and display name.

        Args:
            key (ItemKey): The item type and display name.

        Returns:
            CatalogItem | None: The item, or None when there is none.
        """
        return self._by_key.get(key)

    def at(self, folder: str) -> CatalogItem | None:
        """
        Return the item in a folder.

        Args:
            folder (str): The folder, in any form that points to it.

        Returns:
            CatalogItem | None: The item, or None when there is none.
        """
        return self._by_folder.get(_folder_id(folder))

    def with_logical_id(self, logical_id: str) -> CatalogItem | None:
        """
        Return the item with a logical ID.

        Args:
            logical_id (str): The ``config.logicalId`` of its ``.platform``.

        Returns:
            CatalogItem | None: The item, or None when there is none.
        """
        return self._by_logical_id.get(logical_id)

`iter()`

Iterate over the items, one per type and display name.

Source code in src/pyfabricops/helpers/dependencies.py

def __iter__(self) -> Iterator[CatalogItem]:
    """Iterate over the items, one per type and display name."""
    return iter(self._by_key.values())

`at(folder)`

Return the item in a folder.

Parameters:

Name	Type	Description	Default
`folder`	`str`	The folder, in any form that points to it.	required

Returns:

Type	Description
`CatalogItem \| None`	CatalogItem \| None: The item, or None when there is none.

Source code in src/pyfabricops/helpers/dependencies.py

def at(self, folder: str) -> CatalogItem | None:
    """
    Return the item in a folder.

    Args:
        folder (str): The folder, in any form that points to it.

    Returns:
        CatalogItem | None: The item, or None when there is none.
    """
    return self._by_folder.get(_folder_id(folder))

`get(key)`

Return the item with a type and display name.

Parameters:

Name	Type	Description	Default
`key`	`ItemKey`	The item type and display name.	required

Returns:

Type	Description
`CatalogItem \| None`	CatalogItem \| None: The item, or None when there is none.

Source code in src/pyfabricops/helpers/dependencies.py

def get(self, key: ItemKey) -> CatalogItem | None:
    """
    Return the item with a type and display name.

    Args:
        key (ItemKey): The item type and display name.

    Returns:
        CatalogItem | None: The item, or None when there is none.
    """
    return self._by_key.get(key)

`read(path, item_types)` `classmethod`

List the items of the given types under a folder.

An item whose .platform cannot be read is left out; the planner reports it when it is selected.

Parameters:

Name	Type	Description	Default
`path`	`str`	The folder of the items.	required
`item_types`	`Sequence[str]`	The item types to list.	required

Returns:

Name	Type	Description
`LocalCatalog`	`LocalCatalog`	The items found.

Source code in src/pyfabricops/helpers/dependencies.py

@classmethod
def read(cls, path: str, item_types: Sequence[str]) -> LocalCatalog:
    """
    List the items of the given types under a folder.

    An item whose ``.platform`` cannot be read is left out; the planner
    reports it when it is selected.

    Args:
        path (str): The folder of the items.
        item_types (Sequence[str]): The item types to list.

    Returns:
        LocalCatalog: The items found.
    """
    items: list[CatalogItem] = []
    for item_type in item_types:
        for item_path in sorted(
            str(p) for p in list_paths_of_type(path, item_type)
        ):
            identity = _read_platform(item_path)
            if identity is not None:
                display_name, logical_id = identity
                items.append(
                    CatalogItem(
                        (item_type, display_name), item_path, logical_id
                    )
                )
    return cls(items)

`with_logical_id(logical_id)`

Return the item with a logical ID.

Parameters:

Name	Type	Description	Default
`logical_id`	`str`	The `config.logicalId` of its `.platform`.	required

Returns:

Type	Description
`CatalogItem \| None`	CatalogItem \| None: The item, or None when there is none.

Source code in src/pyfabricops/helpers/dependencies.py

def with_logical_id(self, logical_id: str) -> CatalogItem | None:
    """
    Return the item with a logical ID.

    Args:
        logical_id (str): The ``config.logicalId`` of its ``.platform``.

    Returns:
        CatalogItem | None: The item, or None when there is none.
    """
    return self._by_logical_id.get(logical_id)

`ReferenceScan` `dataclass`

The references found from a selection of items.

Attributes:

Name	Type	Description
`dependencies`	`tuple[Dependency, ...]`	The references to local items.
`broken`	`Mapping[ItemKey, tuple[str, ...]]`	For each item whose definition points to a local item that is not there, why. Read-only.

Source code in src/pyfabricops/helpers/dependencies.py

@dataclass(frozen=True)
class ReferenceScan:
    """
    The references found from a selection of items.

    Attributes:
        dependencies (tuple[Dependency, ...]): The references to local items.
        broken (Mapping[ItemKey, tuple[str, ...]]): For each item whose
            definition points to a local item that is not there, why.
            Read-only.
    """

    dependencies: tuple[Dependency, ...] = ()
    broken: Mapping[ItemKey, tuple[str, ...]] = field(default_factory=dict)

    def __post_init__(self) -> None:
        object.__setattr__(self, "broken", MappingProxyType(dict(self.broken)))

`pipeline_references(folder)`

Read what a data pipeline refers to by ID.

Every activity is read, those nested in ForEach, If Condition, Switch and Until included.

Parameters:

Name	Type	Description	Default
`folder`	`str`	The folder of the pipeline, with its `pipeline-content.json`.	required

Returns:

Type	Description
`tuple[IdReference, ...]`	tuple[IdReference, ...]: The references, each once, in the order found; none when the content cannot be read.

Source code in src/pyfabricops/helpers/dependencies.py

def pipeline_references(folder: str) -> tuple[IdReference, ...]:
    """
    Read what a data pipeline refers to by ID.

    Every activity is read, those nested in ForEach, If Condition, Switch
    and Until included.

    Args:
        folder (str): The folder of the pipeline, with its
            ``pipeline-content.json``.

    Returns:
        tuple[IdReference, ...]: The references, each once, in the order
            found; none when the content cannot be read.
    """
    found: list[IdReference] = []
    _walk_pipeline(
        _read_json(Path(folder) / "pipeline-content.json"),
        "The pipeline",
        found,
    )
    return tuple(dict.fromkeys(found))

`scan_references(catalog, keys)`

Read the references of the given items and of every item they need.

Parameters:

Name	Type	Description	Default
`catalog`	`LocalCatalog`	The local items.	required
`keys`	`Iterable[ItemKey]`	The items to start from. An item not in the catalog, or of a type whose references are not read, gives none.	required

Returns:

Name	Type	Description
`ReferenceScan`	`ReferenceScan`	What the items refer to.

Source code in src/pyfabricops/helpers/dependencies.py

def scan_references(
    catalog: LocalCatalog, keys: Iterable[ItemKey]
) -> ReferenceScan:
    """
    Read the references of the given items and of every item they need.

    Args:
        catalog (LocalCatalog): The local items.
        keys (Iterable[ItemKey]): The items to start from. An item not in
            the catalog, or of a type whose references are not read, gives
            none.

    Returns:
        ReferenceScan: What the items refer to.
    """
    dependencies: list[Dependency] = []
    broken: dict[ItemKey, tuple[str, ...]] = {}
    pending = deque(dict.fromkeys(keys))
    scanned: set[ItemKey] = set()
    while pending:
        key = pending.popleft()
        if key in scanned:
            continue
        scanned.add(key)
        item = catalog.get(key)
        read = _READERS.get(key[0])
        if item is None or read is None:
            continue
        found, problems = read(item, catalog)
        dependencies.extend(found)
        pending.extend(dependency.target for dependency in found)
        if problems:
            broken[key] = tuple(problems)
    return ReferenceScan(tuple(dependencies), broken)

Dependency graph: which local items need which, and in what order.

An item depends on another when its definition refers to it, such as a report on its semantic model. DependencyGraph holds those references between items of the source and answers three questions for the planner: what the selected items need (required_by), in which order to deploy items so that each comes after what it needs (order), and which items need one another (cycles). Like the planner, it calls no Fabric API and reads no file: finding the references is the engine's job.

`ItemKey = tuple[str, str]` `module-attribute`

An item, as (item_type, display_name).

`Dependency` `dataclass`

A reference from one local item to another.

Attributes:

Name	Type	Description
`source`	`ItemKey`	The item whose definition holds the reference.
`target`	`ItemKey`	The item it refers to, needed first.
`via`	`str`	Where the reference is, such as `"definition.pbir"`, to explain it in a plan.

Source code in src/pyfabricops/helpers/dependency_graph.py

@dataclass(frozen=True)
class Dependency:
    """
    A reference from one local item to another.

    Attributes:
        source (ItemKey): The item whose definition holds the reference.
        target (ItemKey): The item it refers to, needed first.
        via (str): Where the reference is, such as ``"definition.pbir"``,
            to explain it in a plan.
    """

    source: ItemKey
    target: ItemKey
    via: str

`DependencyGraph`

The references between local items.

Parameters:

Name	Type	Description	Default
`dependencies`	`Iterable[Dependency]`	The references found in the source; a repeated one counts once. Defaults to none.	`()`

Examples:

report, model = ("Report", "Sales"), ("SemanticModel", "Sales")
graph = DependencyGraph([Dependency(report, model, "definition.pbir")])
graph.order([report, model])  # (model, report)

Source code in src/pyfabricops/helpers/dependency_graph.py

class DependencyGraph:
    """
    The references between local items.

    Args:
        dependencies (Iterable[Dependency], optional): The references found
            in the source; a repeated one counts once. Defaults to none.

    Examples:
        ```python
        report, model = ("Report", "Sales"), ("SemanticModel", "Sales")
        graph = DependencyGraph([Dependency(report, model, "definition.pbir")])
        graph.order([report, model])  # (model, report)
        ```
    """

    def __init__(self, dependencies: Iterable[Dependency] = ()) -> None:
        self._dependencies: dict[ItemKey, list[Dependency]] = {}
        for dependency in dict.fromkeys(dependencies):
            self._dependencies.setdefault(dependency.source, []).append(
                dependency
            )

    def dependencies_of(self, key: ItemKey) -> tuple[Dependency, ...]:
        """
        Return the references an item holds, in the order they were found.

        Args:
            key (ItemKey): The item.

        Returns:
            tuple[Dependency, ...]: Its references, possibly none.
        """
        return tuple(self._dependencies.get(key, ()))

    def required_by(self, keys: Iterable[ItemKey]) -> tuple[ItemKey, ...]:
        """
        Return every item the given ones need, directly or through others.

        Args:
            keys (Iterable[ItemKey]): The items to start from.

        Returns:
            tuple[ItemKey, ...]: The items they need, without the given ones,
                nearest first.
        """
        start = list(dict.fromkeys(keys))
        seen = set(start)
        pending = deque(start)
        required: list[ItemKey] = []
        while pending:
            for dependency in self.dependencies_of(pending.popleft()):
                if dependency.target not in seen:
                    seen.add(dependency.target)
                    required.append(dependency.target)
                    pending.append(dependency.target)
        return tuple(required)

    def order(self, keys: Sequence[ItemKey]) -> tuple[ItemKey, ...]:
        """
        Order items so that each comes after the items it needs.

        Only references between the given items count. Each item keeps its
        place in the order given, preceded by the items it needs that have
        not come yet, so a caller that sorts items by type and path first
        keeps that order wherever nothing else is needed. Items that need one
        another stay together, in the order given, after what they need.

        Args:
            keys (Sequence[ItemKey]): The items, in their default order.

        Returns:
            tuple[ItemKey, ...]: The same items, each after the ones it needs.
        """
        components = self._components(keys)
        component_of = {
            key: n for n, members in enumerate(components) for key in members
        }
        needs = [
            sorted(
                {
                    component_of[dependency.target]
                    for key in members
                    for dependency in self.dependencies_of(key)
                    if dependency.target in component_of
                }
                - {n}
            )
            for n, members in enumerate(components)
        ]

        ordered: list[ItemKey] = []
        placed = [False] * len(components)
        for root in range(len(components)):
            if placed[root]:
                continue
            placed[root] = True
            # Place what a component needs before the component itself.
            stack: list[tuple[int, Iterator[int]]] = [
                (root, iter(needs[root]))
            ]
            while stack:
                component, pending = stack[-1]
                for needed in pending:
                    if not placed[needed]:
                        placed[needed] = True
                        stack.append((needed, iter(needs[needed])))
                        break
                else:
                    stack.pop()
                    ordered.extend(components[component])
        return tuple(ordered)

    def cycles(
        self, keys: Sequence[ItemKey]
    ) -> tuple[tuple[ItemKey, ...], ...]:
        """
        Return the groups of items that need one another.

        Only references between the given items count; an item that refers
        to itself is a group on its own.

        Args:
            keys (Sequence[ItemKey]): The items to check.

        Returns:
            tuple[tuple[ItemKey, ...], ...]: Each group in the order given,
                the groups in the order of their first item.
        """
        return tuple(
            members
            for members in self._components(keys)
            if len(members) > 1
            or any(
                dependency.target == members[0]
                for dependency in self.dependencies_of(members[0])
            )
        )

    def _components(
        self, keys: Sequence[ItemKey]
    ) -> list[tuple[ItemKey, ...]]:
        """
        Group the items that reach one another through their references.

        Returns the strongly connected components of the references between
        the given items, each in the order given, ordered by their first item.
        """
        items = list(dict.fromkeys(keys))
        index = {key: n for n, key in enumerate(items)}
        forward = {
            key: [
                dependency.target
                for dependency in self.dependencies_of(key)
                if dependency.target in index
            ]
            for key in items
        }
        backward: dict[ItemKey, list[ItemKey]] = {key: [] for key in items}
        for key, targets in forward.items():
            for target in targets:
                backward[target].append(key)

        # First pass: the order in which items finish on the references.
        finished: list[ItemKey] = []
        visited: set[ItemKey] = set()
        for root in items:
            if root in visited:
                continue
            visited.add(root)
            stack: list[tuple[ItemKey, Iterator[ItemKey]]] = [
                (root, iter(forward[root]))
            ]
            while stack:
                key, pending = stack[-1]
                for target in pending:
                    if target not in visited:
                        visited.add(target)
                        stack.append((target, iter(forward[target])))
                        break
                else:
                    stack.pop()
                    finished.append(key)

        # Second pass: against the references, last finished first.
        assigned: set[ItemKey] = set()
        components: list[tuple[ItemKey, ...]] = []
        for root in reversed(finished):
            if root in assigned:
                continue
            assigned.add(root)
            members = [root]
            to_visit = [root]
            while to_visit:
                for source in backward[to_visit.pop()]:
                    if source not in assigned:
                        assigned.add(source)
                        members.append(source)
                        to_visit.append(source)
            components.append(tuple(sorted(members, key=index.__getitem__)))
        components.sort(key=lambda members: index[members[0]])
        return components

`cycles(keys)`

Return the groups of items that need one another.

Only references between the given items count; an item that refers to itself is a group on its own.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[ItemKey]`	The items to check.	required

Returns:

Type	Description
`tuple[tuple[ItemKey, ...], ...]`	tuple[tuple[ItemKey, ...], ...]: Each group in the order given, the groups in the order of their first item.

Source code in src/pyfabricops/helpers/dependency_graph.py

def cycles(
    self, keys: Sequence[ItemKey]
) -> tuple[tuple[ItemKey, ...], ...]:
    """
    Return the groups of items that need one another.

    Only references between the given items count; an item that refers
    to itself is a group on its own.

    Args:
        keys (Sequence[ItemKey]): The items to check.

    Returns:
        tuple[tuple[ItemKey, ...], ...]: Each group in the order given,
            the groups in the order of their first item.
    """
    return tuple(
        members
        for members in self._components(keys)
        if len(members) > 1
        or any(
            dependency.target == members[0]
            for dependency in self.dependencies_of(members[0])
        )
    )

`dependencies_of(key)`

Return the references an item holds, in the order they were found.

Parameters:

Name	Type	Description	Default
`key`	`ItemKey`	The item.	required

Returns:

Type	Description
`tuple[Dependency, ...]`	tuple[Dependency, ...]: Its references, possibly none.

Source code in src/pyfabricops/helpers/dependency_graph.py

def dependencies_of(self, key: ItemKey) -> tuple[Dependency, ...]:
    """
    Return the references an item holds, in the order they were found.

    Args:
        key (ItemKey): The item.

    Returns:
        tuple[Dependency, ...]: Its references, possibly none.
    """
    return tuple(self._dependencies.get(key, ()))

`order(keys)`

Order items so that each comes after the items it needs.

Only references between the given items count. Each item keeps its place in the order given, preceded by the items it needs that have not come yet, so a caller that sorts items by type and path first keeps that order wherever nothing else is needed. Items that need one another stay together, in the order given, after what they need.

Parameters:

Name	Type	Description	Default
`keys`	`Sequence[ItemKey]`	The items, in their default order.	required

Returns:

Type	Description
`tuple[ItemKey, ...]`	tuple[ItemKey, ...]: The same items, each after the ones it needs.

Source code in src/pyfabricops/helpers/dependency_graph.py

def order(self, keys: Sequence[ItemKey]) -> tuple[ItemKey, ...]:
    """
    Order items so that each comes after the items it needs.

    Only references between the given items count. Each item keeps its
    place in the order given, preceded by the items it needs that have
    not come yet, so a caller that sorts items by type and path first
    keeps that order wherever nothing else is needed. Items that need one
    another stay together, in the order given, after what they need.

    Args:
        keys (Sequence[ItemKey]): The items, in their default order.

    Returns:
        tuple[ItemKey, ...]: The same items, each after the ones it needs.
    """
    components = self._components(keys)
    component_of = {
        key: n for n, members in enumerate(components) for key in members
    }
    needs = [
        sorted(
            {
                component_of[dependency.target]
                for key in members
                for dependency in self.dependencies_of(key)
                if dependency.target in component_of
            }
            - {n}
        )
        for n, members in enumerate(components)
    ]

    ordered: list[ItemKey] = []
    placed = [False] * len(components)
    for root in range(len(components)):
        if placed[root]:
            continue
        placed[root] = True
        # Place what a component needs before the component itself.
        stack: list[tuple[int, Iterator[int]]] = [
            (root, iter(needs[root]))
        ]
        while stack:
            component, pending = stack[-1]
            for needed in pending:
                if not placed[needed]:
                    placed[needed] = True
                    stack.append((needed, iter(needs[needed])))
                    break
            else:
                stack.pop()
                ordered.extend(components[component])
    return tuple(ordered)

`required_by(keys)`

Return every item the given ones need, directly or through others.

Parameters:

Name	Type	Description	Default
`keys`	`Iterable[ItemKey]`	The items to start from.	required

Returns:

Type	Description
`tuple[ItemKey, ...]`	tuple[ItemKey, ...]: The items they need, without the given ones, nearest first.

Source code in src/pyfabricops/helpers/dependency_graph.py

def required_by(self, keys: Iterable[ItemKey]) -> tuple[ItemKey, ...]:
    """
    Return every item the given ones need, directly or through others.

    Args:
        keys (Iterable[ItemKey]): The items to start from.

    Returns:
        tuple[ItemKey, ...]: The items they need, without the given ones,
            nearest first.
    """
    start = list(dict.fromkeys(keys))
    seen = set(start)
    pending = deque(start)
    required: list[ItemKey] = []
    while pending:
        for dependency in self.dependencies_of(pending.popleft()):
            if dependency.target not in seen:
                seen.add(dependency.target)
                required.append(dependency.target)
                pending.append(dependency.target)
    return tuple(required)

Dependencies

References read

What the plan does with them

Reports and their semantic model

When an item fails

Data pipelines

Deletions

Turning it off

Reference

CatalogItem dataclass

IdReference dataclass

LocalCatalog

__iter__()

at(folder)

get(key)

read(path, item_types) classmethod

with_logical_id(logical_id)

ReferenceScan dataclass

pipeline_references(folder)

scan_references(catalog, keys)

ItemKey = tuple[str, str] module-attribute

Dependency dataclass

DependencyGraph

cycles(keys)

dependencies_of(key)

order(keys)

required_by(keys)

`CatalogItem` `dataclass`

`IdReference` `dataclass`

`LocalCatalog`

`iter()`

`at(folder)`

`get(key)`

`read(path, item_types)` `classmethod`

`with_logical_id(logical_id)`

`ReferenceScan` `dataclass`

`pipeline_references(folder)`

`scan_references(catalog, keys)`

`ItemKey = tuple[str, str]` `module-attribute`

`Dependency` `dataclass`

`DependencyGraph`

`cycles(keys)`

`dependencies_of(key)`

`order(keys)`

`required_by(keys)`