Python API reference
pipeline.substream(name, *, value_type, source=None, appearance=None)Declare a service-local SubStream with Python DSL
Declare a callable business graph with an input argument and a result-producing stream in the same service.
When to use
Use a SubStream for a meaningful reusable business stage called from code, not for every helper function or condition.
Behavior
- value_type declares the input argument T, not the returned result R.
- Connect one body consumer using entry >> body or source=entry on an ordinary operator.
- Connect the result producer with result >> entry, or pass it as source. Its output determines R. This is a return relation, not a second invocation.
- Every declared SubStream gets a typed generated accessor. Business functions need no extra dependency declaration in the model; inject a narrow provider through custom makers.
- A collector returns true when enough results have arrived. False keeps waiting; supply cancellation or a deadline for uncertain completion.
Factory arguments properties
Typed declaration and result binding.
PropertyTypeDescription
namestrHuman-readable name; the symbolic key is derived automatically.
Requiredvalue_typeTypeDefinition | DataType | strInput argument type T.
RequiredsourceStream | NoneResult producer. May be wired later with result >> entry; required before model validation.
Default:NoneappearanceAppearance | NoneOptional Designer coordinates.
Default:NonePython example
from sa_dsl import Function, Golang, Project, ServiceModule
project = Project("SubStream Example")
service = project.service("Worker", language=Golang(),
module=ServiceModule("example.com/substreams/worker"))
text = project.string_type("Text")
pipeline = service.pipeline("lookup")
entry = pipeline.substream("Lookup", value_type=text)
result = pipeline.map("Normalize", source=entry, value_type=text,
function=Function("Normalize"))
result >> entry
assert not project.validate()