Python DSL reference pipeline.map(...) | connector.route(...) | project.int_type(...)

Typed factories

Concrete methods constrain configuration at the Python call site instead of accepting a free-form type field.

When to use

Use concrete factories for all new declarations; they prevent properties from being applied to incompatible entity types.

Behavior

A Pipeline creates concrete stream operators such as map, filter, merge, input, and sink.
A connector creates only endpoints supported by its transport: HTTP route, gRPC method, Kafka topic, Cron schedule, or Temporal Activity/Workflow.
Project type factories encode primitive, container, struct, message, and custom contracts.
Python enums replace raw configuration strings for call semantics, methods, languages, adapters, schedules, and runtime options.

Pipeline stream methods properties

The method selects the stream type.

PropertyTypeDescription

input(), sink()boundary

Consume from or publish to an Endpoint.

map(), filter(), process()transformation

Apply typed business behavior.

flat_map(), flat_map_iterable(), key_by()transformation

Change cardinality or keying.

join(), multi_join(), merge()multi-input

Combine multiple flows.

split(), case(), when()routing

Branch or select flows.

delay(), cycle_link(), error()control

Delay, cycle, or failure-flow operators.

Connector endpoint methods properties

Each method exposes only its supported settings.

PropertyTypeDescription

HttpConnector.route()HTTP

Method, path, function, and tracing.

GrpcConnector.method()gRPC

Method name, streaming mode, function, and tracing.

KafkaConnector.topic()Kafka

Topic, consumer group, partitions, replication, and partitioner settings.

CronConnector.schedule()Cron

Five-field schedule, timezone, and callback.

TemporalConnector.activity()Temporal

Activity task queue, concurrency, retries, heartbeat, and optional schedule.

TemporalConnector.workflow()Temporal

Workflow task queue, concurrency, retries, execution timeout, and optional schedule.

Python example

order_type = project.struct_type("Order", fields={"id": string_type})

orders = project.kafka_connector("Orders", brokers="localhost:9092")
created = orders.topic(
    "Order Created",
    function=Function(name="OrderCreated", module=LOCAL_MODULE),
    topic="orders.created",
)

pipeline = service.pipeline("orders")
incoming = pipeline.input("Consume Orders", endpoint=created, value_type=order_type)