heronpy/api/topology.py - incubator-heron - Git at Google

 #!/usr/bin/env python
 # -*- encoding: utf-8 -*-

 #  Licensed to the Apache Software Foundation (ASF) under one
 #  or more contributor license agreements.  See the NOTICE file
 #  distributed with this work for additional information
 #  regarding copyright ownership.  The ASF licenses this file
 #  to you under the Apache License, Version 2.0 (the
 #  "License"); you may not use this file except in compliance
 #  with the License.  You may obtain a copy of the License at
 #
 #    http://www.apache.org/licenses/LICENSE-2.0
 #
 #  Unless required by applicable law or agreed to in writing,
 #  software distributed under the License is distributed on an
 #  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.

 """
 topology.py: module for defining Heron topologies in Python
 """

 import os
 import uuid

 import heronpy.api.api_constants as api_constants
 import six
 from heronpy.api.component.component_spec import HeronComponentSpec
 from heronpy.api.serializer import default_serializer
 from heronpy.proto import topology_pb2


 class TopologyType(type):
   """Metaclass to define a Heron topology in Python"""
   DEFAULT_TOPOLOGY_CONFIG = {
       api_constants.TOPOLOGY_DEBUG: "false",
       api_constants.TOPOLOGY_STMGRS: "1",
       api_constants.TOPOLOGY_MESSAGE_TIMEOUT_SECS: "30",
       api_constants.TOPOLOGY_COMPONENT_PARALLELISM: "1",
       api_constants.TOPOLOGY_MAX_SPOUT_PENDING: "100",
       api_constants.TOPOLOGY_RELIABILITY_MODE: api_constants.TopologyReliabilityMode.ATMOST_ONCE,
       api_constants.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS: "true"}
   def __new__(mcs, classname, bases, class_dict):
     bolt_specs = {}
     spout_specs = {}
     # Copy HeronComponentSpec items out of class_dict
     specs = TopologyType.class_dict_to_specs(class_dict)
     for spec in iter(list(specs.values())):
       if spec.is_spout:
         TopologyType.add_spout_specs(spec, spout_specs)
       else:
         TopologyType.add_bolt_specs(spec, bolt_specs)
     if classname != 'Topology' and not spout_specs:
       raise ValueError("A Topology requires at least one Spout")

     topology_config = TopologyType.class_dict_to_topo_config(class_dict)

     if classname != 'Topology':
       class_dict['_topo_config'] = topology_config
       class_dict['_protobuf_bolts'] = bolt_specs
       class_dict['_protobuf_spouts'] = spout_specs
       class_dict['_heron_specs'] = list(specs.values())

       # create topology protobuf here
       TopologyType.init_topology(classname, class_dict)

     return type.__new__(mcs, classname, bases, class_dict)

   @classmethod
   def class_dict_to_specs(mcs, class_dict):
     """Takes a class `__dict__` and returns `HeronComponentSpec` entries"""
     specs = {}

     for name, spec in list(class_dict.items()):
       if isinstance(spec, HeronComponentSpec):
         # Use the variable name as the specification name.
         if spec.name is None:
           spec.name = name
         if spec.name in specs:
           raise ValueError("Duplicate component name: %s" % spec.name)
         else:
           specs[spec.name] = spec
     return specs

   @classmethod
   def class_dict_to_topo_config(mcs, class_dict):
     """
     Takes a class `__dict__` and returns a map containing topology-wide
     configuration.

     The returned dictionary is a sanitized `dict` of type `<str ->
     (str|object)>`.

     This classmethod firsts insert default topology configuration and then
     overrides it with a given topology-wide configuration. Note that this
     configuration will be overriden by a component-specific configuration at
     runtime.
     """
     topo_config = {}

     # add defaults
     topo_config.update(mcs.DEFAULT_TOPOLOGY_CONFIG)

     for name, custom_config in list(class_dict.items()):
       if name == 'config' and isinstance(custom_config, dict):
         sanitized_dict = mcs._sanitize_config(custom_config)
         topo_config.update(sanitized_dict)

     return topo_config

   @classmethod
   def add_spout_specs(mcs, spec, spout_specs):
     if not spec.outputs:
       raise ValueError(
           "%s: %s requires at least one output, because it is a spout" %
           (spec.python_class_path, spec.name))
     spout_specs[spec.name] = spec.get_protobuf()

   @classmethod
   def add_bolt_specs(mcs, spec, bolt_specs):
     if not spec.inputs:
       raise ValueError(
           "%s: %s requires at least one input, because it is a bolt" %
           (spec.python_class_path, spec.name))
     bolt_specs[spec.name] = spec.get_protobuf()

   @classmethod
   def get_topology_config_protobuf(mcs, class_dict):
     config = topology_pb2.Config()
     conf_dict = class_dict['_topo_config']

     for key, value in list(conf_dict.items()):
       if isinstance(value, str):
         kvs = config.kvs.add()
         kvs.key = key
         kvs.value = value
         kvs.type = topology_pb2.ConfigValueType.Value("STRING_VALUE")
       else:
         # need to serialize
         kvs = config.kvs.add()
         kvs.key = key
         kvs.serialized_value = default_serializer.serialize(value)
         kvs.type = topology_pb2.ConfigValueType.Value("PYTHON_SERIALIZED_VALUE")

     return config

   @classmethod
   def init_topology(mcs, classname, class_dict):
     """Initializes a topology protobuf"""
     if classname == 'Topology':
       # Base class can't initialize protobuf
       return
     heron_options = TopologyType.get_heron_options_from_env()
     initial_state = heron_options.get("cmdline.topology.initial.state", "RUNNING")
     tmp_directory = heron_options.get("cmdline.topologydefn.tmpdirectory")
     if tmp_directory is None:
       raise RuntimeError("Topology definition temp directory not specified")

     topology_name = heron_options.get("cmdline.topology.name", classname)
     topology_id = topology_name + str(uuid.uuid4())

     # create protobuf
     topology = topology_pb2.Topology()
     topology.id = topology_id
     topology.name = topology_name
     topology.state = topology_pb2.TopologyState.Value(initial_state)
     topology.topology_config.CopyFrom(TopologyType.get_topology_config_protobuf(class_dict))

     TopologyType.add_bolts_and_spouts(topology, class_dict)

     class_dict['topology_name'] = topology_name
     class_dict['topology_id'] = topology_id
     class_dict['protobuf_topology'] = topology
     class_dict['topologydefn_tmpdir'] = tmp_directory
     class_dict['heron_runtime_options'] = heron_options

   @staticmethod
   def get_heron_options_from_env():
     """Retrieves heron options from the `HERON_OPTIONS` environment variable.

     Heron options have the following format:

         cmdline.topologydefn.tmpdirectory=/var/folders/tmpdir
         cmdline.topology.initial.state=PAUSED

     In this case, the returned map will contain:

         #!json
         {
           "cmdline.topologydefn.tmpdirectory": "/var/folders/tmpdir",
           "cmdline.topology.initial.state": "PAUSED"
         }

     Currently supports the following options natively:

     - `cmdline.topologydefn.tmpdirectory`: (required) the directory to which this
     topology's defn file is written
     - `cmdline.topology.initial.state`: (default: "RUNNING") the initial state of the topology
     - `cmdline.topology.name`: (default: class name) topology name on deployment

     Returns: map mapping from key to value
     """
     heron_options_raw = os.environ.get("HERON_OPTIONS")
     if heron_options_raw is None:
       raise RuntimeError("HERON_OPTIONS environment variable not found")

     options = {}
     for option_line in heron_options_raw.replace("%%%%", " ").split(','):
       key, sep, value = option_line.partition("=")
       if sep:
         options[key] = value
       else:
         raise ValueError("Invalid HERON_OPTIONS part %r" % option_line)
     return options

   @classmethod
   def add_bolts_and_spouts(mcs, topology, class_dict):
     spouts = list(class_dict["_protobuf_spouts"].values())
     bolts = list(class_dict["_protobuf_bolts"].values())

     for spout in spouts:
       added = topology.spouts.add()
       added.CopyFrom(spout)
     for bolt in bolts:
       added = topology.bolts.add()
       added.CopyFrom(bolt)

   @staticmethod
   def _sanitize_config(custom_config):
     """Checks whether a given custom_config and returns a sanitized dict <str -> (str|object)>

     It checks if keys are all strings and sanitizes values of a given dictionary as follows:

     - If string, number or boolean is given as a value, it is converted to string.
       For string and number (int, float), it is converted to string by a built-in ``str()`` method.
       For a boolean value, ``True`` is converted to "true" instead of "True", and ``False`` is
       converted to "false" instead of "False", in order to keep the consistency with
       Java configuration.

     - If neither of the above is given as a value, it is inserted into the sanitized dict as it is.
       These values will need to be serialized before adding to a protobuf message.
     """
     sanitized = {}
     for key, value in list(custom_config.items()):
       if not isinstance(key, str):
         raise TypeError("Key for topology-wide configuration must be string, given: %s: %s"
                         % (str(type(key)), str(key)))

       if isinstance(value, bool):
         sanitized[key] = "true" if value else "false"
       elif isinstance(value, (str, int, float)):
         sanitized[key] = str(value)
       else:
         sanitized[key] = value

     return sanitized

 @six.add_metaclass(TopologyType)
 class Topology(object):
   """Topology is an abstract class for defining a topology

   Topology writers can define their custom topology by inheriting this class.
   The usage of this class is compatible with StreamParse API.

   Defining a topology is simple. Topology writers need to create a subclass, in which information
   about the components in their topology and how they connect to each other are specified
   by placing `HeronComponentSpec` as class instances.
   For more information, refer to `spec()` method of both the `Bolt` and `Spout` classes.

   In addition, you can also set a topology-wide configuration, by adding a ``config`` class
   attribute to your topology class, that is dict mapping from option names to their values.
   Note that topology-wide configurations are overridden by component-specific configurations
   that might be specified from ``spec()`` method of ``Bolt`` or ``Spout`` class.

   **Example**: A sample `WordCountTopology` can be defined as follows:

     #!python
     from heronpy.api.src.python import Topology
     from heronpy.examples.src.python import WordSpout, CountBolt

     class WordCount(Topology):
         config = {"topology.wide.config": "some value"}

         word_spout = WordSpout.spec(par=1)
         count_bolt = CountBolt.spec(par=1,
                                     inputs={word_spout: Grouping.fields('word')},
                                     config={"count_bolt.specific.config": "another value"})
   """

   # pylint: disable=no-member
   @classmethod
   def write(cls):
     """Writes the Topology .defn file to ``dest``

     This classmethod is meant be used by heron-cli when submitting a topology.
     """
     if cls.__name__ == 'Topology':
       raise ValueError("The base Topology class cannot be writable")
     filename = "%s.defn" % cls.topology_name
     path = os.path.join(cls.topologydefn_tmpdir, filename)

     with open(path, 'wb') as f:
       f.write(cls.protobuf_topology.SerializeToString())

 class TopologyBuilder(object):
   """Builder for heronpy.api.src.python topology

   This class dynamically creates a subclass of `Topology` with given spouts and
   bolts and writes its definition files when `build_and_submit()` is called.

   **Example**: A sample `WordCountTopology` can be defined as follows:

     #!python
     import sys
     from heronpy.api.src.python import TopologyBuilder
     from heronpy.examples.spout import WordSpout
     from heronpy.examples.bolt import CountBolt

     if __name__ == '__main__':
         builder = TopologyBuilder(name=sys.argv[1])
         word_spout = builder.add_spout("word-spout", WordSpout, 2)

         builder.add_bolt("count-bolt", CountBolt, 2,
                          inputs={word_spout: Grouping.fields('word')},
                          config={"count_bolt.specific.config": "some value"})

         builder.build_and_submit()
   """
   def __init__(self, name):
     """Initialize this TopologyBuilder

     :type name: str
     :param name: topology name
     """
     assert name is not None and isinstance(name, str) and name != "Topology"

     self.topology_name = name

     self._specs = {}
     self._topology_config = {}

   def add_spec(self, *specs):
     """Add specs to the topology

     :type specs: HeronComponentSpec
     :param specs: specs to add to the topology
     """
     for spec in specs:
       if not isinstance(spec, HeronComponentSpec):
         raise TypeError("Argument to add_spec needs to be HeronComponentSpec, given: %s"
                         % str(spec))
       if spec.name is None:
         raise ValueError("TopologyBuilder cannot take a spec without name")
       if spec.name == "config":
         raise ValueError("config is a reserved name")
       if spec.name in self._specs:
         raise ValueError("Attempting to add duplicate spec name: %r %r" % (spec.name, spec))

       self._specs[spec.name] = spec

   def add_spout(self, name, spout_cls, par, config=None, optional_outputs=None):
     """Add a spout to the topology"""
     spout_spec = spout_cls.spec(name=name, par=par, config=config,
                                 optional_outputs=optional_outputs)
     self.add_spec(spout_spec)
     return spout_spec

   def add_bolt(self, name, bolt_cls, par, inputs, config=None, optional_outputs=None):
     """Add a bolt to the topology"""
     bolt_spec = bolt_cls.spec(name=name, par=par, inputs=inputs, config=config,
                               optional_outputs=optional_outputs)
     self.add_spec(bolt_spec)
     return bolt_spec

   def set_config(self, config):
     """Set topology-wide configuration to the topology

     :type config: dict
     :param config: topology-wide config
     """
     if not isinstance(config, dict):
       raise TypeError("Argument to set_config needs to be dict, given: %s" % str(config))
     self._topology_config = config

   def _construct_topo_class_dict(self):
     class_dict = self._specs.copy()
     class_dict["config"] = self._topology_config
     return class_dict

   def build_and_submit(self):
     """Builds the topology and submits to the destination"""
     class_dict = self._construct_topo_class_dict()
     topo_cls = TopologyType(self.topology_name, (Topology,), class_dict)
     topo_cls.write()
	#!/usr/bin/env python
	# -- encoding: utf-8 --

	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.

	"""
	topology.py: module for defining Heron topologies in Python
	"""

	import os
	import uuid

	import heronpy.api.api_constants as api_constants
	import six
	from heronpy.api.component.component_spec import HeronComponentSpec
	from heronpy.api.serializer import default_serializer
	from heronpy.proto import topology_pb2


	class TopologyType(type):
	"""Metaclass to define a Heron topology in Python"""
	DEFAULT_TOPOLOGY_CONFIG = {
	api_constants.TOPOLOGY_DEBUG: "false",
	api_constants.TOPOLOGY_STMGRS: "1",
	api_constants.TOPOLOGY_MESSAGE_TIMEOUT_SECS: "30",
	api_constants.TOPOLOGY_COMPONENT_PARALLELISM: "1",
	api_constants.TOPOLOGY_MAX_SPOUT_PENDING: "100",
	api_constants.TOPOLOGY_RELIABILITY_MODE: api_constants.TopologyReliabilityMode.ATMOST_ONCE,
	api_constants.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS: "true"}
	def __new__(mcs, classname, bases, class_dict):
	bolt_specs = {}
	spout_specs = {}
	# Copy HeronComponentSpec items out of class_dict
	specs = TopologyType.class_dict_to_specs(class_dict)
	for spec in iter(list(specs.values())):
	if spec.is_spout:
	TopologyType.add_spout_specs(spec, spout_specs)
	else:
	TopologyType.add_bolt_specs(spec, bolt_specs)
	if classname != 'Topology' and not spout_specs:
	raise ValueError("A Topology requires at least one Spout")

	topology_config = TopologyType.class_dict_to_topo_config(class_dict)

	if classname != 'Topology':
	class_dict['_topo_config'] = topology_config
	class_dict['_protobuf_bolts'] = bolt_specs
	class_dict['_protobuf_spouts'] = spout_specs
	class_dict['_heron_specs'] = list(specs.values())

	# create topology protobuf here
	TopologyType.init_topology(classname, class_dict)

	return type.__new__(mcs, classname, bases, class_dict)

	@classmethod
	def class_dict_to_specs(mcs, class_dict):
	"""Takes a class `__dict__` and returns `HeronComponentSpec` entries"""
	specs = {}

	for name, spec in list(class_dict.items()):
	if isinstance(spec, HeronComponentSpec):
	# Use the variable name as the specification name.
	if spec.name is None:
	spec.name = name
	if spec.name in specs:
	raise ValueError("Duplicate component name: %s" % spec.name)
	else:
	specs[spec.name] = spec
	return specs

	@classmethod
	def class_dict_to_topo_config(mcs, class_dict):
	"""
	Takes a class `__dict__` and returns a map containing topology-wide
	configuration.

	The returned dictionary is a sanitized `dict` of type `<str ->
	(str\|object)>`.

	This classmethod firsts insert default topology configuration and then
	overrides it with a given topology-wide configuration. Note that this
	configuration will be overriden by a component-specific configuration at
	runtime.
	"""
	topo_config = {}

	# add defaults
	topo_config.update(mcs.DEFAULT_TOPOLOGY_CONFIG)

	for name, custom_config in list(class_dict.items()):
	if name == 'config' and isinstance(custom_config, dict):
	sanitized_dict = mcs._sanitize_config(custom_config)
	topo_config.update(sanitized_dict)

	return topo_config

	@classmethod
	def add_spout_specs(mcs, spec, spout_specs):
	if not spec.outputs:
	raise ValueError(
	"%s: %s requires at least one output, because it is a spout" %
	(spec.python_class_path, spec.name))
	spout_specs[spec.name] = spec.get_protobuf()

	@classmethod
	def add_bolt_specs(mcs, spec, bolt_specs):
	if not spec.inputs:
	raise ValueError(
	"%s: %s requires at least one input, because it is a bolt" %
	(spec.python_class_path, spec.name))
	bolt_specs[spec.name] = spec.get_protobuf()

	@classmethod
	def get_topology_config_protobuf(mcs, class_dict):
	config = topology_pb2.Config()
	conf_dict = class_dict['_topo_config']

	for key, value in list(conf_dict.items()):
	if isinstance(value, str):
	kvs = config.kvs.add()
	kvs.key = key
	kvs.value = value
	kvs.type = topology_pb2.ConfigValueType.Value("STRING_VALUE")
	else:
	# need to serialize
	kvs = config.kvs.add()
	kvs.key = key
	kvs.serialized_value = default_serializer.serialize(value)
	kvs.type = topology_pb2.ConfigValueType.Value("PYTHON_SERIALIZED_VALUE")

	return config

	@classmethod
	def init_topology(mcs, classname, class_dict):
	"""Initializes a topology protobuf"""
	if classname == 'Topology':
	# Base class can't initialize protobuf
	return
	heron_options = TopologyType.get_heron_options_from_env()
	initial_state = heron_options.get("cmdline.topology.initial.state", "RUNNING")
	tmp_directory = heron_options.get("cmdline.topologydefn.tmpdirectory")
	if tmp_directory is None:
	raise RuntimeError("Topology definition temp directory not specified")

	topology_name = heron_options.get("cmdline.topology.name", classname)
	topology_id = topology_name + str(uuid.uuid4())

	# create protobuf
	topology = topology_pb2.Topology()
	topology.id = topology_id
	topology.name = topology_name
	topology.state = topology_pb2.TopologyState.Value(initial_state)
	topology.topology_config.CopyFrom(TopologyType.get_topology_config_protobuf(class_dict))

	TopologyType.add_bolts_and_spouts(topology, class_dict)

	class_dict['topology_name'] = topology_name
	class_dict['topology_id'] = topology_id
	class_dict['protobuf_topology'] = topology
	class_dict['topologydefn_tmpdir'] = tmp_directory
	class_dict['heron_runtime_options'] = heron_options

	@staticmethod
	def get_heron_options_from_env():
	"""Retrieves heron options from the `HERON_OPTIONS` environment variable.

	Heron options have the following format:

	cmdline.topologydefn.tmpdirectory=/var/folders/tmpdir
	cmdline.topology.initial.state=PAUSED

	In this case, the returned map will contain:

	#!json
	{
	"cmdline.topologydefn.tmpdirectory": "/var/folders/tmpdir",
	"cmdline.topology.initial.state": "PAUSED"
	}

	Currently supports the following options natively:

	- `cmdline.topologydefn.tmpdirectory`: (required) the directory to which this
	topology's defn file is written
	- `cmdline.topology.initial.state`: (default: "RUNNING") the initial state of the topology
	- `cmdline.topology.name`: (default: class name) topology name on deployment

	Returns: map mapping from key to value
	"""
	heron_options_raw = os.environ.get("HERON_OPTIONS")
	if heron_options_raw is None:
	raise RuntimeError("HERON_OPTIONS environment variable not found")

	options = {}
	for option_line in heron_options_raw.replace("%%%%", " ").split(','):
	key, sep, value = option_line.partition("=")
	if sep:
	options[key] = value
	else:
	raise ValueError("Invalid HERON_OPTIONS part %r" % option_line)
	return options

	@classmethod
	def add_bolts_and_spouts(mcs, topology, class_dict):
	spouts = list(class_dict["_protobuf_spouts"].values())
	bolts = list(class_dict["_protobuf_bolts"].values())

	for spout in spouts:
	added = topology.spouts.add()
	added.CopyFrom(spout)
	for bolt in bolts:
	added = topology.bolts.add()
	added.CopyFrom(bolt)

	@staticmethod
	def _sanitize_config(custom_config):
	"""Checks whether a given custom_config and returns a sanitized dict <str -> (str\|object)>

	It checks if keys are all strings and sanitizes values of a given dictionary as follows:

	- If string, number or boolean is given as a value, it is converted to string.
	For string and number (int, float), it is converted to string by a built-in ``str()`` method.
	For a boolean value, ``True`` is converted to "true" instead of "True", and ``False`` is
	converted to "false" instead of "False", in order to keep the consistency with
	Java configuration.

	- If neither of the above is given as a value, it is inserted into the sanitized dict as it is.
	These values will need to be serialized before adding to a protobuf message.
	"""
	sanitized = {}
	for key, value in list(custom_config.items()):
	if not isinstance(key, str):
	raise TypeError("Key for topology-wide configuration must be string, given: %s: %s"
	% (str(type(key)), str(key)))

	if isinstance(value, bool):
	sanitized[key] = "true" if value else "false"
	elif isinstance(value, (str, int, float)):
	sanitized[key] = str(value)
	else:
	sanitized[key] = value

	return sanitized

	@six.add_metaclass(TopologyType)
	class Topology(object):
	"""Topology is an abstract class for defining a topology

	Topology writers can define their custom topology by inheriting this class.
	The usage of this class is compatible with StreamParse API.

	Defining a topology is simple. Topology writers need to create a subclass, in which information
	about the components in their topology and how they connect to each other are specified
	by placing `HeronComponentSpec` as class instances.
	For more information, refer to `spec()` method of both the `Bolt` and `Spout` classes.

	In addition, you can also set a topology-wide configuration, by adding a ``config`` class
	attribute to your topology class, that is dict mapping from option names to their values.
	Note that topology-wide configurations are overridden by component-specific configurations
	that might be specified from ``spec()`` method of ``Bolt`` or ``Spout`` class.

	Example: A sample `WordCountTopology` can be defined as follows:

	#!python
	from heronpy.api.src.python import Topology
	from heronpy.examples.src.python import WordSpout, CountBolt

	class WordCount(Topology):
	config = {"topology.wide.config": "some value"}

	word_spout = WordSpout.spec(par=1)
	count_bolt = CountBolt.spec(par=1,
	inputs={word_spout: Grouping.fields('word')},
	config={"count_bolt.specific.config": "another value"})
	"""

	# pylint: disable=no-member
	@classmethod
	def write(cls):
	"""Writes the Topology .defn file to ``dest``

	This classmethod is meant be used by heron-cli when submitting a topology.
	"""
	if cls.__name__ == 'Topology':
	raise ValueError("The base Topology class cannot be writable")
	filename = "%s.defn" % cls.topology_name
	path = os.path.join(cls.topologydefn_tmpdir, filename)

	with open(path, 'wb') as f:
	f.write(cls.protobuf_topology.SerializeToString())

	class TopologyBuilder(object):
	"""Builder for heronpy.api.src.python topology

	This class dynamically creates a subclass of `Topology` with given spouts and
	bolts and writes its definition files when `build_and_submit()` is called.

	Example: A sample `WordCountTopology` can be defined as follows:

	#!python
	import sys
	from heronpy.api.src.python import TopologyBuilder
	from heronpy.examples.spout import WordSpout
	from heronpy.examples.bolt import CountBolt

	if __name__ == '__main__':
	builder = TopologyBuilder(name=sys.argv[1])
	word_spout = builder.add_spout("word-spout", WordSpout, 2)

	builder.add_bolt("count-bolt", CountBolt, 2,
	inputs={word_spout: Grouping.fields('word')},
	config={"count_bolt.specific.config": "some value"})

	builder.build_and_submit()
	"""
	def __init__(self, name):
	"""Initialize this TopologyBuilder

	:type name: str
	:param name: topology name
	"""
	assert name is not None and isinstance(name, str) and name != "Topology"

	self.topology_name = name

	self._specs = {}
	self._topology_config = {}

	def add_spec(self, *specs):
	"""Add specs to the topology

	:type specs: HeronComponentSpec
	:param specs: specs to add to the topology
	"""
	for spec in specs:
	if not isinstance(spec, HeronComponentSpec):
	raise TypeError("Argument to add_spec needs to be HeronComponentSpec, given: %s"
	% str(spec))
	if spec.name is None:
	raise ValueError("TopologyBuilder cannot take a spec without name")
	if spec.name == "config":
	raise ValueError("config is a reserved name")
	if spec.name in self._specs:
	raise ValueError("Attempting to add duplicate spec name: %r %r" % (spec.name, spec))

	self._specs[spec.name] = spec

	def add_spout(self, name, spout_cls, par, config=None, optional_outputs=None):
	"""Add a spout to the topology"""
	spout_spec = spout_cls.spec(name=name, par=par, config=config,
	optional_outputs=optional_outputs)
	self.add_spec(spout_spec)
	return spout_spec

	def add_bolt(self, name, bolt_cls, par, inputs, config=None, optional_outputs=None):
	"""Add a bolt to the topology"""
	bolt_spec = bolt_cls.spec(name=name, par=par, inputs=inputs, config=config,
	optional_outputs=optional_outputs)
	self.add_spec(bolt_spec)
	return bolt_spec

	def set_config(self, config):
	"""Set topology-wide configuration to the topology

	:type config: dict
	:param config: topology-wide config
	"""
	if not isinstance(config, dict):
	raise TypeError("Argument to set_config needs to be dict, given: %s" % str(config))
	self._topology_config = config

	def _construct_topo_class_dict(self):
	class_dict = self._specs.copy()
	class_dict["config"] = self._topology_config
	return class_dict

	def build_and_submit(self):
	"""Builds the topology and submits to the destination"""
	class_dict = self._construct_topo_class_dict()
	topo_cls = TopologyType(self.topology_name, (Topology,), class_dict)
	topo_cls.write()