Deployment

Deployment engine shared by the deploy_all_* helpers.

A run plans first, then applies. It selects the local items (every item, or only those changed in Git since a baseline commit, which a deployment state can supply per item type), lists the workspace items and folders once, and builds a DeploymentPlan from them without changing anything. DeploymentExecutor then applies the plan, and the run returns a DeploymentReport with the outcome of each item, so a partial failure reaches the caller instead of being logged and lost. The state is recorded only when every item succeeded. An item deleted from Git is deleted from the workspace only when the run allows deletions, after every other item succeeded, and only while no item that stays in the source refers to it.

DeploymentExecutor

Apply a deployment plan to a workspace through the Fabric API.

The executor decides nothing: it creates the items planned as CREATE, updates the items planned as UPDATE (moving them first when their folder differs), moves the items planned as MOVE without sending their definition, deletes the items planned as DELETE, and reports BLOCKED items as failed, creating missing folders on the way. A NOOP gets a log line but no result. An item to create, update or move is skipped when an item of its needs failed or was skipped, so nothing is deployed without what it needs. A deletion is refused, and reported as failed, unless allow_deletions is set; it is skipped when any action before it failed or was skipped. A report that points to its semantic model by path is sent with a connection to the model's ID in the workspace instead, since the Fabric API accepts no path. Used by deploy_all_items; not exported from pyfabricops yet.

Parameters:

Name	Type	Description	Default
index	_WorkspaceIndex	The workspace items and folders the plan was built from.	required
fail_fast	bool	Stop at the first failed action and report the remaining ones as skipped. Defaults to False.	False
allow_deletions	bool	Delete the items planned as DELETE. Defaults to False: each is reported as failed and left in the workspace.	False
on_result	Callable	Called with each action and its result as soon as the result is known, such as to write a journal. Defaults to none.	None

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

class DeploymentExecutor:
    """
    Apply a deployment plan to a workspace through the Fabric API.

    The executor decides nothing: it creates the items planned as CREATE,
    updates the items planned as UPDATE (moving them first when their folder
    differs), moves the items planned as MOVE without sending their
    definition, deletes the items planned as DELETE, and reports BLOCKED
    items as failed, creating missing folders on the way. A NOOP gets a log
    line but no result. An item to create, update or move is skipped when
    an item of its ``needs`` failed or was skipped, so nothing is deployed
    without what it needs. A deletion is refused, and reported as failed,
    unless ``allow_deletions`` is set; it is skipped when any action before
    it failed or was skipped. A report that points to its semantic model by
    path is sent with a connection to the model's ID in the workspace
    instead, since the Fabric API accepts no path. Used by
    ``deploy_all_items``; not exported from ``pyfabricops`` yet.

    Args:
        index (_WorkspaceIndex): The workspace items and folders the plan was
            built from.
        fail_fast (bool, optional): Stop at the first failed action and
            report the remaining ones as skipped. Defaults to False.
        allow_deletions (bool, optional): Delete the items planned as
            DELETE. Defaults to False: each is reported as failed and left
            in the workspace.
        on_result (Callable, optional): Called with each action and its
            result as soon as the result is known, such as to write a
            journal. Defaults to none.
    """

    def __init__(
        self,
        index: _WorkspaceIndex,
        *,
        fail_fast: bool = False,
        allow_deletions: bool = False,
        on_result: Callable[[DeploymentAction, DeploymentResult], None]
        | None = None,
    ) -> None:
        self._index = index
        self._fail_fast = fail_fast
        self._allow_deletions = allow_deletions
        self._on_result = on_result

    def apply(self, plan: DeploymentPlan) -> list[DeploymentResult]:
        """
        Execute the actions of a plan, in order.

        Args:
            plan (DeploymentPlan): The plan to execute.

        Returns:
            list[DeploymentResult]: One result per action other than NOOP,
                in plan order.
        """
        results: list[DeploymentResult] = []
        # Each item of the run that was not deployed, as a detail for what
        # needs it: "Name.Type, which failed".
        not_deployed: dict[tuple[str, str], str] = {}
        # Whether every action so far succeeded, which a deletion needs.
        clean = True
        for position, action in enumerate(plan.actions):
            if action.action is DeploymentActionType.NOOP:
                _log_noop(action)
                continue

            unmet = [key for key in action.needs if key in not_deployed]
            if unmet and action.action in _APPLIED:
                result = _action_result(
                    action, "skipped", error=f"Needs {not_deployed[unmet[0]]}."
                )
            elif action.action is DeploymentActionType.DELETE and not clean:
                result = _action_result(
                    action,
                    "skipped",
                    error="Not deleted: an item before it failed or was "
                    "skipped.",
                )
            else:
                result = _apply_action(
                    self._index,
                    action,
                    allow_deletions=self._allow_deletions,
                )
            results.append(result)
            _log_result(result)
            self._tell(action, result)
            which = _NOT_DEPLOYED.get(result.action)
            if which is not None:
                clean = False
                if action.display_name:
                    not_deployed[_identity(action)] = (
                        f"{_label(result)}, which {which}"
                    )

            if result.action == "failed" and self._fail_fast:
                for skipped in plan.actions[position + 1 :]:
                    if skipped.action is not DeploymentActionType.NOOP:
                        left = _action_result(skipped, "skipped")
                        results.append(left)
                        self._tell(skipped, left)
                break
        return results

    def _tell(
        self, action: DeploymentAction, result: DeploymentResult
    ) -> None:
        """Pass a result on, to whoever asked to know."""
        if self._on_result is not None:
            self._on_result(action, result)

apply(plan)

Execute the actions of a plan, in order.

Parameters:

Name	Type	Description	Default
plan	DeploymentPlan	The plan to execute.	required

Returns:

Type	Description
list[DeploymentResult]	list[DeploymentResult]: One result per action other than NOOP, in plan order.

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

def apply(self, plan: DeploymentPlan) -> list[DeploymentResult]:
    """
    Execute the actions of a plan, in order.

    Args:
        plan (DeploymentPlan): The plan to execute.

    Returns:
        list[DeploymentResult]: One result per action other than NOOP,
            in plan order.
    """
    results: list[DeploymentResult] = []
    # Each item of the run that was not deployed, as a detail for what
    # needs it: "Name.Type, which failed".
    not_deployed: dict[tuple[str, str], str] = {}
    # Whether every action so far succeeded, which a deletion needs.
    clean = True
    for position, action in enumerate(plan.actions):
        if action.action is DeploymentActionType.NOOP:
            _log_noop(action)
            continue

        unmet = [key for key in action.needs if key in not_deployed]
        if unmet and action.action in _APPLIED:
            result = _action_result(
                action, "skipped", error=f"Needs {not_deployed[unmet[0]]}."
            )
        elif action.action is DeploymentActionType.DELETE and not clean:
            result = _action_result(
                action,
                "skipped",
                error="Not deleted: an item before it failed or was "
                "skipped.",
            )
        else:
            result = _apply_action(
                self._index,
                action,
                allow_deletions=self._allow_deletions,
            )
        results.append(result)
        _log_result(result)
        self._tell(action, result)
        which = _NOT_DEPLOYED.get(result.action)
        if which is not None:
            clean = False
            if action.display_name:
                not_deployed[_identity(action)] = (
                    f"{_label(result)}, which {which}"
                )

        if result.action == "failed" and self._fail_fast:
            for skipped in plan.actions[position + 1 :]:
                if skipped.action is not DeploymentActionType.NOOP:
                    left = _action_result(skipped, "skipped")
                    results.append(left)
                    self._tell(skipped, left)
            break
    return results

DeploymentReport dataclass

Per-item outcome of a deploy_all_* run.

Attributes:

Name	Type	Description
workspace	str	The workspace name or ID the run targeted.
workspace_id	str | None	The resolved workspace ID.
results	list[DeploymentResult]	One result per item acted on, in deployment order. An item that needs nothing has none.

Examples:

report = deploy_all_items('Sales-PRD', 'stg/workspace')
if report.failed:
    raise SystemExit(1)
report.durations_by_type()

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

@dataclass
class DeploymentReport:
    """
    Per-item outcome of a ``deploy_all_*`` run.

    Attributes:
        workspace (str): The workspace name or ID the run targeted.
        workspace_id (str | None): The resolved workspace ID.
        results (list[DeploymentResult]): One result per item acted on, in
            deployment order. An item that needs nothing has none.

    Examples:
        ```python
        report = deploy_all_items('Sales-PRD', 'stg/workspace')
        if report.failed:
            raise SystemExit(1)
        report.durations_by_type()
        ```
    """

    workspace: str
    workspace_id: str | None = None
    results: list[DeploymentResult] = field(default_factory=list)

    @property
    def failed(self) -> list[DeploymentResult]:
        """The items that failed."""
        return [r for r in self.results if r.action == "failed"]

    @property
    def skipped(self) -> list[DeploymentResult]:
        """
        The items not attempted.

        An item they need was not deployed, or an earlier item failed with
        ``fail_fast``.
        """
        return [r for r in self.results if r.action == "skipped"]

    @property
    def ok(self) -> bool:
        """True when every item was created, updated, moved or deleted."""
        return all(
            r.action in ("created", "updated", "moved", "deleted")
            for r in self.results
        )

    @property
    def duration_seconds(self) -> float:
        """Total wall-clock time spent on the items."""
        return sum(r.duration_seconds for r in self.results)

    def summary(self) -> dict[str, int]:
        """
        Count the items by action.

        Returns:
            dict[str, int]: The number of items created, updated, moved,
                deleted, failed and skipped.
        """
        counts: dict[str, int] = {action: 0 for action in _ACTIONS}
        for result in self.results:
            counts[result.action] += 1
        return counts

    def describe(self) -> str:
        """
        Describe the run: one line per item, then the counts and the time.

        Returns:
            str: Such as ``updated  Orders.Notebook  (2.1s)``, a failed or
                skipped item followed by why, and a last line of counts.

        Examples:
            ```python
            report = deploy_all_items('Sales-PRD', staging)
            print(report.describe())
            ```
        """
        lines = []
        for result in self.results:
            line = f"{result.action:<8} {_label(result)}"
            if result.error:
                line += f": {result.error}"
            elif result.duration_seconds:
                line += f"  ({result.duration_seconds:.1f}s)"
            lines.append(line)
        counts = self.summary()
        lines.append(
            ", ".join(f"{counts[action]} {action}" for action in _ACTIONS)
            + f" in {self.duration_seconds:.1f}s"
        )
        return "\n".join(lines)

    def durations_by_type(self) -> dict[str, float]:
        """
        Sum the wall-clock time spent per item type.

        Returns:
            dict[str, float]: Seconds per item type, in deployment order.
        """
        durations: dict[str, float] = {}
        for result in self.results:
            durations[result.item_type] = (
                durations.get(result.item_type, 0.0) + result.duration_seconds
            )
        return durations

    def to_df(self) -> DataFrame:
        """
        Return the results as a DataFrame.

        Returns:
            DataFrame: One row per item, one column per result attribute.
        """
        columns = [f.name for f in fields(DeploymentResult)]
        return DataFrame([asdict(r) for r in self.results], columns=columns)

duration_seconds property

Total wall-clock time spent on the items.

failed property

The items that failed.

ok property

True when every item was created, updated, moved or deleted.

skipped property

The items not attempted.

An item they need was not deployed, or an earlier item failed with fail_fast.

describe()

Describe the run: one line per item, then the counts and the time.

Returns:

Name	Type	Description
str	str	Such as updated Orders.Notebook (2.1s), a failed or skipped item followed by why, and a last line of counts.

Examples:

report = deploy_all_items('Sales-PRD', staging)
print(report.describe())

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

def describe(self) -> str:
    """
    Describe the run: one line per item, then the counts and the time.

    Returns:
        str: Such as ``updated  Orders.Notebook  (2.1s)``, a failed or
            skipped item followed by why, and a last line of counts.

    Examples:
        ```python
        report = deploy_all_items('Sales-PRD', staging)
        print(report.describe())
        ```
    """
    lines = []
    for result in self.results:
        line = f"{result.action:<8} {_label(result)}"
        if result.error:
            line += f": {result.error}"
        elif result.duration_seconds:
            line += f"  ({result.duration_seconds:.1f}s)"
        lines.append(line)
    counts = self.summary()
    lines.append(
        ", ".join(f"{counts[action]} {action}" for action in _ACTIONS)
        + f" in {self.duration_seconds:.1f}s"
    )
    return "\n".join(lines)

durations_by_type()

Sum the wall-clock time spent per item type.

Returns:

Type	Description
dict[str, float]	dict[str, float]: Seconds per item type, in deployment order.

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

def durations_by_type(self) -> dict[str, float]:
    """
    Sum the wall-clock time spent per item type.

    Returns:
        dict[str, float]: Seconds per item type, in deployment order.
    """
    durations: dict[str, float] = {}
    for result in self.results:
        durations[result.item_type] = (
            durations.get(result.item_type, 0.0) + result.duration_seconds
        )
    return durations

summary()

Count the items by action.

Returns:

Type	Description
dict[str, int]	dict[str, int]: The number of items created, updated, moved, deleted, failed and skipped.

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

def summary(self) -> dict[str, int]:
    """
    Count the items by action.

    Returns:
        dict[str, int]: The number of items created, updated, moved,
            deleted, failed and skipped.
    """
    counts: dict[str, int] = {action: 0 for action in _ACTIONS}
    for result in self.results:
        counts[result.action] += 1
    return counts

to_df()

Return the results as a DataFrame.

Returns:

Name	Type	Description
DataFrame	DataFrame	One row per item, one column per result attribute.

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

def to_df(self) -> DataFrame:
    """
    Return the results as a DataFrame.

    Returns:
        DataFrame: One row per item, one column per result attribute.
    """
    columns = [f.name for f in fields(DeploymentResult)]
    return DataFrame([asdict(r) for r in self.results], columns=columns)

DeploymentResult dataclass

Outcome of deploying one local item.

Attributes:

Name	Type	Description
item_type	str	The Fabric item type, from the folder suffix.
display_name	str | None	The display name from .platform, or None when it could not be read.
path	str	The local item folder.
action	str	"created", "updated", "moved" (only its folder changed, so its definition was not sent), "deleted", "failed", or "skipped" when the item was not attempted: an item it needs was not deployed, an earlier one failed with fail_fast=True, or, for a deletion, an earlier item was not deployed.
item_id	str | None	The item ID in the workspace, when known.
moved	bool	Whether the item was moved to another folder.
duration_seconds	float	Wall-clock time spent on the item.
error	str | None	Why the item failed, or which item it needs was not deployed.

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

@dataclass(frozen=True)
class DeploymentResult:
    """
    Outcome of deploying one local item.

    Attributes:
        item_type (str): The Fabric item type, from the folder suffix.
        display_name (str | None): The display name from ``.platform``, or
            None when it could not be read.
        path (str): The local item folder.
        action (str): ``"created"``, ``"updated"``, ``"moved"`` (only its
            folder changed, so its definition was not sent), ``"deleted"``,
            ``"failed"``, or ``"skipped"`` when the item was not attempted:
            an item it needs was not deployed, an earlier one failed with
            ``fail_fast=True``, or, for a deletion, an earlier item was not
            deployed.
        item_id (str | None): The item ID in the workspace, when known.
        moved (bool): Whether the item was moved to another folder.
        duration_seconds (float): Wall-clock time spent on the item.
        error (str | None): Why the item failed, or which item it needs
            was not deployed.
    """

    item_type: str
    display_name: str | None
    path: str
    action: DeploymentOutcome
    item_id: str | None = None
    moved: bool = False
    duration_seconds: float = 0.0
    error: str | None = None