isynth-provisioning Documentation

isynth-provisioning lets you provision test data into arbitrary target systems (REST, Kafka, databases, files, ...) by configuring workflows and export programs, instead of hand-rolling an orchestration engine per project.

Getting Started

This guide gets you from "nothing installed" to a running workflow that exports one iSynth object type to a target system.

Install

Add this to your project's requirements.txt:

isynth-provisioning

Concrete adapters for specific target systems (REST, Kafka, a database, ...) are not part of this framework — it only ships the adapter convention itself. See writing-an-adapter.md for how to write one; it's a small amount of code, following a documented convention.

Then pip install -r requirements.txt as usual.

Book-keeping columns (prerequisite)

Before any workflow can run, every object type you plan to provision needs 7 columns that the framework's book-keeping reads and writes automatically (loadts, loaderrors, loaderror_traceback, load_success, last_load_program, last_load_step_number, last_load_step_name — see concepts.md's "Book-keeping" section for what each one means). These are declared in your iSynth project (not a workflow file) as Attribute(...) entries on a base.types.ObjectType — typically once, on a common abstract supertype everything else inherits from.

isynth_provisioning generates the correct Attribute(...) list for you via bookkeeping_attributes(...), so you don't hand-write (or accidentally get wrong) 7 fiddly Field/Treat/Table declarations:

from base.types import ObjectType, BaseObject, Attribute, Field, Treat, Table, Graph
from isynth_provisioning import bookkeeping_attributes

SynthObject = ObjectType(
    "SynthObject",
    doc="Contains attributes common to all objects to be synthesized. All other object types inherit from this object type",
    super_type=BaseObject,
    is_abstract=True,
    attributes=bookkeeping_attributes(Attribute, Field, Treat, Table, Graph) + [
        # add any other project-specific attributes here, e.g.:
        # Attribute("alt_label", Field.CHAR, Treat.EDIT, Table.SHOW, Graph.SHOW, preset=None),
    ],
    children=[],
)

bookkeeping_attributes takes Attribute/Field/Treat/Table/Graph as arguments — the real classes from your own from base.types import ... — rather than being a plain importable list. This is deliberate: isynth_provisioning needs to stay installable and testable without the iSynth engine present at all, so it never imports base.types (or any other base.* module) itself; you supply the real classes at the one place that's supposed to import them, same as how you build the gateway yourself from base.data_access functions (see "The gateway" in concepts.md) instead of the framework importing that either.

Every other object type you plan to provision can then extend this common supertype, rather than declaring the 7 book-keeping columns again itself:

Person = ObjectType(
    "Person",
    doc="Just some example object to show how to extend the SynthObject",
    super_type=SynthObject, # this super_type gives you all the bookkeeping attributes
    # other settings
    attributes=[
        # add your attributes
    ],
)

The smallest possible workflow

A workflow is just a Python file the iSynth engine runs as an ordinary script — there's no special entrypoint signature it calls with arguments. Your file builds the gateway itself from the functions iSynth provides, builds a Workflow object, and runs it. Here's one export program, one step, sourcing directly from an iSynth object type:

# workflows/provision_addresses.py
from typing import Any

from base.data_access import table_rows, update_row, execute_fkrel_updates
from isynth_provisioning import (
    Workflow, ExportProgram, Step, ObjectTypeSource, ExecutionContext,
    StepOutcome, AdapterOutcome, run_workflow,
)
from isynth_provisioning.gateway import IsynthGateway, IsynthGatewayProtocol
from isynth_provisioning.logging_utils import configure_logging

def map_address(ctx: ExecutionContext, row: Any) -> list[Any]:
    return [{
        "street": row.street,
        "city": row.city,
        "country": row.country,
    }]

def send_address(mapped_item: Any, *, debug: bool = False) -> Any:
    # replace with a real adapter for your target system, see writing-an-adapter.md
    print("would send:", mapped_item)
    return {"external_id": "addr-123"}

def moveback_address(ctx: ExecutionContext, row: Any, outcome: AdapterOutcome) -> StepOutcome:
    if outcome.error is not None:
        return StepOutcome(success=False, error_message=str(outcome.error))
    return StepOutcome(success=True, moveback={"external_address_id": outcome.result["external_id"]})

export_addresses = ExportProgram(
    id="export_address",
    source=ObjectTypeSource(object_type="PostalAddress"),
    steps=[
        Step(
            name="send_to_target",
            mapper=map_address,
            send=send_address,
            post_processor=moveback_address,
        ),
    ],
)

workflow = Workflow(name="provision_addresses", programs=[export_addresses])

def build_gateway() -> IsynthGatewayProtocol:
    return IsynthGateway(table_rows=table_rows, update_row=update_row,
                          execute_fkrel_updates=execute_fkrel_updates)

def main() -> None:
    configure_logging()   # see "Logging" below - without this, log output is silently dropped
    gateway = build_gateway()
    run_workflow(workflow, gateway)

if __name__ == "__main__":
    main()

That's the whole shape:

• configure_logging() turns on the framework's structured (JSON-per-line) logging — see "Logging" below.
• build_gateway() wraps the three functions iSynth provides (table_rows/update_row/execute_fkrel_updates) into one IsynthGatewayProtocol — you build this yourself, once; nothing hands it to you. See concepts.md's "The gateway" section.
• ObjectTypeSource("PostalAddress") tells the framework to read rows via table_rows("PostalAddress", ...) automatically — you never call table_rows yourself.
• map_address turns one source row into 0..n items in whatever shape your adapter expects.
• send_address is a bare function plugged in as Step.send — the simplest way to wire up an adapter without writing a class; swap it for a bound method on a real adapter package's adapter object once you have one (see writing-an-adapter.md).
• moveback_address decides success/failure and what to write back onto the source row (via update_row, handled for you).

Logging

Call configure_logging() exactly once, as early as possible (the very first line of main() is the usual spot) — without it, nothing attaches a handler to the framework's logger, so even the workflow/program start-end, periodic progress, and failure log lines run_workflow already emits for free are silently dropped (Python's own "handler of last resort" only shows bare WARNING+ text, not the structured output below).

from isynth_provisioning.logging_utils import configure_logging, LogFormat
import logging

configure_logging()                                     # INFO level, JSON lines to stderr - the defaults
configure_logging(level=logging.DEBUG)                  # also surfaces bookkeeping.py's per-row/step detail
                                                          # and "step skipped: unmet row-level dependency" lines
configure_logging(stream=my_open_file)                  # write to a file/anything IO[str] instead of stderr
configure_logging(log_format=LogFormat.HUMAN)            # multi-line, indented text instead of JSON - for reading directly in a terminal

By default, output is one JSON object per line — {"timestamp": ..., "level": ..., "logger": ..., "message": ..., ...}, with workflow/program/step identity and other context included via extra fields, e.g.:

{"timestamp": "2026-07-09T12:00:00", "level": "INFO", "logger": "isynth_provisioning.export_address", "message": "program progress", "workflow": "provision_addresses", "program": "export_address", "rows_read": 100}

LogFormat.HUMAN renders the exact same fields as readable, indented text instead — most noticeable on "finished program"/"finished workflow", whose statistics extra is a multi-level nested dict that's unreadable as one JSON line but reads as a real tree here:

2026-07-09 12:00:00 INFO     isynth_provisioning.provision_addresses: finished workflow
  workflow: provision_addresses
  status: success
  has_errors: False
  statistics:
    export_address:
      rows_read: 100
      rows_skipped_already_done: 0
      steps:
        create_address:
          attempted: 100
          succeeded: 100
          total_successful: 100

Pick whichever suits how you're consuming the output — JSON (default) if logs are shipped somewhere and parsed, HUMAN if you're watching a workflow run in a terminal.

No workflow code needs to log anything itself to get this — see error-handling-and-retries.md's "Live progress and statistics" section for the full list of what's logged automatically (including a per-row-count progress line, controlled by Workflow.progress_log_every) and how to read the same numbers programmatically via WorkflowReport.statistics.

Adding a second export program

Day-2 usage is almost always just this — append another ExportProgram to the list:

workflow = Workflow(name="provision_addresses", programs=[
    export_addresses,
    export_natural_persons,   # a second ExportProgram, same pattern
])

Export programs run sequentially, in list order. If you need one program to only run after another has succeeded, see error-handling-and-retries.md's section on dependencies.