Pipeline¶

def plot(self) -> str:
    """Return a one-edge Mermaid ``graph TD`` diagram for this pipeline.

    Equivalent to wrapping ``self`` in a ``FullPipeline`` and calling
    ``plot`` on it. Drop the returned string into a fenced
    ` ```mermaid ` block in markdown to render the diagram.
    """
    return FullPipeline([self]).plot()

def run(self, *, dry_run: bool = False) -> list[Any]:
    """Execute the pipeline over every ``new`` row in the source topic.

    For each row: decode the payload, call ``fn``, validate the result
    against the target's model (if any), then within a single
    transaction insert the result and mark the source row as
    ``handled``. Returns the list of transformed outputs in
    insertion order.

    Args:
        dry_run: If ``True``, run ``fn`` and validate the results but
            do not write anything to the target topic or transition
            source rows. Useful for previewing.

    Returns:
        The list of outputs in source-row order. If the pipeline has
        no target, this is the raw return value of ``fn``; with a
        target, it is the validated target-model instance (or ``None``
        for rows where ``fn`` returned ``None``).

    Raises:
        DuplicateMessageError: An output row would violate the target
            topic's dedup constraint.
        pydantic.ValidationError: ``fn`` returned a value that does not
            validate against the target topic's model.

    Emits:
        ``pipeline_start`` before processing and ``pipeline_end`` after
        (with ``count`` and ``dry_run``). Each handled source row also
        emits ``message_handled``.
    """
    source = self.source_topic.source
    source._emit(
        "pipeline_start",
        source=self.source_topic.name,
        target=self.target_name,
    )
    rows = list(self.source_topic.iter_new(records=True))
    results: list[Any] = []
    for record in rows:
        assert isinstance(record, Record)
        result = self.fn(record.data)
        if self.target_topic is None:
            results.append(result)
            continue
        if result is None:
            results.append(None)
            if not dry_run:
                self.source_topic.set_handled(record=record)
            continue
        target_model = self.target_topic._validate(result)
        results.append(target_model)
        if dry_run:
            continue
        target_payload = _model_to_payload(target_model)
        with source._conn:
            self.target_topic._insert_payload_in_current_transaction(target_payload)
            source._conn.execute(
                """
                UPDATE messages
                SET handled_at = ?, status = 'handled'
                WHERE id = ? AND topic = ?
                """,
                (_utc_now(), record.id, self.source_topic.name),
            )
        source._emit(
            "message_handled", topic=self.source_topic.name, id=record.id
        )
    source._emit(
        "pipeline_end",
        source=self.source_topic.name,
        target=self.target_name,
        count=len(results),
        dry_run=dry_run,
    )
    return results

def to(self, target: Topic[OutputT] | str) -> Pipeline[ModelT, OutputT]:
    """Designate the target topic for this pipeline.

    Args:
        target: Either a ``Topic`` instance or the string name of a
            topic already registered on the source's ``Source``.

    Returns:
        ``self`` — so calls can be chained
        (``topic.pipe(fn).to(target).run()``).

    Raises:
        KeyError: ``target`` is a string and no topic by that name is
            registered on the underlying ``Source`` (call
            ``Source.topic(name, model)`` first to register it).
    """
    if isinstance(target, str):
        target = self.source_topic.source._registered_topic(target)
    self.target_topic = target
    return self