Python DSL reference pipeline.map(...) | http.post(...) | 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 get/post, typed gRPC methods, 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.get() / post()HTTP

The method is selected by the factory; provide path, function, and tracing.

GrpcConnector.unary_method() and streaming methodsgRPC

The method shape is selected by the factory; provide method name, function, and tracing.

KafkaConnector.topic()Kafka

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

CronConnector.schedule()Cron

CronSchedule object and callback.

TemporalConnector.activity()Temporal

ActivityWorker, ActivityTimeouts, RetryPolicy, and optional TemporalSchedule objects.

TemporalConnector.workflow()Temporal

WorkflowWorker, WorkflowTimeouts, RetryPolicy, and optional TemporalSchedule objects.

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)