Toolkits > com.ibm.streamsx.topology 1.5.13.__dev__ > com.ibm.streamsx.topology.python > Python Application API > Sample Application
Example code that builds and then submits a simple topology.
from streamsx.topology.topology import Topology
import streamsx.topology.context
import transform_sample_functions;
topo = Topology("transform_sample")
source = topo.source(transform_sample_functions.int_strings_transform)
i1 = source.transform(transform_sample_functions.string_to_int)
i2 = i1.transform(transform_sample_functions.AddNum(17))
i2.print()
streamsx.topology.context.submit("STANDALONE", topo.graph)
The source function is passed a function that returns an Iterable, in this case transform_sample_functions.int_strings_transform.
def int_strings_transform():
return ["325", "457", "9325"]
The first transform function is passed a function that returns an integer converted from the string object, in this case transform_sample_functions.string_to_int.
def string_to_int(t):
return int(t)
The second transform function is passed an instance of a callable class that adds 17 to the integer, in this case transform_sample_functions.AddNum(17).
class AddNum:
def __init__(self, increment):
self.increment = increment
def __call__(self, tuple):
return tuple + self.increment
When building the topology the directory com.ibm.streamsx.topology/opt/python/packages must be in $PYTHONPATH.
The sample transform_sample.py can be found under samples/python/topology/simple. After updating the PYTHONPATH, the sample can be executed using python3 transform_sample.py.
342 474 9342