Pipeline

Core pipeline model and pipeline-level configuration.

fabric_data_pipelines.Pipeline

Bases: FabricModel

A Fabric data pipeline.

Activities are declared declaratively; call :meth:then / :meth:after on them to express dependencies, then pass the list to Pipeline.

Schedules must be attached on the pipeline itself via schedules=[...] (up to 20). Fabric stores them in a sibling .schedules file written by :meth:save_item.

Example::

from fabric_data_pipelines import Pipeline, Wait, Notebook, Schedule, Weekly

wait = Wait(name="pause", wait_time_in_seconds=5)
nb = Notebook(name="run", notebook_id="...", workspace_id="...")
wait.then(nb)

schedule = Schedule(
    configuration=Weekly(times=["21:30"], weekdays=["Monday"]),
)
pipeline = Pipeline(
    name="daily_load",
    activities=[wait, nb],
    schedules=[schedule],
)
pipeline.save_item("out")

resolve_logical_id

resolve_logical_id(
    item_dir: str | Path | None = None,
) -> str

Resolve a stable Fabric logicalId.

Priority: 1. explicit Pipeline.logical_id 2. existing .platform on disk (if item_dir points at the item folder) 3. deterministic UUIDv5 derived from pipeline name

platform_dict

platform_dict(logical_id: str) -> dict[str, Any]

Serialize .platform file payload.

schedules_dict

schedules_dict() -> dict[str, Any] | None

Serialize .schedules payload (or None if no schedules are set).

validate_graph

validate_graph() -> None

Validate activity names and the dependency graph.

Raises:

Type	Description
DuplicateActivityNameError	Two activities share a name.
UnknownDependencyError	A dependsOn target does not exist.
CrossScopeDependencyError	A dependsOn target is in another scope.
CyclicDependencyError	Dependencies form a cycle within a scope.

to_dict

to_dict() -> dict[str, Any]

Build the Fabric pipeline-content.json structure.

Returns:

Type	Description
dict[str, Any]	{"properties": {"activities": [...], ...}}

to_json

to_json(*, indent: int | None = 2) -> str

Serialize to Fabric pipeline JSON text.

from_dict classmethod

from_dict(data: dict[str, Any], *, name: str) -> Pipeline

Parse Fabric pipeline-content JSON into a typed pipeline.

The pipeline name is required because Fabric content JSON does not include it (it lives on the item folder / .platform display name).

Unknown activity types become :class:~fabric_data_pipelines.RawActivity.

from_json classmethod

from_json(text: str, *, name: str) -> Pipeline

Parse Fabric pipeline JSON text into a typed pipeline.

See :meth:from_dict for naming and typing rules.

load_item classmethod

load_item(directory: str | Path) -> Pipeline

Load a Fabric *.DataPipeline Git item folder.

Reads pipeline-content.json, optional .schedules, and logicalId from .platform when present.

save

save(path: str | Path) -> None

Write pipeline-content.json (or any path) to disk.

Example::

pipeline.save("daily_load.json")

save_item

save_item(
    directory: str | Path, *, prune: bool = True
) -> Path

Write a Fabric item folder with pipeline-content, platform, and schedules.

Creates <directory>/<name>.DataPipeline/ containing all item files.

Example::

pipeline.save_item("out")
# -> out/daily_load.DataPipeline/pipeline-content.json
# -> out/daily_load.DataPipeline/.platform

to_definition

to_definition(
    *, item_dir: str | Path | None = None
) -> dict[str, Any]

Build the REST API item definition payload with base64 parts.

Returns a dict suitable for the Fabric Items API definition field::

{
  "parts": [
    {"path": "pipeline-content.json", "payload": "...", "payloadType": "InlineBase64"},
    {"path": ".platform", "payload": "...", "payloadType": "InlineBase64"},
    # NOTE: `.schedules` is not part of the Items API definition payload.
  ]
}

fabric_data_pipelines.Parameter

Bases: FabricModel

A pipeline parameter definition.

Example::

Parameter(type="Int", default_value=0)
Parameter(type="String", default_value="full")

fabric_data_pipelines.Variable

Bases: FabricModel

A pipeline variable definition.

Example::

Variable(type="String")
Variable(type="Array", default_value=[])

fabric_data_pipelines.LibraryVariable

Bases: FabricModel

A reference to a Fabric Variable Library value.

Example::

LibraryVariable(
    type="String",
    variable_name="SourceDb",
    library_name="Demo_ETL_Library",
)