Metadata-Version: 2.4
Name: isynth-provisioning
Version: 0.13.0
Summary: Generic provisioning framework for exporting iSynth test data into arbitrary target systems
Project-URL: Homepage, https://isynth.io
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: python-box<8,>=7
Requires-Dist: pydantic<3,>=2
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: ruff==0.16.0; extra == "dev"
Requires-Dist: hypothesis; extra == "dev"
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"

# 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:

```python
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:

```python
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:

```python
# 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).

```python
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
configure_logging(log_format=LogFormat.BEAUTIFIED)       # banners + one line per export program - for watching a run happen
```

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.:

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

`event` is a stable, typo-proof name for what this log line is about — every log line the framework emits on its own carries one, so you can match on it instead of the human-readable `message` text. A line with no `event` field is one your own code logged (e.g. via `ctx.logger.info(...)` in a mapper or hook).

`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
```

`LogFormat.BEAUTIFIED` is a third option, for when you just want to watch a run happen in a terminal rather than read every counter: a boxed banner for each block, one line per export program with a status icon (⏳ while it's running, then ✅ on success, ⚠️ if it failed but the workflow kept going, or ❌ if it failed and cancelled the run), and a minimal `(N read, N successful, N errors)` count instead of the full statistics tree. Chatty framework detail is left out to keep this readable, but anything your own code logs still shows, at whatever level you configured. Pass `full_report=True` alongside it if you still want the full statistics tree printed at the very end of the run (combining `full_report=True` with `JSON`/`HUMAN` raises `ValueError`, since those already print that tree on every "finished workflow" line):

```python
configure_logging(log_format=LogFormat.BEAUTIFIED, full_report=True)
```

Pick whichever suits how you're consuming the output — `JSON` (default) if logs are shipped somewhere and parsed, `HUMAN` if you want every detail while reading directly in a terminal, `BEAUTIFIED` if you just want to watch a run happen.

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:

```python
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.

