Python API: add get_input_topics() (apache#5526)

Fixes apache#5525
atdavidpark · Oct 31, 2019 · a4c330e · a4c330e
1 parent 3d5bc0a
commit a4c330e
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 0 deletions.
diff --git a/pulsar-client-cpp/python/pulsar/functions/context.py b/pulsar-client-cpp/python/pulsar/functions/context.py
@@ -146,6 +146,11 @@ def publish(self, topic_name, message, serde_class_name="serde.IdentitySerDe", p
     """
     pass
 
+  @abstractmethod
+  def get_input_topics(self):
+    """Returns the input topics of function"""
+    pass
+
   @abstractmethod
   def get_output_topic(self):
     """Returns the output topic of function"""

diff --git a/pulsar-functions/instance/src/main/python/contextimpl.py b/pulsar-functions/instance/src/main/python/contextimpl.py
@@ -135,6 +135,9 @@ def record_metric(self, metric_name, metric_value):
 
     self.user_metrics_map[metric_name].observe(metric_value)
 
+  def get_input_topics(self):
+    return list(self.instance_config.function_details.source.inputSpecs.keys())
+
   def get_output_topic(self):
     return self.instance_config.function_details.output
 

diff --git a/site2/docs/functions-api.md b/site2/docs/functions-api.md
@@ -588,6 +588,7 @@ Method | What it provides
 `get_user_config_map` | Returns the entire user-defined config as a dict
 `record_metric` | Records a per-key [metric](#python-metrics)
 `publish` | Publishes a message to the specified Pulsar topic
+`get_input_topics` | The name(s) of the input(topics), as a list
 `get_output_topic` | The name of the output topic
 `get_output_serde_class_name` | The name of the output [SerDe](#python-serde) class
 `ack` | [Acks](reference-terminology.md#acknowledgment-ack) the message being processed to Pulsar

diff --git a/site2/docs/functions-develop.md b/site2/docs/functions-develop.md
@@ -356,6 +356,8 @@ class ContextImpl(pulsar.Context):
     ...
   def record_metric(self, metric_name, metric_value):
     ...
+  def get_input_topics(self):
+    ...
   def get_output_topic(self):
     ...
   def get_output_serde_class_name(self):