Deployment State

The deployment state records what the last successful deployment to an environment sent: the commit of each item type, and the hash and folder of each item. With it, deploy_all_items() deploys only what changed since then, and reconcile_items() tells drift in the workspace from a deployment still to run. It is written only when every item of a run succeeded, and it holds no secret.

A state backend keeps the states, one per environment. Pass it to deploy_all_items(), plan_all_items() or reconcile_items() with the environment's name:

report = deploy_all_items(
    "Sales-PRD",
    staging,
    start_path=staging,
    repository_path="workspace",
    state_backend=OneLakeStateBackend("Ops", "DeploymentState"),
    environment="PRD",
)

In a OneLake lakehouse

OneLakeStateBackend(workspace, lakehouse) keeps each state as Files/pyfabricops/state/<environment>.json in a lakehouse, and survives every CI run. Keep the lakehouse in a workspace of its own, such as one for operations: in the workspace deployed to, reconcile_items() would report it as unmanaged.

The identity that deploys needs to write to that lakehouse, as a Contributor of its workspace. Its token for OneLake comes from set_auth_provider(), as for the Fabric API.
The workspace and the lakehouse are given by name or ID, looked up once, then reached by ID.
folder changes the folder under Files. endpoint takes a regional endpoint, such as https://westus-onelake.blob.fabric.microsoft.com, so the state stays in the region of the lakehouse's capacity.
A state is saved only over the one the run read. When another run saved one in between, the save fails and the other run's state is kept:

Deployment state 'PRD' changed in Ops/DeploymentState/Files/pyfabricops/state/PRD.json since this run read it, so it was not overwritten: another run deployed to the environment meanwhile.

In a local folder

LocalJsonStateBackend(folder) keeps each state as <folder>/<environment>.json. Keep the folder between runs: without it, a run deploys every item, which is safe but slower. A CI cache is not enough for that, as GitHub drops a cache unused for 7 days, and Azure DevOps keeps one per pipeline.

Locks

A deployment holds the lock of its environment from before it reads the state until after it records it, so two runs never deploy to one environment at a time. Both backends keep the lock next to the state, as <environment>.lock, which says who holds it and until when:

DeploymentLockedError: Deployment state 'PRD' is locked, held by runner@ci-host (GitHub Actions run 1234) since 2026-09-25T10:00:00Z, until 2026-09-25T12:00:00Z. Wait for that run to finish; if it is gone, call force_unlock('PRD') on the state backend.

A run fails at once when another holds the lock, before it reads or deploys anything. lock_timeout=<seconds> on the backend waits for the lock instead.
A lock holds for lock_ttl seconds, two hours by default, unless its run releases it; a failed run releases it too. After that another run takes it over, so a run that died holding it blocks the environment for no longer.
force_unlock(environment) removes the lock whoever holds it, for a run that is gone. Make sure no run is deploying first.

OneLakeStateBackend("Ops", "DeploymentState").force_unlock("PRD")

plan_all_items() and reconcile_items() only read the state, and take no lock.
A lock holds no secret: the user and host that took it, the CI run when there is one, and when it was taken.

Journal and resume

A deployment writes the journal of its run next to the state, as <environment>.journal.json, each time an item ends: what it did to the item, and the hash and folder it sent. When a run fails, or dies, before it records the state, the next run knows what it already sent, and skips each item still in the workspace that has not changed since:

NOOP     Orders.Notebook  SOURCE_CHANGED: Definition and folder unchanged since an interrupted run sent it at 2026-09-26T10:04:12Z.
UPDATE   Daily.DataPipeline  SOURCE_CHANGED

The state is still recorded only when a whole run succeeds; the journal never takes its place. The run that succeeds records what the interrupted runs sent, too.
The journal holds the last run. A run that resumes another carries its entries over, so they are not lost if it is interrupted too.
Each item costs one small write of the journal, a request of its own in OneLake. A journal that cannot be written costs only the resume: the run warns and goes on.
plan_all_items() shows the resume; it writes no journal.

Other backends

Any object with load(environment) and save(environment, state) is a DeploymentStateBackend. One that also has lock(environment) and force_unlock(environment) is a LockingStateBackend, and deployments hold its lock; one with load_journal(environment) and save_journal(environment, journal) is a JournalingStateBackend, and deployments keep their journal there. One without them works as well, without a lock or a resume.

Reference

Deployment state: what was last deployed successfully to an environment.

A DeploymentState records the workspace an environment targets, the last commit deployed successfully for each item type, so the next deployment of a type compares only what changed since then, and the hash and folder last sent for each item, so an item sent unchanged needs nothing. It is written only after a deployment in which every item succeeded, and it never holds secrets.

DeploymentStateBackend is where states are kept; the engine knows nothing more. LocalJsonStateBackend keeps them as JSON files in a local folder, and OneLakeStateBackend, in pyfabricops.helpers.onelake_state, in the Files of a lakehouse, where they outlive any CI run.

A backend that also locks (LockingStateBackend) keeps two runs from deploying to one environment at a time: a run holds a DeploymentLock from before it reads the state until after it records it. A lock has a validity, after which another run may take it over, so a run that died holding it does not block the environment for good.

A backend that also keeps journals (JournalingStateBackend) holds the DeploymentJournal of the last run of each environment: what it did, item by item, written as it went. The state is still recorded only when a whole run succeeds; the journal tells the next run what a run that failed, or died, already sent.

`DeploymentJournal` `dataclass`

The journal of a deployment run: what it did, item by item.

A run writes its journal as it goes, next to the state, so that when it fails, or dies, before it records the state, the next run knows what it already sent. The journal never takes the place of the state, which is recorded only when a whole run succeeds. It holds no secret.

Attributes:

Name	Type	Description
`environment`	`str`	The environment the run deployed to.
`workspace`	`str`	The workspace name or ID, as given to `deploy_all_items`.
`run_id`	`str`	A random ID of the run.
`source_commit`	`str`	The commit the run deployed.
`started_at_utc`	`str`	When the run started, as `YYYY-MM-DDTHH:MM:SSZ`.
`entries`	`Sequence[JournalEntry]`	What each action did, in the order the actions ended; the first ones may come from the interrupted run this one resumed. Stored as a tuple.
`finished_at_utc`	`str \| None`	When the run ended, or None while it runs, or when it died.
`ok`	`bool \| None`	Whether every item succeeded, once the run ended.

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

@dataclass(frozen=True)
class DeploymentJournal:
    """
    The journal of a deployment run: what it did, item by item.

    A run writes its journal as it goes, next to the state, so that when it
    fails, or dies, before it records the state, the next run knows what it
    already sent. The journal never takes the place of the state, which is
    recorded only when a whole run succeeds. It holds no secret.

    Attributes:
        environment (str): The environment the run deployed to.
        workspace (str): The workspace name or ID, as given to
            ``deploy_all_items``.
        run_id (str): A random ID of the run.
        source_commit (str): The commit the run deployed.
        started_at_utc (str): When the run started, as
            ``YYYY-MM-DDTHH:MM:SSZ``.
        entries (Sequence[JournalEntry]): What each action did, in the order
            the actions ended; the first ones may come from the interrupted
            run this one resumed. Stored as a tuple.
        finished_at_utc (str | None): When the run ended, or None while it
            runs, or when it died.
        ok (bool | None): Whether every item succeeded, once the run ended.
    """

    environment: str
    workspace: str
    run_id: str
    source_commit: str
    started_at_utc: str
    entries: Sequence[JournalEntry] = ()
    finished_at_utc: str | None = None
    ok: bool | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "entries", tuple(self.entries))

    @classmethod
    def start(
        cls,
        environment: str,
        workspace: str,
        source_commit: str,
        *,
        resumed: DeploymentJournal | None = None,
    ) -> DeploymentJournal:
        """
        Begin the journal of a run.

        Args:
            environment (str): The environment of the run.
            workspace (str): The workspace, as given to ``deploy_all_items``.
            source_commit (str): The commit the run deploys.
            resumed (DeploymentJournal, optional): The journal of the
                interrupted run this one resumes: its entries are carried
                over, so that what it sent is not lost if this run is
                interrupted too.

        Returns:
            DeploymentJournal: The journal, with no finish yet.
        """
        return cls(
            environment=environment,
            workspace=workspace,
            run_id=uuid.uuid4().hex,
            source_commit=source_commit,
            started_at_utc=_utc_text(datetime.now(timezone.utc)),
            entries=resumed.entries if resumed is not None else (),
        )

    @property
    def interrupted(self) -> bool:
        """Whether the run failed, or ended before finishing."""
        return self.ok is not True

    def with_entry(self, entry: JournalEntry) -> DeploymentJournal:
        """
        Add what one more action did.

        Args:
            entry (JournalEntry): The entry.

        Returns:
            DeploymentJournal: A journal with the entry last.
        """
        return replace(self, entries=(*self.entries, entry))

    def finish(self, ok: bool) -> DeploymentJournal:
        """
        Mark the run ended.

        Args:
            ok (bool): Whether every item succeeded.

        Returns:
            DeploymentJournal: A journal with its finish.
        """
        return replace(
            self,
            finished_at_utc=_utc_text(datetime.now(timezone.utc)),
            ok=ok,
        )

    def sent(self) -> dict[tuple[str, str], DeployedItem]:
        """
        Return what the journaled runs sent successfully, by item.

        Returns:
            dict[tuple[str, str], DeployedItem]: For each item created,
                updated or moved, the hash and folder it was sent with, as
                of its last entry; an item deleted since is left out.
        """
        sent: dict[tuple[str, str], DeployedItem] = {}
        for entry in self.entries:
            key = (entry.item_type, entry.display_name)
            if entry.outcome in _SENT and entry.content_hash:
                sent[key] = DeployedItem(
                    entry.content_hash,
                    entry.folder_path,
                    sent_by=f"an interrupted run sent it at {entry.at_utc}",
                )
            elif entry.outcome == "deleted":
                sent.pop(key, None)
        return sent

    def to_dict(self) -> dict[str, Any]:
        """
        Return the journal as JSON-ready data.

        Returns:
            dict[str, Any]: The fields, with each entry as a dict.
        """
        return {
            **{name: getattr(self, name) for name in _JOURNAL_FIELDS},
            "finished_at_utc": self.finished_at_utc,
            "ok": self.ok,
            "entries": [asdict(entry) for entry in self.entries],
        }

    @classmethod
    def from_dict(cls, data: Any, *, source: str) -> DeploymentJournal:
        """
        Read a journal from JSON data.

        Args:
            data (Any): The parsed JSON.
            source (str): Where the data came from, for error messages.

        Returns:
            DeploymentJournal: The journal.

        Raises:
            ConfigurationError: If the data is not a journal.
        """
        if not isinstance(data, dict) or not all(
            isinstance(data.get(name), str) and data[name]
            for name in _JOURNAL_FIELDS
        ):
            raise ConfigurationError(f"{source} is not a deployment journal.")
        finished, ok = data.get("finished_at_utc"), data.get("ok")
        raw = data.get("entries")
        if not (
            (finished is None or isinstance(finished, str))
            and (ok is None or isinstance(ok, bool))
            and isinstance(raw, list)
        ):
            raise ConfigurationError(f"{source} is not a deployment journal.")
        return cls(
            **{name: data[name] for name in _JOURNAL_FIELDS},
            entries=[_read_entry(entry, source) for entry in raw],
            finished_at_utc=finished,
            ok=ok,
        )

`interrupted` `property`

Whether the run failed, or ended before finishing.

`finish(ok)`

Mark the run ended.

Parameters:

Name	Type	Description	Default
`ok`	`bool`	Whether every item succeeded.	required

Returns:

Name	Type	Description
`DeploymentJournal`	`DeploymentJournal`	A journal with its finish.

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

def finish(self, ok: bool) -> DeploymentJournal:
    """
    Mark the run ended.

    Args:
        ok (bool): Whether every item succeeded.

    Returns:
        DeploymentJournal: A journal with its finish.
    """
    return replace(
        self,
        finished_at_utc=_utc_text(datetime.now(timezone.utc)),
        ok=ok,
    )

`from_dict(data, *, source)` `classmethod`

Read a journal from JSON data.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The parsed JSON.	required
`source`	`str`	Where the data came from, for error messages.	required

Returns:

Name	Type	Description
`DeploymentJournal`	`DeploymentJournal`	The journal.

Raises:

Type	Description
`ConfigurationError`	If the data is not a journal.

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

@classmethod
def from_dict(cls, data: Any, *, source: str) -> DeploymentJournal:
    """
    Read a journal from JSON data.

    Args:
        data (Any): The parsed JSON.
        source (str): Where the data came from, for error messages.

    Returns:
        DeploymentJournal: The journal.

    Raises:
        ConfigurationError: If the data is not a journal.
    """
    if not isinstance(data, dict) or not all(
        isinstance(data.get(name), str) and data[name]
        for name in _JOURNAL_FIELDS
    ):
        raise ConfigurationError(f"{source} is not a deployment journal.")
    finished, ok = data.get("finished_at_utc"), data.get("ok")
    raw = data.get("entries")
    if not (
        (finished is None or isinstance(finished, str))
        and (ok is None or isinstance(ok, bool))
        and isinstance(raw, list)
    ):
        raise ConfigurationError(f"{source} is not a deployment journal.")
    return cls(
        **{name: data[name] for name in _JOURNAL_FIELDS},
        entries=[_read_entry(entry, source) for entry in raw],
        finished_at_utc=finished,
        ok=ok,
    )

`sent()`

Return what the journaled runs sent successfully, by item.

Returns:

Type	Description
`dict[tuple[str, str], DeployedItem]`	dict[tuple[str, str], DeployedItem]: For each item created, updated or moved, the hash and folder it was sent with, as of its last entry; an item deleted since is left out.

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

def sent(self) -> dict[tuple[str, str], DeployedItem]:
    """
    Return what the journaled runs sent successfully, by item.

    Returns:
        dict[tuple[str, str], DeployedItem]: For each item created,
            updated or moved, the hash and folder it was sent with, as
            of its last entry; an item deleted since is left out.
    """
    sent: dict[tuple[str, str], DeployedItem] = {}
    for entry in self.entries:
        key = (entry.item_type, entry.display_name)
        if entry.outcome in _SENT and entry.content_hash:
            sent[key] = DeployedItem(
                entry.content_hash,
                entry.folder_path,
                sent_by=f"an interrupted run sent it at {entry.at_utc}",
            )
        elif entry.outcome == "deleted":
            sent.pop(key, None)
    return sent

`start(environment, workspace, source_commit, *, resumed=None)` `classmethod`

Begin the journal of a run.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment of the run.	required
`workspace`	`str`	The workspace, as given to `deploy_all_items`.	required
`source_commit`	`str`	The commit the run deploys.	required
`resumed`	`DeploymentJournal`	The journal of the interrupted run this one resumes: its entries are carried over, so that what it sent is not lost if this run is interrupted too.	`None`

Returns:

Name	Type	Description
`DeploymentJournal`	`DeploymentJournal`	The journal, with no finish yet.

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

@classmethod
def start(
    cls,
    environment: str,
    workspace: str,
    source_commit: str,
    *,
    resumed: DeploymentJournal | None = None,
) -> DeploymentJournal:
    """
    Begin the journal of a run.

    Args:
        environment (str): The environment of the run.
        workspace (str): The workspace, as given to ``deploy_all_items``.
        source_commit (str): The commit the run deploys.
        resumed (DeploymentJournal, optional): The journal of the
            interrupted run this one resumes: its entries are carried
            over, so that what it sent is not lost if this run is
            interrupted too.

    Returns:
        DeploymentJournal: The journal, with no finish yet.
    """
    return cls(
        environment=environment,
        workspace=workspace,
        run_id=uuid.uuid4().hex,
        source_commit=source_commit,
        started_at_utc=_utc_text(datetime.now(timezone.utc)),
        entries=resumed.entries if resumed is not None else (),
    )

`to_dict()`

Return the journal as JSON-ready data.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The fields, with each entry as a dict.

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

def to_dict(self) -> dict[str, Any]:
    """
    Return the journal as JSON-ready data.

    Returns:
        dict[str, Any]: The fields, with each entry as a dict.
    """
    return {
        **{name: getattr(self, name) for name in _JOURNAL_FIELDS},
        "finished_at_utc": self.finished_at_utc,
        "ok": self.ok,
        "entries": [asdict(entry) for entry in self.entries],
    }

`with_entry(entry)`

Add what one more action did.

Parameters:

Name	Type	Description	Default
`entry`	`JournalEntry`	The entry.	required

Returns:

Name	Type	Description
`DeploymentJournal`	`DeploymentJournal`	A journal with the entry last.

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

def with_entry(self, entry: JournalEntry) -> DeploymentJournal:
    """
    Add what one more action did.

    Args:
        entry (JournalEntry): The entry.

    Returns:
        DeploymentJournal: A journal with the entry last.
    """
    return replace(self, entries=(*self.entries, entry))

`DeploymentLock` `dataclass`

The lock a deployment run holds on the state of an environment.

A lock holds no secret: who took it, when, and until when it holds.

Attributes:

Name	Type	Description
`environment`	`str`	The environment whose state is locked.
`lock_id`	`str`	A random ID, which tells this lock from any other.
`holder`	`str`	Who took it: `user@host`, and the CI run when there is one.
`acquired_at_utc`	`str`	When it was taken, as `YYYY-MM-DDTHH:MM:SSZ`.
`expires_at_utc`	`str`	When another run may take it over, if its run has not released it.

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

@dataclass(frozen=True)
class DeploymentLock:
    """
    The lock a deployment run holds on the state of an environment.

    A lock holds no secret: who took it, when, and until when it holds.

    Attributes:
        environment (str): The environment whose state is locked.
        lock_id (str): A random ID, which tells this lock from any other.
        holder (str): Who took it: ``user@host``, and the CI run when there
            is one.
        acquired_at_utc (str): When it was taken, as
            ``YYYY-MM-DDTHH:MM:SSZ``.
        expires_at_utc (str): When another run may take it over, if its run
            has not released it.
    """

    environment: str
    lock_id: str
    holder: str
    acquired_at_utc: str
    expires_at_utc: str

    @classmethod
    def new(cls, environment: str, ttl_seconds: float) -> DeploymentLock:
        """
        Make a lock for this process, valid for ``ttl_seconds``.

        Args:
            environment (str): The environment to lock.
            ttl_seconds (float): How long the lock holds unless released.

        Returns:
            DeploymentLock: The lock, not taken yet.
        """
        now = datetime.now(timezone.utc).replace(microsecond=0)
        return cls(
            environment=environment,
            lock_id=uuid.uuid4().hex,
            holder=_holder(),
            acquired_at_utc=_utc_text(now),
            expires_at_utc=_utc_text(now + timedelta(seconds=ttl_seconds)),
        )

    @property
    def expired(self) -> bool:
        """Whether another run may take the lock over."""
        return _parse_utc(self.expires_at_utc) <= datetime.now(timezone.utc)

    def describe(self) -> str:
        """
        Say who holds the lock, since when and until when.

        Returns:
            str: Such as ``held by ci@runner since 2026-09-25T10:00:00Z,
                until 2026-09-25T12:00:00Z``.
        """
        return (
            f"held by {self.holder} since {self.acquired_at_utc}, until "
            f"{self.expires_at_utc}"
        )

    def to_dict(self) -> dict[str, str]:
        """
        Return the lock as JSON-ready data.

        Returns:
            dict[str, str]: The fields.
        """
        return asdict(self)

    @classmethod
    def from_dict(cls, data: Any, *, source: str) -> DeploymentLock:
        """
        Read a lock from JSON data.

        Args:
            data (Any): The parsed JSON.
            source (str): Where the data came from, for error messages.

        Returns:
            DeploymentLock: The lock.

        Raises:
            ConfigurationError: If the data is not a lock.
        """
        names = ("environment", "lock_id", "holder")
        times = ("acquired_at_utc", "expires_at_utc")
        if not isinstance(data, dict) or not all(
            isinstance(data.get(name), str) and data[name]
            for name in (*names, *times)
        ):
            raise ConfigurationError(f"{source} is not a deployment lock.")
        try:
            for name in times:
                _parse_utc(data[name])
        except ValueError as e:
            raise ConfigurationError(
                f"{source} is not a deployment lock."
            ) from e
        return cls(**{name: data[name] for name in (*names, *times)})

`expired` `property`

Whether another run may take the lock over.

`describe()`

Say who holds the lock, since when and until when.

Returns:

Name	Type	Description
`str`	`str`	Such as `held by ci@runner since 2026-09-25T10:00:00Z, until 2026-09-25T12:00:00Z`.

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

def describe(self) -> str:
    """
    Say who holds the lock, since when and until when.

    Returns:
        str: Such as ``held by ci@runner since 2026-09-25T10:00:00Z,
            until 2026-09-25T12:00:00Z``.
    """
    return (
        f"held by {self.holder} since {self.acquired_at_utc}, until "
        f"{self.expires_at_utc}"
    )

`from_dict(data, *, source)` `classmethod`

Read a lock from JSON data.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The parsed JSON.	required
`source`	`str`	Where the data came from, for error messages.	required

Returns:

Name	Type	Description
`DeploymentLock`	`DeploymentLock`	The lock.

Raises:

Type	Description
`ConfigurationError`	If the data is not a lock.

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

@classmethod
def from_dict(cls, data: Any, *, source: str) -> DeploymentLock:
    """
    Read a lock from JSON data.

    Args:
        data (Any): The parsed JSON.
        source (str): Where the data came from, for error messages.

    Returns:
        DeploymentLock: The lock.

    Raises:
        ConfigurationError: If the data is not a lock.
    """
    names = ("environment", "lock_id", "holder")
    times = ("acquired_at_utc", "expires_at_utc")
    if not isinstance(data, dict) or not all(
        isinstance(data.get(name), str) and data[name]
        for name in (*names, *times)
    ):
        raise ConfigurationError(f"{source} is not a deployment lock.")
    try:
        for name in times:
            _parse_utc(data[name])
    except ValueError as e:
        raise ConfigurationError(
            f"{source} is not a deployment lock."
        ) from e
    return cls(**{name: data[name] for name in (*names, *times)})

`new(environment, ttl_seconds)` `classmethod`

Make a lock for this process, valid for ttl_seconds.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment to lock.	required
`ttl_seconds`	`float`	How long the lock holds unless released.	required

Returns:

Name	Type	Description
`DeploymentLock`	`DeploymentLock`	The lock, not taken yet.

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

@classmethod
def new(cls, environment: str, ttl_seconds: float) -> DeploymentLock:
    """
    Make a lock for this process, valid for ``ttl_seconds``.

    Args:
        environment (str): The environment to lock.
        ttl_seconds (float): How long the lock holds unless released.

    Returns:
        DeploymentLock: The lock, not taken yet.
    """
    now = datetime.now(timezone.utc).replace(microsecond=0)
    return cls(
        environment=environment,
        lock_id=uuid.uuid4().hex,
        holder=_holder(),
        acquired_at_utc=_utc_text(now),
        expires_at_utc=_utc_text(now + timedelta(seconds=ttl_seconds)),
    )

`to_dict()`

Return the lock as JSON-ready data.

Returns:

Type	Description
`dict[str, str]`	dict[str, str]: The fields.

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

def to_dict(self) -> dict[str, str]:
    """
    Return the lock as JSON-ready data.

    Returns:
        dict[str, str]: The fields.
    """
    return asdict(self)

`DeploymentState` `dataclass`

The last successful deployment to an environment.

Attributes:

Name	Type	Description
`environment`	`str`	The environment the state is kept under.
`workspace`	`str`	The workspace name or ID the deployments target, as given to `deploy_all_items`.
`workspace_id`	`str`	The resolved workspace ID.
`source_commit`	`str`	The commit the last successful deployment deployed.
`commits`	`Mapping[str, str]`	The last commit deployed successfully for each item type: where its next deployment compares from. Read-only.
`deployed_at_utc`	`str`	When the last successful deployment finished, as `YYYY-MM-DDTHH:MM:SSZ`.
`items`	`Mapping[tuple[str, str], DeployedItem]`	What was last sent for each item, by `(item_type, display_name)`: the hash of its definition and its folder. Read-only.
`schema_version`	`int`	The layout of the stored state.

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

@dataclass(frozen=True)
class DeploymentState:
    """
    The last successful deployment to an environment.

    Attributes:
        environment (str): The environment the state is kept under.
        workspace (str): The workspace name or ID the deployments target, as
            given to ``deploy_all_items``.
        workspace_id (str): The resolved workspace ID.
        source_commit (str): The commit the last successful deployment
            deployed.
        commits (Mapping[str, str]): The last commit deployed successfully
            for each item type: where its next deployment compares from.
            Read-only.
        deployed_at_utc (str): When the last successful deployment finished,
            as ``YYYY-MM-DDTHH:MM:SSZ``.
        items (Mapping[tuple[str, str], DeployedItem]): What was last sent
            for each item, by ``(item_type, display_name)``: the hash of its
            definition and its folder. Read-only.
        schema_version (int): The layout of the stored state.
    """

    environment: str
    workspace: str
    workspace_id: str
    source_commit: str
    commits: Mapping[str, str]
    deployed_at_utc: str
    items: Mapping[tuple[str, str], DeployedItem] = field(default_factory=dict)
    schema_version: int = _SCHEMA_VERSION

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

    def targets(self, workspace: str) -> bool:
        """
        Tell whether the state was recorded for a workspace.

        Args:
            workspace (str): A workspace name or ID.

        Returns:
            bool: True when it is the workspace given or its resolved ID.
        """
        return workspace in (self.workspace, self.workspace_id)

    def to_dict(self) -> dict[str, Any]:
        """
        Return the state as JSON-ready data.

        Returns:
            dict[str, Any]: The fields, with ``commits`` as a plain dict and
                ``items`` keyed ``"<item_type>.<display_name>"``.
        """
        return {
            "schema_version": self.schema_version,
            **{name: getattr(self, name) for name in _TEXT_FIELDS},
            "commits": dict(self.commits),
            "items": {
                f"{item_type}.{display_name}": {
                    "item_type": item_type,
                    "display_name": display_name,
                    "content_hash": deployed.content_hash,
                    "folder_path": deployed.folder_path,
                }
                for (item_type, display_name), deployed in self.items.items()
            },
        }

    @classmethod
    def from_dict(cls, data: Any, *, source: str) -> DeploymentState:
        """
        Read a state from JSON data.

        Args:
            data (Any): The parsed JSON.
            source (str): Where the data came from, for error messages.

        Returns:
            DeploymentState: The state.

        Raises:
            ConfigurationError: If the data is not a state this version of
                pyfabricops can read.
        """
        if not isinstance(data, dict):
            raise ConfigurationError(f"{source} is not a deployment state.")
        version = data.get("schema_version")
        if version != _SCHEMA_VERSION:
            raise ConfigurationError(
                f"{source} has schema version {version!r}; this version of "
                f"pyfabricops reads version {_SCHEMA_VERSION}."
            )

        text: dict[str, str] = {}
        for name in _TEXT_FIELDS:
            value = data.get(name)
            if not isinstance(value, str) or not value:
                raise ConfigurationError(f"{source} has no valid {name}.")
            text[name] = value

        commits = data.get("commits")
        if not isinstance(commits, dict) or not all(
            isinstance(k, str) and isinstance(v, str) and v
            for k, v in commits.items()
        ):
            raise ConfigurationError(f"{source} has no valid commits.")

        return cls(
            environment=text["environment"],
            workspace=text["workspace"],
            workspace_id=text["workspace_id"],
            source_commit=text["source_commit"],
            commits=commits,
            deployed_at_utc=text["deployed_at_utc"],
            items=_read_items(data.get("items", {}), source),
        )

`from_dict(data, *, source)` `classmethod`

Read a state from JSON data.

Parameters:

Name	Type	Description	Default
`data`	`Any`	The parsed JSON.	required
`source`	`str`	Where the data came from, for error messages.	required

Returns:

Name	Type	Description
`DeploymentState`	`DeploymentState`	The state.

Raises:

Type	Description
`ConfigurationError`	If the data is not a state this version of pyfabricops can read.

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

@classmethod
def from_dict(cls, data: Any, *, source: str) -> DeploymentState:
    """
    Read a state from JSON data.

    Args:
        data (Any): The parsed JSON.
        source (str): Where the data came from, for error messages.

    Returns:
        DeploymentState: The state.

    Raises:
        ConfigurationError: If the data is not a state this version of
            pyfabricops can read.
    """
    if not isinstance(data, dict):
        raise ConfigurationError(f"{source} is not a deployment state.")
    version = data.get("schema_version")
    if version != _SCHEMA_VERSION:
        raise ConfigurationError(
            f"{source} has schema version {version!r}; this version of "
            f"pyfabricops reads version {_SCHEMA_VERSION}."
        )

    text: dict[str, str] = {}
    for name in _TEXT_FIELDS:
        value = data.get(name)
        if not isinstance(value, str) or not value:
            raise ConfigurationError(f"{source} has no valid {name}.")
        text[name] = value

    commits = data.get("commits")
    if not isinstance(commits, dict) or not all(
        isinstance(k, str) and isinstance(v, str) and v
        for k, v in commits.items()
    ):
        raise ConfigurationError(f"{source} has no valid commits.")

    return cls(
        environment=text["environment"],
        workspace=text["workspace"],
        workspace_id=text["workspace_id"],
        source_commit=text["source_commit"],
        commits=commits,
        deployed_at_utc=text["deployed_at_utc"],
        items=_read_items(data.get("items", {}), source),
    )

`targets(workspace)`

Tell whether the state was recorded for a workspace.

Parameters:

Name	Type	Description	Default
`workspace`	`str`	A workspace name or ID.	required

Returns:

Name	Type	Description
`bool`	`bool`	True when it is the workspace given or its resolved ID.

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

def targets(self, workspace: str) -> bool:
    """
    Tell whether the state was recorded for a workspace.

    Args:
        workspace (str): A workspace name or ID.

    Returns:
        bool: True when it is the workspace given or its resolved ID.
    """
    return workspace in (self.workspace, self.workspace_id)

`to_dict()`

Return the state as JSON-ready data.

Returns:

Type	Description
`dict[str, Any]`	dict[str, Any]: The fields, with `commits` as a plain dict and `items` keyed `"<item_type>.<display_name>"`.

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

def to_dict(self) -> dict[str, Any]:
    """
    Return the state as JSON-ready data.

    Returns:
        dict[str, Any]: The fields, with ``commits`` as a plain dict and
            ``items`` keyed ``"<item_type>.<display_name>"``.
    """
    return {
        "schema_version": self.schema_version,
        **{name: getattr(self, name) for name in _TEXT_FIELDS},
        "commits": dict(self.commits),
        "items": {
            f"{item_type}.{display_name}": {
                "item_type": item_type,
                "display_name": display_name,
                "content_hash": deployed.content_hash,
                "folder_path": deployed.folder_path,
            }
            for (item_type, display_name), deployed in self.items.items()
        },
    }

`DeploymentStateBackend`

Bases: Protocol

Where deployment states are kept, one per environment.

Any object with these two methods is a backend, so states can live in a local folder, a blob container or a pipeline artifact without the engine knowing. A backend must never store secrets; a state has none.

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

class DeploymentStateBackend(Protocol):
    """
    Where deployment states are kept, one per environment.

    Any object with these two methods is a backend, so states can live in a
    local folder, a blob container or a pipeline artifact without the
    engine knowing. A backend must never store secrets; a state has none.
    """

    def load(self, environment: str) -> DeploymentState | None:
        """Return the state of an environment, or None when it has none."""
        ...

    def save(self, environment: str, state: DeploymentState) -> None:
        """Store the state of an environment, replacing the previous one."""
        ...

`load(environment)`

Return the state of an environment, or None when it has none.

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

def load(self, environment: str) -> DeploymentState | None:
    """Return the state of an environment, or None when it has none."""
    ...

`save(environment, state)`

Store the state of an environment, replacing the previous one.

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

def save(self, environment: str, state: DeploymentState) -> None:
    """Store the state of an environment, replacing the previous one."""
    ...

`JournalEntry` `dataclass`

What one action of a deployment run did.

Attributes:

Name	Type	Description
`item_type`	`str`	The Fabric item type.
`display_name`	`str`	The item display name.
`outcome`	`str`	`"created"`, `"updated"`, `"moved"`, `"deleted"`, `"failed"` or `"skipped"`, as the report says.
`at_utc`	`str`	When the action ended, as `YYYY-MM-DDTHH:MM:SSZ`.
`content_hash`	`str \| None`	The hash of the definition the action sent, when known.
`folder_path`	`str \| None`	The workspace folder the item belongs in, or None for the workspace root.

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

@dataclass(frozen=True)
class JournalEntry:
    """
    What one action of a deployment run did.

    Attributes:
        item_type (str): The Fabric item type.
        display_name (str): The item display name.
        outcome (str): ``"created"``, ``"updated"``, ``"moved"``,
            ``"deleted"``, ``"failed"`` or ``"skipped"``, as the report says.
        at_utc (str): When the action ended, as ``YYYY-MM-DDTHH:MM:SSZ``.
        content_hash (str | None): The hash of the definition the action
            sent, when known.
        folder_path (str | None): The workspace folder the item belongs in,
            or None for the workspace root.
    """

    item_type: str
    display_name: str
    outcome: str
    at_utc: str
    content_hash: str | None = None
    folder_path: str | None = None

`JournalingStateBackend`

Bases: DeploymentStateBackend, Protocol

A state backend that also keeps the journal of each environment's last run.

deploy_all_items writes the journal as each item ends. The backends of pyfabricops keep journals; a backend without these methods still works, and a run that fails is then not resumed.

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

@runtime_checkable
class JournalingStateBackend(DeploymentStateBackend, Protocol):
    """
    A state backend that also keeps the journal of each environment's last
    run.

    ``deploy_all_items`` writes the journal as each item ends. The backends
    of pyfabricops keep journals; a backend without these methods still
    works, and a run that fails is then not resumed.
    """

    def load_journal(self, environment: str) -> DeploymentJournal | None:
        """Return the journal of the last run, or None without one."""
        ...

    def save_journal(
        self, environment: str, journal: DeploymentJournal
    ) -> None:
        """Store the journal of a run, replacing the one before."""
        ...

`load_journal(environment)`

Return the journal of the last run, or None without one.

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

def load_journal(self, environment: str) -> DeploymentJournal | None:
    """Return the journal of the last run, or None without one."""
    ...

`save_journal(environment, journal)`

Store the journal of a run, replacing the one before.

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

def save_journal(
    self, environment: str, journal: DeploymentJournal
) -> None:
    """Store the journal of a run, replacing the one before."""
    ...

`LocalJsonStateBackend`

Keep deployment states as JSON files in a local folder.

Each environment gets <folder>/<environment>.json. In the file name, characters other than letters, digits, ., _ and - become -; the environment is also stored inside the file, so two names that map to one file are caught. A file is replaced atomically, so a failed write leaves the previous state intact.

The lock of an environment is <folder>/<environment>.lock, created only when there is none. It says who holds it and until when; once expired, another run takes it over. It keeps apart runs that share the folder, such as two on one machine. The journal of the environment's last run is <folder>/<environment>.journal.json.

Parameters:

Name	Type	Description	Default
`folder`	`str \| Path`	The folder of the state files, created on the first save.	required
`lock_timeout`	`float`	How many seconds to wait for a lock another run holds. Defaults to 0: fail at once.	`0`
`lock_ttl`	`float`	How many seconds a lock holds unless its run releases it. Defaults to two hours.	`LOCK_TTL_SECONDS`

Examples:

state = LocalJsonStateBackend('.pyfabricops/state')
report = deploy_all_items(
    'Sales-PRD',
    'workspace',
    start_path='workspace',
    state_backend=state,
    environment='prod',
)

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

class LocalJsonStateBackend:
    """
    Keep deployment states as JSON files in a local folder.

    Each environment gets ``<folder>/<environment>.json``. In the file name,
    characters other than letters, digits, ``.``, ``_`` and ``-`` become
    ``-``; the environment is also stored inside the file, so two names that
    map to one file are caught. A file is replaced atomically, so a failed
    write leaves the previous state intact.

    The lock of an environment is ``<folder>/<environment>.lock``, created
    only when there is none. It says who holds it and until when; once
    expired, another run takes it over. It keeps apart runs that share the
    folder, such as two on one machine. The journal of the environment's
    last run is ``<folder>/<environment>.journal.json``.

    Args:
        folder (str | Path): The folder of the state files, created on the
            first save.
        lock_timeout (float, optional): How many seconds to wait for a lock
            another run holds. Defaults to 0: fail at once.
        lock_ttl (float, optional): How many seconds a lock holds unless its
            run releases it. Defaults to two hours.

    Examples:
        ```python
        state = LocalJsonStateBackend('.pyfabricops/state')
        report = deploy_all_items(
            'Sales-PRD',
            'workspace',
            start_path='workspace',
            state_backend=state,
            environment='prod',
        )
        ```
    """

    def __init__(
        self,
        folder: str | Path,
        *,
        lock_timeout: float = 0,
        lock_ttl: float = LOCK_TTL_SECONDS,
    ) -> None:
        self._folder = Path(folder)
        self._lock_timeout = lock_timeout
        self._lock_ttl = lock_ttl
        self._locks = _LocalLocks(self._lock_path, lock_ttl)

    def load(self, environment: str) -> DeploymentState | None:
        """
        Return the state of an environment, or None when it has none.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentState | None: The state, or None without a file.

        Raises:
            ConfigurationError: If the file is not a valid state, or holds the
                state of another environment.
        """
        path = self._path(environment)
        try:
            content = path.read_bytes()
        except FileNotFoundError:
            return None
        try:
            data = json.loads(content)
        except ValueError as e:
            raise ConfigurationError(f"{path} is not valid JSON: {e}") from e

        state = DeploymentState.from_dict(data, source=str(path))
        if state.environment != environment:
            raise ConfigurationError(
                f"{path} holds the state of environment "
                f"'{state.environment}', not '{environment}'."
            )
        return state

    def save(self, environment: str, state: DeploymentState) -> None:
        """
        Store the state of an environment, replacing the previous one.

        Args:
            environment (str): The environment name.
            state (DeploymentState): The state to store.

        Raises:
            ConfigurationError: If the state belongs to another environment.
            OSError: If the file cannot be written.
        """
        if state.environment != environment:
            raise ConfigurationError(
                f"Cannot save the state of environment '{state.environment}' "
                f"as '{environment}'."
            )
        path = self._path(environment)
        path.parent.mkdir(parents=True, exist_ok=True)
        content = json.dumps(state.to_dict(), indent=2, sort_keys=True)
        _replace(path, content + "\n")

    def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
        """
        Hold the lock of an environment for the length of a ``with`` block.

        Args:
            environment (str): The environment name.

        Returns:
            AbstractContextManager[DeploymentLock]: Gives the lock held, and
                releases it when the block ends, even on an error.

        Raises:
            DeploymentLockedError: On entering the block, if another run
                holds the lock and still does after ``lock_timeout``
                seconds.
        """
        return _hold_lock(
            self._locks,
            environment,
            timeout=self._lock_timeout,
            ttl=self._lock_ttl,
        )

    def force_unlock(self, environment: str) -> DeploymentLock | None:
        """
        Remove the lock of an environment, whoever holds it.

        For a lock whose run is gone, when waiting for it to expire is not
        an option. Make sure no run is deploying to the environment first.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentLock | None: The lock removed, or None without one.
        """
        return _force_unlock(self._locks, environment)

    def load_journal(self, environment: str) -> DeploymentJournal | None:
        """
        Return the journal of the environment's last run, if any.

        A file that holds no journal of the environment is left alone with
        a warning: a journal only saves work, and its loss costs none.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentJournal | None: The journal, or None.
        """
        path = self._journal_path(environment)
        try:
            content = path.read_bytes()
        except FileNotFoundError:
            return None
        return _parse_journal(content, str(path), environment)

    def save_journal(
        self, environment: str, journal: DeploymentJournal
    ) -> None:
        """
        Store the journal of a run, replacing the one before.

        Args:
            environment (str): The environment name.
            journal (DeploymentJournal): The journal.

        Raises:
            OSError: If the file cannot be written.
        """
        path = self._journal_path(environment)
        path.parent.mkdir(parents=True, exist_ok=True)
        _replace(path, journal_json(journal))

    def _path(self, environment: str) -> Path:
        """Return the file that keeps the state of an environment."""
        return self._folder / f"{state_file_name(environment)}.json"

    def _journal_path(self, environment: str) -> Path:
        """Return the file that keeps the journal of an environment."""
        return self._folder / f"{state_file_name(environment)}.journal.json"

    def _lock_path(self, environment: str) -> Path:
        """Return the file that holds the lock of an environment."""
        return self._folder / f"{state_file_name(environment)}.lock"

`force_unlock(environment)`

Remove the lock of an environment, whoever holds it.

For a lock whose run is gone, when waiting for it to expire is not an option. Make sure no run is deploying to the environment first.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentLock \| None`	DeploymentLock \| None: The lock removed, or None without one.

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

def force_unlock(self, environment: str) -> DeploymentLock | None:
    """
    Remove the lock of an environment, whoever holds it.

    For a lock whose run is gone, when waiting for it to expire is not
    an option. Make sure no run is deploying to the environment first.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentLock | None: The lock removed, or None without one.
    """
    return _force_unlock(self._locks, environment)

`load(environment)`

Return the state of an environment, or None when it has none.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentState \| None`	DeploymentState \| None: The state, or None without a file.

Raises:

Type	Description
`ConfigurationError`	If the file is not a valid state, or holds the state of another environment.

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

def load(self, environment: str) -> DeploymentState | None:
    """
    Return the state of an environment, or None when it has none.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentState | None: The state, or None without a file.

    Raises:
        ConfigurationError: If the file is not a valid state, or holds the
            state of another environment.
    """
    path = self._path(environment)
    try:
        content = path.read_bytes()
    except FileNotFoundError:
        return None
    try:
        data = json.loads(content)
    except ValueError as e:
        raise ConfigurationError(f"{path} is not valid JSON: {e}") from e

    state = DeploymentState.from_dict(data, source=str(path))
    if state.environment != environment:
        raise ConfigurationError(
            f"{path} holds the state of environment "
            f"'{state.environment}', not '{environment}'."
        )
    return state

`load_journal(environment)`

Return the journal of the environment's last run, if any.

A file that holds no journal of the environment is left alone with a warning: a journal only saves work, and its loss costs none.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentJournal \| None`	DeploymentJournal \| None: The journal, or None.

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

def load_journal(self, environment: str) -> DeploymentJournal | None:
    """
    Return the journal of the environment's last run, if any.

    A file that holds no journal of the environment is left alone with
    a warning: a journal only saves work, and its loss costs none.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentJournal | None: The journal, or None.
    """
    path = self._journal_path(environment)
    try:
        content = path.read_bytes()
    except FileNotFoundError:
        return None
    return _parse_journal(content, str(path), environment)

`lock(environment)`

Hold the lock of an environment for the length of a with block.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`AbstractContextManager[DeploymentLock]`	AbstractContextManager[DeploymentLock]: Gives the lock held, and releases it when the block ends, even on an error.

Raises:

Type	Description
`DeploymentLockedError`	On entering the block, if another run holds the lock and still does after `lock_timeout` seconds.

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

def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
    """
    Hold the lock of an environment for the length of a ``with`` block.

    Args:
        environment (str): The environment name.

    Returns:
        AbstractContextManager[DeploymentLock]: Gives the lock held, and
            releases it when the block ends, even on an error.

    Raises:
        DeploymentLockedError: On entering the block, if another run
            holds the lock and still does after ``lock_timeout``
            seconds.
    """
    return _hold_lock(
        self._locks,
        environment,
        timeout=self._lock_timeout,
        ttl=self._lock_ttl,
    )

`save(environment, state)`

Store the state of an environment, replacing the previous one.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required
`state`	`DeploymentState`	The state to store.	required

Raises:

Type	Description
`ConfigurationError`	If the state belongs to another environment.
`OSError`	If the file cannot be written.

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

def save(self, environment: str, state: DeploymentState) -> None:
    """
    Store the state of an environment, replacing the previous one.

    Args:
        environment (str): The environment name.
        state (DeploymentState): The state to store.

    Raises:
        ConfigurationError: If the state belongs to another environment.
        OSError: If the file cannot be written.
    """
    if state.environment != environment:
        raise ConfigurationError(
            f"Cannot save the state of environment '{state.environment}' "
            f"as '{environment}'."
        )
    path = self._path(environment)
    path.parent.mkdir(parents=True, exist_ok=True)
    content = json.dumps(state.to_dict(), indent=2, sort_keys=True)
    _replace(path, content + "\n")

`save_journal(environment, journal)`

Store the journal of a run, replacing the one before.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required
`journal`	`DeploymentJournal`	The journal.	required

Raises:

Type	Description
`OSError`	If the file cannot be written.

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

def save_journal(
    self, environment: str, journal: DeploymentJournal
) -> None:
    """
    Store the journal of a run, replacing the one before.

    Args:
        environment (str): The environment name.
        journal (DeploymentJournal): The journal.

    Raises:
        OSError: If the file cannot be written.
    """
    path = self._journal_path(environment)
    path.parent.mkdir(parents=True, exist_ok=True)
    _replace(path, journal_json(journal))

`LockingStateBackend`

Bases: DeploymentStateBackend, Protocol

A state backend that also locks the state of an environment.

deploy_all_items holds the lock from before it reads the state until after it records it, so two runs never deploy to one environment at a time. The backends of pyfabricops lock; a backend with only load and save still works, without a lock.

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

@runtime_checkable
class LockingStateBackend(DeploymentStateBackend, Protocol):
    """
    A state backend that also locks the state of an environment.

    ``deploy_all_items`` holds the lock from before it reads the state until
    after it records it, so two runs never deploy to one environment at a
    time. The backends of pyfabricops lock; a backend with only ``load``
    and ``save`` still works, without a lock.
    """

    def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
        """
        Hold the lock of an environment for the length of a ``with`` block.

        Raises:
            DeploymentLockedError: If another run holds it.
        """
        ...

    def force_unlock(self, environment: str) -> DeploymentLock | None:
        """Remove the lock of an environment, whoever holds it."""
        ...

`force_unlock(environment)`

Remove the lock of an environment, whoever holds it.

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

def force_unlock(self, environment: str) -> DeploymentLock | None:
    """Remove the lock of an environment, whoever holds it."""
    ...

`lock(environment)`

Hold the lock of an environment for the length of a with block.

Raises:

Type	Description
`DeploymentLockedError`	If another run holds it.

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

def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
    """
    Hold the lock of an environment for the length of a ``with`` block.

    Raises:
        DeploymentLockedError: If another run holds it.
    """
    ...

`journal_json(journal)`

Write a journal as the JSON its file holds.

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

def journal_json(journal: DeploymentJournal) -> str:
    """Write a journal as the JSON its file holds."""
    return json.dumps(journal.to_dict(), indent=2) + "\n"

`state_file_name(environment)`

Name the files of an environment's state and lock, without extension.

Characters other than letters, digits, ., _ and - become -.

Raises:

Type	Description
`ConfigurationError`	If nothing of the name is left.

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

def state_file_name(environment: str) -> str:
    """
    Name the files of an environment's state and lock, without extension.

    Characters other than letters, digits, ``.``, ``_`` and ``-`` become
    ``-``.

    Raises:
        ConfigurationError: If nothing of the name is left.
    """
    name = re.sub(r"[^A-Za-z0-9._-]+", "-", environment).strip(".-")
    if not name:
        raise ConfigurationError(
            f"'{environment}' cannot name a deployment state file."
        )
    return name

Deployment states kept in a OneLake lakehouse.

OneLakeStateBackend keeps the state of each environment as a JSON file in the Files of a lakehouse, and its lock next to it, through the OneLake Blob API. Every write is conditional: a state is saved only over the one the run read, and a lock is created only where there is none, so two runs never overwrite each other. It needs a token for OneLake, which accepts only the Azure Storage audience, and no dependency beyond requests.

`OneLakeStateBackend`

Keep deployment states as JSON files in a OneLake lakehouse.

Each environment gets Files/<folder>/<environment>.json in the lakehouse, named as LocalJsonStateBackend names its files, and its lock <environment>.lock and the journal of its last run <environment>.journal.json next to it. The lakehouse can be in any workspace the identity can write to, such as one kept for operations.

A state is saved only over the one load read, or where there was none: when another run saved one in between, the save fails instead of overwriting it. A lock is created only where there is none; it says who holds it and until when, and once expired another run takes it over.

The workspace and the lakehouse are looked up once, by name or ID, and then reached by ID. The identity needs a token for OneLake, which set_auth_provider gets as for the Fabric API.

Parameters:

Name	Type	Description	Default
`workspace`	`str`	The name or ID of the lakehouse's workspace.	required
`lakehouse`	`str`	The name or ID of the lakehouse.	required
`folder`	`str`	The folder under `Files`. Defaults to `"pyfabricops/state"`.	`'pyfabricops/state'`
`endpoint`	`str`	The OneLake Blob endpoint. Defaults to the global one, `ONELAKE_BLOB_ENDPOINT`.	`ONELAKE_BLOB_ENDPOINT`
`lock_timeout`	`float`	How many seconds to wait for a lock another run holds. Defaults to 0: fail at once.	`0`
`lock_ttl`	`float`	How many seconds a lock holds unless its run releases it. Defaults to two hours.	`LOCK_TTL_SECONDS`

Examples:

state = OneLakeStateBackend('Ops', 'DeploymentState')
report = deploy_all_items(
    'Sales-PRD',
    staging,
    start_path=staging,
    repository_path='workspace',
    state_backend=state,
    environment='prod',
)

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

class OneLakeStateBackend:
    """
    Keep deployment states as JSON files in a OneLake lakehouse.

    Each environment gets ``Files/<folder>/<environment>.json`` in the
    lakehouse, named as ``LocalJsonStateBackend`` names its files, and its
    lock ``<environment>.lock`` and the journal of its last run
    ``<environment>.journal.json`` next to it. The lakehouse can be in any
    workspace the identity can write to, such as one kept for operations.

    A state is saved only over the one ``load`` read, or where there was
    none: when another run saved one in between, the save fails instead of
    overwriting it. A lock is created only where there is none; it says who
    holds it and until when, and once expired another run takes it over.

    The workspace and the lakehouse are looked up once, by name or ID, and
    then reached by ID. The identity needs a token for OneLake, which
    ``set_auth_provider`` gets as for the Fabric API.

    Args:
        workspace (str): The name or ID of the lakehouse's workspace.
        lakehouse (str): The name or ID of the lakehouse.
        folder (str, optional): The folder under ``Files``. Defaults to
            ``"pyfabricops/state"``.
        endpoint (str, optional): The OneLake Blob endpoint. Defaults to the
            global one, ``ONELAKE_BLOB_ENDPOINT``.
        lock_timeout (float, optional): How many seconds to wait for a lock
            another run holds. Defaults to 0: fail at once.
        lock_ttl (float, optional): How many seconds a lock holds unless its
            run releases it. Defaults to two hours.

    Examples:
        ```python
        state = OneLakeStateBackend('Ops', 'DeploymentState')
        report = deploy_all_items(
            'Sales-PRD',
            staging,
            start_path=staging,
            repository_path='workspace',
            state_backend=state,
            environment='prod',
        )
        ```
    """

    def __init__(
        self,
        workspace: str,
        lakehouse: str,
        folder: str = "pyfabricops/state",
        *,
        endpoint: str = ONELAKE_BLOB_ENDPOINT,
        lock_timeout: float = 0,
        lock_ttl: float = LOCK_TTL_SECONDS,
    ) -> None:
        self._workspace = workspace
        self._lakehouse = lakehouse
        self._folder = folder.strip("/")
        self._endpoint = endpoint.rstrip("/")
        self._lock_timeout = lock_timeout
        self._lock_ttl = lock_ttl
        # The URL of the folder, once the workspace and lakehouse are found.
        self._base: str | None = None
        # For each environment loaded, the ETag of its state, or None when
        # it had none.
        self._etags: dict[str, str | None] = {}
        self._locks = _OneLakeLocks(self)

    def load(self, environment: str) -> DeploymentState | None:
        """
        Return the state of an environment, or None when it has none.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentState | None: The state, or None without a file.

        Raises:
            ConfigurationError: If the file is not a valid state, or holds the
                state of another environment, or the lakehouse is not found.
            RequestError: If OneLake cannot be read.
        """
        file = f"{state_file_name(environment)}.json"
        response = self._request("GET", file)
        if response.status_code == 404:
            self._etags[environment] = None
            return None
        self._check(response, "read", file)

        where = self._where(file)
        try:
            data = json.loads(response.content)
        except ValueError as e:
            raise ConfigurationError(f"{where} is not valid JSON: {e}") from e
        state = DeploymentState.from_dict(data, source=where)
        if state.environment != environment:
            raise ConfigurationError(
                f"{where} holds the state of environment "
                f"'{state.environment}', not '{environment}'."
            )
        self._etags[environment] = response.headers.get("ETag")
        return state

    def save(self, environment: str, state: DeploymentState) -> None:
        """
        Store the state of an environment, over the one ``load`` read.

        Without an earlier ``load`` of the environment, the state is written
        whatever the file holds.

        Args:
            environment (str): The environment name.
            state (DeploymentState): The state to store.

        Raises:
            ConfigurationError: If the state belongs to another environment,
                or the lakehouse is not found.
            RequestError: If another run saved a state since ``load``, which
                is left as it is, or OneLake cannot be written.
        """
        if state.environment != environment:
            raise ConfigurationError(
                f"Cannot save the state of environment '{state.environment}' "
                f"as '{environment}'."
            )
        file = f"{state_file_name(environment)}.json"
        content = (
            json.dumps(state.to_dict(), indent=2, sort_keys=True) + "\n"
        ).encode("utf-8")
        headers = {
            "x-ms-blob-type": "BlockBlob",
            "Content-Type": "application/json",
        }
        if environment in self._etags:
            etag = self._etags[environment]
            headers.update(
                {"If-Match": etag} if etag else {"If-None-Match": "*"}
            )

        response = self._request("PUT", file, data=content, headers=headers)
        if response.status_code in (409, 412):
            # A request tried again after its answer was lost finds its own
            # write: then the state is saved.
            current = self._request("GET", file)
            if current.status_code != 200 or current.content != content:
                raise RequestError(
                    f"Deployment state '{environment}' changed in "
                    f"{self._where(file)} since this run read it, so it was "
                    "not overwritten: another run deployed to the "
                    "environment meanwhile."
                )
            response = current
        else:
            self._check(response, "write", file)
        self._etags[environment] = response.headers.get("ETag")

    def load_journal(self, environment: str) -> DeploymentJournal | None:
        """
        Return the journal of the environment's last run, if any.

        A file that holds no journal of the environment is left alone with
        a warning: a journal only saves work, and its loss costs none.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentJournal | None: The journal, or None.

        Raises:
            RequestError: If OneLake cannot be read.
        """
        file = _journal_file(environment)
        response = self._request("GET", file)
        if response.status_code == 404:
            return None
        self._check(response, "read the journal", file)
        return _parse_journal(response.content, self._where(file), environment)

    def save_journal(
        self, environment: str, journal: DeploymentJournal
    ) -> None:
        """
        Store the journal of a run, replacing the one before.

        Only the run that holds the lock writes the journal, so it is
        written whatever the file holds.

        Args:
            environment (str): The environment name.
            journal (DeploymentJournal): The journal.

        Raises:
            RequestError: If OneLake cannot be written.
        """
        file = _journal_file(environment)
        response = self._request(
            "PUT",
            file,
            data=journal_json(journal).encode("utf-8"),
            headers=dict(_LOCK_HEADERS),
        )
        self._check(response, "write the journal", file)

    def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
        """
        Hold the lock of an environment for the length of a ``with`` block.

        Args:
            environment (str): The environment name.

        Returns:
            AbstractContextManager[DeploymentLock]: Gives the lock held, and
                releases it when the block ends, even on an error.

        Raises:
            DeploymentLockedError: On entering the block, if another run
                holds the lock and still does after ``lock_timeout``
                seconds.
        """
        return _hold_lock(
            self._locks,
            environment,
            timeout=self._lock_timeout,
            ttl=self._lock_ttl,
        )

    def force_unlock(self, environment: str) -> DeploymentLock | None:
        """
        Remove the lock of an environment, whoever holds it.

        For a lock whose run is gone, when waiting for it to expire is not
        an option. Make sure no run is deploying to the environment first.

        Args:
            environment (str): The environment name.

        Returns:
            DeploymentLock | None: The lock removed, or None without one.
        """
        return _force_unlock(self._locks, environment)

    def _request(
        self,
        method: str,
        file: str,
        *,
        data: bytes | None = None,
        headers: dict[str, str] | None = None,
    ) -> requests.Response:
        """
        Send a Blob request for a file of the folder.

        A request is tried again after a transient failure: the callers
        tell a write of their own from another run's.

        Raises:
            AuthenticationError: If no token for OneLake can be had.
            RequestError: If OneLake cannot be reached.
        """
        token = _get_token(audience="storage")
        if not token or not token.get("access_token"):
            raise AuthenticationError(
                "Failed to retrieve a token for OneLake. Ensure that the "
                "authentication is set up correctly."
            )
        try:
            return _send(
                retry_transient=True,
                method=method,
                url=f"{self._base_url()}/{file}",
                headers={
                    "Authorization": f"Bearer {token['access_token']}",
                    "x-ms-version": _API_VERSION,
                    **(headers or {}),
                },
                data=data,
                timeout=_TIMEOUT_SECONDS,
            )
        except requests.exceptions.RequestException as e:
            raise RequestError(
                f"OneLake could not be reached for {self._where(file)}: {e}"
            ) from e

    def _base_url(self) -> str:
        """
        Return the URL of the folder, by the IDs of workspace and lakehouse.

        Raises:
            ConfigurationError: If the workspace or the lakehouse is not
                found.
        """
        if self._base is None:
            workspace_id = resolve_workspace(self._workspace)
            if workspace_id is None:
                raise ConfigurationError(
                    f"Workspace '{self._workspace}' not found."
                )
            lakehouse_id = resolve_lakehouse(workspace_id, self._lakehouse)
            if lakehouse_id is None:
                raise ConfigurationError(
                    f"Lakehouse '{self._lakehouse}' not found in workspace "
                    f"'{self._workspace}'."
                )
            files = f"Files/{self._folder}" if self._folder else "Files"
            self._base = (
                f"{self._endpoint}/{workspace_id}/{lakehouse_id}/{files}"
            )
        return self._base

    def _where(self, file: str) -> str:
        """Name a file for a message, by the names it was given."""
        folder = f"{self._folder}/" if self._folder else ""
        return f"{self._workspace}/{self._lakehouse}/Files/{folder}{file}"

    def _check(
        self, response: requests.Response, action: str, file: str
    ) -> None:
        """
        Raise when a request failed.

        Raises:
            RequestError: With the status and OneLake's error code.
        """
        if not response.ok:
            code = response.headers.get("x-ms-error-code")
            status = (
                f"{response.status_code} {code}"
                if code
                else (f"HTTP {response.status_code}")
            )
            raise RequestError(
                f"OneLake could not {action} {self._where(file)}: {status}."
            )

`force_unlock(environment)`

Remove the lock of an environment, whoever holds it.

For a lock whose run is gone, when waiting for it to expire is not an option. Make sure no run is deploying to the environment first.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentLock \| None`	DeploymentLock \| None: The lock removed, or None without one.

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

def force_unlock(self, environment: str) -> DeploymentLock | None:
    """
    Remove the lock of an environment, whoever holds it.

    For a lock whose run is gone, when waiting for it to expire is not
    an option. Make sure no run is deploying to the environment first.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentLock | None: The lock removed, or None without one.
    """
    return _force_unlock(self._locks, environment)

`load(environment)`

Return the state of an environment, or None when it has none.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentState \| None`	DeploymentState \| None: The state, or None without a file.

Raises:

Type	Description
`ConfigurationError`	If the file is not a valid state, or holds the state of another environment, or the lakehouse is not found.
`RequestError`	If OneLake cannot be read.

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

def load(self, environment: str) -> DeploymentState | None:
    """
    Return the state of an environment, or None when it has none.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentState | None: The state, or None without a file.

    Raises:
        ConfigurationError: If the file is not a valid state, or holds the
            state of another environment, or the lakehouse is not found.
        RequestError: If OneLake cannot be read.
    """
    file = f"{state_file_name(environment)}.json"
    response = self._request("GET", file)
    if response.status_code == 404:
        self._etags[environment] = None
        return None
    self._check(response, "read", file)

    where = self._where(file)
    try:
        data = json.loads(response.content)
    except ValueError as e:
        raise ConfigurationError(f"{where} is not valid JSON: {e}") from e
    state = DeploymentState.from_dict(data, source=where)
    if state.environment != environment:
        raise ConfigurationError(
            f"{where} holds the state of environment "
            f"'{state.environment}', not '{environment}'."
        )
    self._etags[environment] = response.headers.get("ETag")
    return state

`load_journal(environment)`

Return the journal of the environment's last run, if any.

A file that holds no journal of the environment is left alone with a warning: a journal only saves work, and its loss costs none.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`DeploymentJournal \| None`	DeploymentJournal \| None: The journal, or None.

Raises:

Type	Description
`RequestError`	If OneLake cannot be read.

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

def load_journal(self, environment: str) -> DeploymentJournal | None:
    """
    Return the journal of the environment's last run, if any.

    A file that holds no journal of the environment is left alone with
    a warning: a journal only saves work, and its loss costs none.

    Args:
        environment (str): The environment name.

    Returns:
        DeploymentJournal | None: The journal, or None.

    Raises:
        RequestError: If OneLake cannot be read.
    """
    file = _journal_file(environment)
    response = self._request("GET", file)
    if response.status_code == 404:
        return None
    self._check(response, "read the journal", file)
    return _parse_journal(response.content, self._where(file), environment)

`lock(environment)`

Hold the lock of an environment for the length of a with block.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required

Returns:

Type	Description
`AbstractContextManager[DeploymentLock]`	AbstractContextManager[DeploymentLock]: Gives the lock held, and releases it when the block ends, even on an error.

Raises:

Type	Description
`DeploymentLockedError`	On entering the block, if another run holds the lock and still does after `lock_timeout` seconds.

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

def lock(self, environment: str) -> AbstractContextManager[DeploymentLock]:
    """
    Hold the lock of an environment for the length of a ``with`` block.

    Args:
        environment (str): The environment name.

    Returns:
        AbstractContextManager[DeploymentLock]: Gives the lock held, and
            releases it when the block ends, even on an error.

    Raises:
        DeploymentLockedError: On entering the block, if another run
            holds the lock and still does after ``lock_timeout``
            seconds.
    """
    return _hold_lock(
        self._locks,
        environment,
        timeout=self._lock_timeout,
        ttl=self._lock_ttl,
    )

`save(environment, state)`

Store the state of an environment, over the one load read.

Without an earlier load of the environment, the state is written whatever the file holds.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required
`state`	`DeploymentState`	The state to store.	required

Raises:

Type	Description
`ConfigurationError`	If the state belongs to another environment, or the lakehouse is not found.
`RequestError`	If another run saved a state since `load`, which is left as it is, or OneLake cannot be written.

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

def save(self, environment: str, state: DeploymentState) -> None:
    """
    Store the state of an environment, over the one ``load`` read.

    Without an earlier ``load`` of the environment, the state is written
    whatever the file holds.

    Args:
        environment (str): The environment name.
        state (DeploymentState): The state to store.

    Raises:
        ConfigurationError: If the state belongs to another environment,
            or the lakehouse is not found.
        RequestError: If another run saved a state since ``load``, which
            is left as it is, or OneLake cannot be written.
    """
    if state.environment != environment:
        raise ConfigurationError(
            f"Cannot save the state of environment '{state.environment}' "
            f"as '{environment}'."
        )
    file = f"{state_file_name(environment)}.json"
    content = (
        json.dumps(state.to_dict(), indent=2, sort_keys=True) + "\n"
    ).encode("utf-8")
    headers = {
        "x-ms-blob-type": "BlockBlob",
        "Content-Type": "application/json",
    }
    if environment in self._etags:
        etag = self._etags[environment]
        headers.update(
            {"If-Match": etag} if etag else {"If-None-Match": "*"}
        )

    response = self._request("PUT", file, data=content, headers=headers)
    if response.status_code in (409, 412):
        # A request tried again after its answer was lost finds its own
        # write: then the state is saved.
        current = self._request("GET", file)
        if current.status_code != 200 or current.content != content:
            raise RequestError(
                f"Deployment state '{environment}' changed in "
                f"{self._where(file)} since this run read it, so it was "
                "not overwritten: another run deployed to the "
                "environment meanwhile."
            )
        response = current
    else:
        self._check(response, "write", file)
    self._etags[environment] = response.headers.get("ETag")

`save_journal(environment, journal)`

Store the journal of a run, replacing the one before.

Only the run that holds the lock writes the journal, so it is written whatever the file holds.

Parameters:

Name	Type	Description	Default
`environment`	`str`	The environment name.	required
`journal`	`DeploymentJournal`	The journal.	required

Raises:

Type	Description
`RequestError`	If OneLake cannot be written.

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

def save_journal(
    self, environment: str, journal: DeploymentJournal
) -> None:
    """
    Store the journal of a run, replacing the one before.

    Only the run that holds the lock writes the journal, so it is
    written whatever the file holds.

    Args:
        environment (str): The environment name.
        journal (DeploymentJournal): The journal.

    Raises:
        RequestError: If OneLake cannot be written.
    """
    file = _journal_file(environment)
    response = self._request(
        "PUT",
        file,
        data=journal_json(journal).encode("utf-8"),
        headers=dict(_LOCK_HEADERS),
    )
    self._check(response, "write the journal", file)

Deployment State

In a OneLake lakehouse

In a local folder

Locks

Journal and resume

Other backends

Reference

DeploymentJournal dataclass

interrupted property

finish(ok)

from_dict(data, *, source) classmethod

sent()

start(environment, workspace, source_commit, *, resumed=None) classmethod

to_dict()

with_entry(entry)

DeploymentLock dataclass

expired property

describe()

from_dict(data, *, source) classmethod

new(environment, ttl_seconds) classmethod

to_dict()

DeploymentState dataclass

from_dict(data, *, source) classmethod

targets(workspace)

to_dict()

DeploymentStateBackend

load(environment)

save(environment, state)

JournalEntry dataclass

JournalingStateBackend

load_journal(environment)

save_journal(environment, journal)

LocalJsonStateBackend

force_unlock(environment)

load(environment)

load_journal(environment)

lock(environment)

save(environment, state)

save_journal(environment, journal)

LockingStateBackend

force_unlock(environment)

lock(environment)

journal_json(journal)

state_file_name(environment)

OneLakeStateBackend

force_unlock(environment)

load(environment)

load_journal(environment)

lock(environment)

save(environment, state)

save_journal(environment, journal)

`DeploymentJournal` `dataclass`

`interrupted` `property`

`finish(ok)`

`from_dict(data, *, source)` `classmethod`

`sent()`

`start(environment, workspace, source_commit, *, resumed=None)` `classmethod`

`to_dict()`

`with_entry(entry)`

`DeploymentLock` `dataclass`

`expired` `property`

`describe()`

`from_dict(data, *, source)` `classmethod`

`new(environment, ttl_seconds)` `classmethod`

`to_dict()`

`DeploymentState` `dataclass`

`from_dict(data, *, source)` `classmethod`

`targets(workspace)`

`to_dict()`

`DeploymentStateBackend`

`load(environment)`

`save(environment, state)`

`JournalEntry` `dataclass`

`JournalingStateBackend`

`load_journal(environment)`

`save_journal(environment, journal)`

`LocalJsonStateBackend`

`force_unlock(environment)`

`load(environment)`

`load_journal(environment)`

`lock(environment)`

`save(environment, state)`

`save_journal(environment, journal)`

`LockingStateBackend`

`force_unlock(environment)`

`lock(environment)`

`journal_json(journal)`

`state_file_name(environment)`

`OneLakeStateBackend`

`force_unlock(environment)`

`load(environment)`

`load_journal(environment)`

`lock(environment)`

`save(environment, state)`

`save_journal(environment, journal)`