Allow decorator usage; document register_row is preferred

Author: Abacn

sdks/python/apache_beam/coders/row_coder_test.py

@@ -60,13 +60,13 @@
      ("one_more_field", typing.Optional[str])])
 
 
+@coders_registry.register_row
 class People(typing.NamedTuple):
   primary: Person
   partner: typing.Optional[Person]
 
 
-coders_registry.register_coder(Person, RowCoder)
-coders_registry.register_coder(People, RowCoder)
+coders_registry.register_row(Person)
 
 
 class RowCoderTest(unittest.TestCase):

sdks/python/apache_beam/coders/typecoders.py

@@ -124,7 +124,19 @@ def _normalize_typehint_type(typehint_type):
   def register_coder(
       self, typehint_type: Any,
       typehint_coder_class: Type[coders.Coder]) -> None:
-    "Register a user type with a coder"
+    """
+    Register a user type with a coder.
+
+    Typical usage::
+
+      class MyCustomType:
+        pass
+
+      coders.registry.register_coder(MyCustomType, MyCustomCoder)
+
+    To register a supported user type (data class or named tuple) with Beam Row,
+    use :meth:`register_row` instead, as it registers both coder and schema.
+    """
     if not isinstance(typehint_coder_class, type):
       raise TypeError(
           'Coder registration requires a coder class object. '
@@ -134,11 +146,23 @@ def register_coder(
     self._register_coder_internal(
         self._normalize_typehint_type(typehint_type), typehint_coder_class)
 
-  def register_row(self, typehint_type: Any) -> None:
+  def register_row(self, typehint_type: type[Any]) -> type[Any]:
     """
     Register a user type with a Beam Row.
 
     This registers the type with a RowCoder and register its schema.
+
+    Register a dataclass::
+
+      @typecoders.registry.register_row
+      @dataclass
+      class MyDataClass:
+        id: int
+        name: str
+
+    Register a named tuple::
+
+      coders.registry.register_row(MyNamedTuple)
     """
     from apache_beam.coders import RowCoder
     from apache_beam.typehints.schemas import typing_to_runner_api
@@ -148,6 +172,7 @@ def register_row(self, typehint_type: Any) -> None:
     # This call generated a schema id for the type and register it with
     # schema registry
     typing_to_runner_api(typehint_type)
+    return typehint_type
 
   def get_coder(self, typehint: Any) -> coders.Coder:
     if typehint and typehint.__module__ == '__main__':