Reconciliation

reconcile_items() tells how a workspace stands against the source: what the workspace lacks, what changed in it, and what it holds that the source does not. It changes nothing. It reads the definition of every item found on both sides, so it suits a scheduled run, or a check before a release, more than every deployment.

staging = copy_to_staging("workspace", staging_dir=tmp)
find_and_replace(staging, {(r".*\.tmdl$", r"#\{environment\}#"): "PRD"})

reconciliation = reconcile_items(
    "Sales-PRD",
    staging,
    start_path=staging,
    state_backend=LocalJsonStateBackend(".deploy-state"),
    environment="PRD",
)
print(reconciliation.describe())
if not reconciliation.ok:
    raise SystemExit(1)

Compare the staging copy, with the placeholders of the environment replaced, as a deployment would send it.

What it finds

UPDATE   Orders.Notebook  WORKSPACE_DRIFT: Changed in the workspace since the last deployment: notebook-content.py.
CREATE   Sales.Report  TARGET_MISSING: Deleted from the workspace since the last deployment.
MOVE     Utils.Notebook  WORKSPACE_DRIFT: Moved to 'Archive' in the workspace since the last deployment; the source has it in 'Data'.
UNMANAGED Draft.Notebook: in the workspace, not in the source.
UNCHECKED Logs.Eventhouse: Fabric returned no definition: 400: OperationNotSupportedForItem - ...
3 in sync, 1 create, 1 update, 1 move, 0 blocked, 1 unmanaged, 1 unchecked

Line	Meaning
CREATE TARGET_MISSING	In the source, not in the workspace.
UPDATE WORKSPACE_DRIFT	The definition differs: changed in the workspace since the last deployment or, without a deployment state, different from the source.
UPDATE SOURCE_CHANGED	Changed in the source since the last deployment: a deployment still to run, not drift.
MOVE	In another workspace folder than the source says.
BLOCKED	A local item that cannot be read, or that the source defines twice.
UNMANAGED	In the workspace, not in the source, of any type but the SQL endpoint that comes with a lakehouse. Types pyfabricops does not deploy are pointed out.
UNCHECKED	Fabric returned no definition to compare, so only the item's presence and folder were checked.

Reconciliation.plan holds what would bring the workspace back to the source, as a DeploymentPlan; unmanaged, unchecked and in_sync hold the rest. ok is True when nothing differs and nothing is unmanaged. An unchecked item does not count, as some item types never return a definition.

How definitions are compared

Each item's definition in the source is compared with the one the workspace returns. Fabric rewrites some parts on its own; measured against a workspace, a definition read back right after a deployment differs only in these, which the comparison leaves out:

• the layout: line endings, the end of a file, JSON indentation and key order;
• the logical ID in .platform, which Fabric assigns to each item;
• a report's reference to its semantic model: by path in Git, by connection in the workspace. Both are compared by the model they point to;
• the ref lines Fabric adds to model.tmdl to order a model's tables;
• a part that holds nothing, such as the empty shortcuts.metadata.json Fabric adds to a lakehouse sent without one, when the other side lacks it.

A difference names the parts of the definition that differ.

With a deployment state

The deployment state records a hash of what the last successful deployment sent. With it, a difference tells where it comes from:

• the source is as it was deployed, so the workspace changed: WORKSPACE_DRIFT;
• the source changed since: SOURCE_CHANGED, which the next selective deployment will send.

It tells the same of a folder: moved in the workspace, or in the source.

Restoring what drifted

restore_items() takes the same arguments, and brings the workspace back to the source where it drifted: it reconciles, then applies what undoes the drift, as deploy_all_items() applies a plan.

report = restore_items(
    "Sales-PRD",
    staging,
    start_path=staging,
    state_backend=OneLakeStateBackend("Ops", "DeploymentState"),
    environment="PRD",
)
print(report.describe())

updated  Orders.Notebook  (2.4s)
moved    Utils.Notebook  (0.6s)
created  Sales.Report  (3.1s)
1 created, 1 updated, 1 moved, 0 deleted, 0 failed, 0 skipped in 6.1s

• An item deleted from the workspace is created again (TARGET_MISSING), and one edited or moved there is updated or moved back (WORKSPACE_DRIFT). A report goes back bound to its semantic model's ID.
• An item changed in the source since the last deployment (SOURCE_CHANGED) is left to the next deployment, which records the state. Without a deployment state every difference counts as drift, so the workspace gets the source as it is, pending deployments included: pass the state to leave those alone.
• Nothing is deleted: an unmanaged item is only reported. An item whose definition could not be compared is not touched.
• It holds the lock of the environment, as a deployment does, and never records the state: what goes back is what the last deployment sent.
• It returns a DeploymentReport; a local item that cannot be read is reported as failed.

What it does not do

• reconcile_items() never changes the workspace; restore_items() changes only what drifted. Delete unmanaged items by hand when they should go.
• It reads the definitions one after the other, which takes a while on a large workspace.

Reference

Reconciliation: how a workspace stands against the whole source.

reconcile takes the local items in scope, every item of the workspace, which parts of each item's definition differ between the two (as drift.differing_parts names them), and, when there is one, what the last successful deployment sent. It returns a Reconciliation:

• a plan of what would bring the workspace back to the source: CREATE for an item the workspace lacks (TARGET_MISSING), UPDATE for one whose definition differs, MOVE for one in another folder, BLOCKED for a local item that cannot be read. A difference is WORKSPACE_DRIFT, unless the deployment state shows the source changed since the last deployment (SOURCE_CHANGED): then it is a deployment still to run;
• the items the workspace holds and the source does not (unmanaged);
• the items whose definition could not be compared (unchecked);
• the items that match (in sync).

Like the planner, it calls no API and reads no file, and nothing here changes the workspace: a reconciliation only reports.

Reconciliation dataclass

How a workspace stands against the source.

Attributes:

Name	Type	Description
plan	DeploymentPlan	What would bring the workspace back to the source, one action per item that differs, with its reason and what differs. Items that match are not in it.
unmanaged	Sequence[UnmanagedItem]	The items the workspace holds and the source does not, in the order the workspace lists them. Stored as a tuple.
unchecked	Mapping[ItemKey, str]	For each item whose definition could not be compared, why. It is in the workspace, and in the folder the source says, or it would be in the plan. Read-only.
in_sync	Sequence[ItemKey]	The items that match the workspace. Stored as a tuple.

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

@dataclass(frozen=True)
class Reconciliation:
    """
    How a workspace stands against the source.

    Attributes:
        plan (DeploymentPlan): What would bring the workspace back to the
            source, one action per item that differs, with its reason and
            what differs. Items that match are not in it.
        unmanaged (Sequence[UnmanagedItem]): The items the workspace holds
            and the source does not, in the order the workspace lists them.
            Stored as a tuple.
        unchecked (Mapping[ItemKey, str]): For each item whose definition
            could not be compared, why. It is in the workspace, and in the
            folder the source says, or it would be in the plan. Read-only.
        in_sync (Sequence[ItemKey]): The items that match the workspace.
            Stored as a tuple.
    """

    plan: DeploymentPlan = field(default_factory=DeploymentPlan)
    unmanaged: Sequence[UnmanagedItem] = ()
    unchecked: Mapping[ItemKey, str] = field(default_factory=dict)
    in_sync: Sequence[ItemKey] = ()

    def __post_init__(self) -> None:
        object.__setattr__(self, "unmanaged", tuple(self.unmanaged))
        object.__setattr__(
            self, "unchecked", MappingProxyType(dict(self.unchecked))
        )
        object.__setattr__(self, "in_sync", tuple(self.in_sync))

    @property
    def ok(self) -> bool:
        """
        True when nothing differs: the plan is empty and nothing is
        unmanaged. An unchecked item does not count, as some item types
        never return a definition to compare.
        """
        return not self.plan.actions and not self.unmanaged

    def describe(self) -> str:
        """
        Describe the reconciliation: what differs, then the counts.

        Returns:
            str: One line per action, unmanaged item and unchecked item,
                then a line of counts.
        """
        lines = [action.describe() for action in self.plan.actions]
        lines += [item.describe() for item in self.unmanaged]
        lines += [
            f"UNCHECKED {display_name}.{item_type}: {why}"
            for (item_type, display_name), why in self.unchecked.items()
        ]
        counts = self.plan.summary()
        lines.append(
            f"{len(self.in_sync)} in sync, "
            + ", ".join(
                f"{counts[action.value]} {action.value.lower()}"
                for action in _PLANNED
            )
            + f", {len(self.unmanaged)} unmanaged, "
            f"{len(self.unchecked)} unchecked"
        )
        return "\n".join(lines)

ok property

True when nothing differs: the plan is empty and nothing is unmanaged. An unchecked item does not count, as some item types never return a definition to compare.

describe()

Describe the reconciliation: what differs, then the counts.

Returns:

Name	Type	Description
str	str	One line per action, unmanaged item and unchecked item, then a line of counts.

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

def describe(self) -> str:
    """
    Describe the reconciliation: what differs, then the counts.

    Returns:
        str: One line per action, unmanaged item and unchecked item,
            then a line of counts.
    """
    lines = [action.describe() for action in self.plan.actions]
    lines += [item.describe() for item in self.unmanaged]
    lines += [
        f"UNCHECKED {display_name}.{item_type}: {why}"
        for (item_type, display_name), why in self.unchecked.items()
    ]
    counts = self.plan.summary()
    lines.append(
        f"{len(self.in_sync)} in sync, "
        + ", ".join(
            f"{counts[action.value]} {action.value.lower()}"
            for action in _PLANNED
        )
        + f", {len(self.unmanaged)} unmanaged, "
        f"{len(self.unchecked)} unchecked"
    )
    return "\n".join(lines)

UnmanagedItem dataclass

An item the workspace holds and the source does not.

Attributes:

Name	Type	Description
item_type	str	The Fabric item type.
display_name	str	The item display name.
folder_path	str | None	The workspace folder it is in, or None at the workspace root.
deployable	bool	Whether pyfabricops deploys items of its type.

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

@dataclass(frozen=True)
class UnmanagedItem:
    """
    An item the workspace holds and the source does not.

    Attributes:
        item_type (str): The Fabric item type.
        display_name (str): The item display name.
        folder_path (str | None): The workspace folder it is in, or None at
            the workspace root.
        deployable (bool): Whether pyfabricops deploys items of its type.
    """

    item_type: str
    display_name: str
    folder_path: str | None = None
    deployable: bool = True

    def describe(self) -> str:
        """
        Describe the item on one line.

        Returns:
            str: Such as ``UNMANAGED Draft.Notebook: in the workspace, not
                in the source.``
        """
        where = f" in {_folder(self.folder_path)}" if self.folder_path else ""
        note = "" if self.deployable else "; pyfabricops does not deploy it"
        return (
            f"UNMANAGED {self.display_name}.{self.item_type}: in the "
            f"workspace{where}, not in the source{note}."
        )

describe()

Describe the item on one line.

Returns:

Name	Type	Description
str	str	Such as UNMANAGED Draft.Notebook: in the workspace, not in the source.

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

def describe(self) -> str:
    """
    Describe the item on one line.

    Returns:
        str: Such as ``UNMANAGED Draft.Notebook: in the workspace, not
            in the source.``
    """
    where = f" in {_folder(self.folder_path)}" if self.folder_path else ""
    note = "" if self.deployable else "; pyfabricops does not deploy it"
    return (
        f"UNMANAGED {self.display_name}.{self.item_type}: in the "
        f"workspace{where}, not in the source{note}."
    )

WorkspaceItem dataclass

An item of the workspace, as its listing gives it.

Attributes:

Name	Type	Description
item_type	str	The Fabric item type.
display_name	str	The item display name.
folder_path	str | None	The workspace folder it is in, such as "Sales/Staging", or None at the workspace root.

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

@dataclass(frozen=True)
class WorkspaceItem:
    """
    An item of the workspace, as its listing gives it.

    Attributes:
        item_type (str): The Fabric item type.
        display_name (str): The item display name.
        folder_path (str | None): The workspace folder it is in, such as
            ``"Sales/Staging"``, or None at the workspace root.
    """

    item_type: str
    display_name: str
    folder_path: str | None = None

reconcile(items, workspace, *, differences, unchecked=None, deployed=None, source_keys=None, deployable_types=None, root=None)

Tell how a workspace stands against the source.

Parameters:

Name	Type	Description	Default
items	Iterable[SourceItem]	The local items in scope, in deployment order.	required
workspace	Iterable[WorkspaceItem]	Every item of the workspace.	required
differences	Mapping[ItemKey, Sequence[str]]	For each item in both, the parts of its definition that differ, empty when they match.	required
unchecked	Mapping[ItemKey, str]	For each item in both whose definition could not be compared, why.	None
deployed	Mapping[ItemKey, DeployedItem]	What the last successful deployment sent for each item, from the deployment state. It tells a change made in the workspace from a change made in the source.	None
source_keys	Collection[ItemKey]	Every item of the source, of any type, when items holds only some types: an item of the source is never unmanaged. Defaults to the items.	None
deployable_types	Collection[str]	The item types pyfabricops deploys, to tell unmanaged items of other types. Defaults to any type.	None
root	str	The folder the items were read from. A detail that points to another item folder shows it relative to this one. Defaults to showing it as given.	None

Returns:

Name	Type	Description
Reconciliation	Reconciliation	What differs, what is unmanaged, what could not be compared, and what matches.

Examples:

reconciliation = reconcile(
    [SourceItem("Notebook", "ws/Orders.Notebook", "Orders")],
    [WorkspaceItem("Notebook", "Orders")],
    differences={("Notebook", "Orders"): ["notebook-content.py"]},
)
print(reconciliation.describe())

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

def reconcile(
    items: Iterable[SourceItem],
    workspace: Iterable[WorkspaceItem],
    *,
    differences: Mapping[ItemKey, Sequence[str]],
    unchecked: Mapping[ItemKey, str] | None = None,
    deployed: Mapping[ItemKey, DeployedItem] | None = None,
    source_keys: Collection[ItemKey] | None = None,
    deployable_types: Collection[str] | None = None,
    root: str | None = None,
) -> Reconciliation:
    """
    Tell how a workspace stands against the source.

    Args:
        items (Iterable[SourceItem]): The local items in scope, in
            deployment order.
        workspace (Iterable[WorkspaceItem]): Every item of the workspace.
        differences (Mapping[ItemKey, Sequence[str]]): For each item in
            both, the parts of its definition that differ, empty when they
            match.
        unchecked (Mapping[ItemKey, str], optional): For each item in both
            whose definition could not be compared, why.
        deployed (Mapping[ItemKey, DeployedItem], optional): What the last
            successful deployment sent for each item, from the deployment
            state. It tells a change made in the workspace from a change
            made in the source.
        source_keys (Collection[ItemKey], optional): Every item of the
            source, of any type, when ``items`` holds only some types: an
            item of the source is never unmanaged. Defaults to the items.
        deployable_types (Collection[str], optional): The item types
            pyfabricops deploys, to tell unmanaged items of other types.
            Defaults to any type.
        root (str, optional): The folder the items were read from. A detail
            that points to another item folder shows it relative to this
            one. Defaults to showing it as given.

    Returns:
        Reconciliation: What differs, what is unmanaged, what could not be
            compared, and what matches.

    Examples:
        ```python
        reconciliation = reconcile(
            [SourceItem("Notebook", "ws/Orders.Notebook", "Orders")],
            [WorkspaceItem("Notebook", "Orders")],
            differences={("Notebook", "Orders"): ["notebook-content.py"]},
        )
        print(reconciliation.describe())
        ```
    """
    unchecked = dict(unchecked or {})
    deployed = dict(deployed or {})
    listed = list(workspace)
    found = {(i.item_type, i.display_name): i for i in listed}

    actions: list[DeploymentAction] = []
    not_compared: dict[ItemKey, str] = {}
    in_sync: list[ItemKey] = []
    defined: dict[ItemKey, str] = {}
    for item in items:
        if item.error is not None or item.display_name is None:
            actions.append(
                _action(
                    item,
                    DeploymentActionType.BLOCKED,
                    item.error or f"{item.source_path} has no display name.",
                )
            )
            continue
        key = (item.item_type, item.display_name)
        if key in defined:
            actions.append(
                _action(
                    item,
                    DeploymentActionType.BLOCKED,
                    f"{item.display_name}.{item.item_type} is also defined "
                    f"at {_where(defined[key], root)}; the source cannot "
                    "hold it twice.",
                )
            )
            continue
        defined[key] = item.source_path

        target = found.get(key)
        record = deployed.get(key)
        if target is None:
            problem = name_problem(item.item_type, item.display_name)
            if problem is not None:
                actions.append(
                    _action(
                        item,
                        DeploymentActionType.BLOCKED,
                        f"Fabric refuses this name: {problem}.",
                    )
                )
                continue
            actions.append(
                _action(
                    item,
                    DeploymentActionType.CREATE,
                    "Deleted from the workspace since the last deployment."
                    if record is not None
                    else "In the source, not in the workspace.",
                    reason=DeploymentReason.TARGET_MISSING,
                )
            )
            continue

        parts = differences.get(key)
        if parts is None:
            not_compared[key] = unchecked.get(key, "Definition not compared.")
        moved = (target.folder_path or None) != (item.folder_path or None)
        if parts:
            reason, detail = _content_changed(item, record, tuple(parts))
            if moved:
                detail += (
                    f" It is in {_folder(target.folder_path)} in the "
                    f"workspace, {_folder(item.folder_path)} in the source."
                )
            actions.append(
                _action(item, DeploymentActionType.UPDATE, detail, reason)
            )
        elif moved:
            reason, detail = _folder_changed(item, target, record)
            actions.append(
                _action(item, DeploymentActionType.MOVE, detail, reason)
            )
        elif key not in not_compared:
            in_sync.append(key)

    known = set(defined) | set(source_keys or ())
    unmanaged = [
        UnmanagedItem(
            item.item_type,
            item.display_name,
            item.folder_path,
            deployable_types is None or item.item_type in deployable_types,
        )
        for item in listed
        if (item.item_type, item.display_name) not in known
        and item.item_type not in CHILD_TYPES
    ]
    return Reconciliation(
        plan=DeploymentPlan(actions=actions),
        unmanaged=unmanaged,
        unchecked=not_compared,
        in_sync=in_sync,
    )

The form in which an item of the source and of the workspace are compared.

A reconciliation compares each item's definition in the source, after staging, with the definition the workspace returns. Fabric rewrites some parts on its own, so comparing bytes, or the content hash of what a deployment sends, would report drift where there is none. comparable_parts starts from the canonical form of the content hash (no byte order mark, Unix line endings, JSON by content) and also leaves out what Fabric changes by itself, as measured against a workspace:

• the end of a text part: trailing whitespace and blank lines;
• the logical ID in .platform, which Fabric assigns to each item;
• how a report's definition.pbir points to its semantic model: by the path of its folder in Git, by a connection in the workspace. Given the name of the model it points to, the reference is compared by that name;
• the ref lines Fabric adds to a semantic model's model.tmdl to order its tables and other objects;
• a part that holds nothing, such as the empty shortcuts.metadata.json Fabric adds to a lakehouse sent without one, when the other side lacks it.

Internal for now: nothing here is exported from pyfabricops.

comparable_parts(definition, *, semantic_model=None)

Put each part of a definition in the form used to compare it.

Parameters:

Name	Type	Description	Default
definition	Mapping[str, Any]	The definition, with parts of path and base64 payload: what pack_item_definition builds from the source, or the definition the workspace returns.	required
semantic_model	str	For a report, the display name of the semantic model its definition.pbir points to, however it points to it. Without it, the reference is compared as it is.	None

Returns:

Type	Description
dict[str, bytes]	dict[str, bytes]: The comparable content of each part, by path.

Examples:

source = comparable_parts(pack_item_definition(report_folder),
                          semantic_model="Sales")

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

def comparable_parts(
    definition: Mapping[str, Any],
    *,
    semantic_model: str | None = None,
) -> dict[str, bytes]:
    """
    Put each part of a definition in the form used to compare it.

    Args:
        definition (Mapping[str, Any]): The definition, with ``parts`` of
            ``path`` and base64 ``payload``: what ``pack_item_definition``
            builds from the source, or the ``definition`` the workspace
            returns.
        semantic_model (str, optional): For a report, the display name of
            the semantic model its ``definition.pbir`` points to, however it
            points to it. Without it, the reference is compared as it is.

    Returns:
        dict[str, bytes]: The comparable content of each part, by path.

    Examples:
        ```python
        source = comparable_parts(pack_item_definition(report_folder),
                                  semantic_model="Sales")
        ```
    """
    parts: dict[str, bytes] = {}
    for part in definition.get("parts", []):
        path = part["path"]
        content = canonical_part(path, base64.b64decode(part["payload"]))
        parts[path] = _comparable(path, content, semantic_model)
    return parts

differing_parts(source, workspace)

Name the parts that differ between two definitions.

A part that only one side has differs, unless it holds nothing (empty, [] or {}), as the parts Fabric adds by itself do. .platform is left out when one side has none, as not every item type returns one.

Parameters:

Name	Type	Description	Default
source	Mapping[str, bytes]	The comparable parts of the item in the source.	required
workspace	Mapping[str, bytes]	The comparable parts of the item in the workspace.	required

Returns:

Type	Description
tuple[str, ...]	tuple[str, ...]: The paths of the parts that differ, in order; empty when the definitions match.

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

def differing_parts(
    source: Mapping[str, bytes], workspace: Mapping[str, bytes]
) -> tuple[str, ...]:
    """
    Name the parts that differ between two definitions.

    A part that only one side has differs, unless it holds nothing (empty,
    ``[]`` or ``{}``), as the parts Fabric adds by itself do. ``.platform``
    is left out when one side has none, as not every item type returns one.

    Args:
        source (Mapping[str, bytes]): The comparable parts of the item in
            the source.
        workspace (Mapping[str, bytes]): The comparable parts of the item in
            the workspace.

    Returns:
        tuple[str, ...]: The paths of the parts that differ, in order; empty
            when the definitions match.
    """
    paths = set(source) | set(workspace)
    if _PLATFORM not in source or _PLATFORM not in workspace:
        paths.discard(_PLATFORM)
    return tuple(
        sorted(
            path
            for path in paths
            if source.get(path) != workspace.get(path)
            and not _empty_on_one_side(source.get(path), workspace.get(path))
        )
    )