Add yaml to programming guide. (apache#30269)

sdks/python/apache_beam/transforms/combiners.py

@@ -763,6 +763,23 @@ def extract_output(self, accumulator):
     return accumulator
 
 
+@with_input_types(T)
+@with_output_types(T)
+class ConcatListCombineFn(core.CombineFn):
+  """CombineFn for concatenating lists together."""
+  def create_accumulator(self):
+    return []
+
+  def add_input(self, accumulator, element):
+    return accumulator + element
+
+  def merge_accumulators(self, accumulators):
+    return sum(accumulators, [])
+
+  def extract_output(self, accumulator):
+    return accumulator
+
+
 @with_input_types(Tuple[K, V])
 @with_output_types(Dict[K, V])
 class ToDict(CombinerWithoutDefaults):

sdks/python/apache_beam/typehints/native_type_compatibility.py

@@ -435,6 +435,9 @@ def convert_to_typing_type(typ):
   if isinstance(typ, typehints.IterableTypeConstraint):
     return typing.Iterable[convert_to_typing_type(typ.inner_type)]
   if isinstance(typ, typehints.UnionConstraint):
+    if not typ.union_types:
+      # Gracefully handle the empty union type.
+      return typing.Any
     return typing.Union[tuple(convert_to_typing_types(typ.union_types))]
   if isinstance(typ, typehints.SetTypeConstraint):
     return typing.Set[convert_to_typing_type(typ.inner_type)]

sdks/python/apache_beam/utils/timestamp.py

@@ -77,12 +77,15 @@ def of(seconds):
       Corresponding Timestamp object.
     """
 
-    if not isinstance(seconds, (int, float, Timestamp)):
-      raise TypeError(
-          'Cannot interpret %s %s as Timestamp.' % (seconds, type(seconds)))
     if isinstance(seconds, Timestamp):
       return seconds
-    return Timestamp(seconds)
+    elif isinstance(seconds, (int, float)):
+      return Timestamp(seconds)
+    elif isinstance(seconds, datetime.datetime):
+      return Timestamp.from_utc_datetime(seconds)
+    else:
+      raise TypeError(
+          'Cannot interpret %s %s as Timestamp.' % (seconds, type(seconds)))
 
   @staticmethod
   def now():