{"record":{"id":"f82be864014d5404","repo":"apache/beam","slug":"either-self-class-name-process-batch-must-have-a-type","errorCode":null,"errorMessage":"Either {self.__class__.__name__}.process_batch() must have a type annotation on its first parameter, or {self.__class__.__name__} must override get_input_batch_type.","messagePattern":"Either (.+?)\\.process_batch\\(\\) must have a type annotation on its first parameter, or (.+?) must override get_input_batch_type\\.","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/transforms/core.py","lineNumber":898,"sourceCode":"    input typehint for the first parameter of ``process_batch``. A Batched DoFn\n    may override this method if a dynamic approach is required.\n\n    Args:\n      input_element_type: The **element type** of the input PCollection this\n        DoFn is being applied to.\n\n    Returns:\n      ``None`` if this DoFn cannot accept batches, else a Beam typehint or\n      a native Python typehint.\n    \"\"\"\n    if not self._process_batch_defined:\n      return None\n    input_type = list(\n        inspect.signature(self.process_batch).parameters.values())[0].annotation\n    if input_type == inspect.Signature.empty:\n      # TODO(https://github.com/apache/beam/issues/21652): Consider supporting\n      # an alternative (dynamic?) approach for declaring input type\n      raise TypeError(\n          f\"Either {self.__class__.__name__}.process_batch() must have a type \"\n          f\"annotation on its first parameter, or {self.__class__.__name__} \"\n          \"must override get_input_batch_type.\")\n    return input_type\n\n  def _get_input_batch_type_normalized(self, input_element_type):\n    return typehints.native_type_compatibility.convert_to_beam_type(\n        self.get_input_batch_type(input_element_type))\n\n  def _get_output_batch_type_normalized(self, input_element_type):\n    return typehints.native_type_compatibility.convert_to_beam_type(\n        self.get_output_batch_type(input_element_type))\n\n  @staticmethod\n  def _get_element_type_from_return_annotation(method, input_type):\n    return_type = inspect.signature(method).return_annotation\n    if return_type == inspect.Signature.empty:\n      # output type not annotated, try to infer it","sourceCodeStart":880,"sourceCodeEnd":916,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/transforms/core.py#L880-L916","documentation":"get_input_batch_type (core.py:898) infers a batched DoFn's input element type from the annotation on process_batch's first parameter. If that parameter has no annotation and the DoFn does not override get_input_batch_type, Beam has no way to know the batch input type and raises this TypeError.","triggerScenarios":"Defining `def process_batch(self, batch):` without any type annotation on `batch` in a DoFn used with batching, without overriding get_input_batch_type.","commonSituations":"Old-style DoFns written before annotations; batch conversion helpers (e.g. beam.transforms.batch) applied to DoFns whose process_batch lacks hints.","solutions":["Add a type annotation to process_batch's first parameter, e.g. def process_batch(self, batch: List[MyType]).","Override get_input_batch_type in the DoFn class to return the element type explicitly.","Annotate with a batched type like pa.Table or List[T] depending on the batch framework."],"exampleFix":"# before\nclass MyDoFn(DoFn):\n    def process_batch(self, batch):\n        ...\n\n# after\nclass MyDoFn(DoFn):\n    def process_batch(self, batch: List[MyType]):\n        ...","handlingStrategy":"validation","validationCode":"import inspect\ndef check_process_batch_annotated(cls):\n    if hasattr(cls, 'process_batch') and hasattr(type(cls).get_input_batch_type, '__func__'):\n        sig = inspect.signature(cls.process_batch)\n        first = list(sig.parameters.values())[0]\n        if first.annotation is inspect.Signature.empty:\n            raise TypeError(f'{cls.__class__.__name__}.process_batch first param needs a type annotation')","typeGuard":"def batch_input_annotated(cls) -> bool:\n    import inspect\n    params = list(inspect.signature(cls.process_batch).parameters.values())\n    return bool(params) and params[0].annotation is not inspect.Signature.empty","tryCatchPattern":"try:\n    out = dofn.get_input_batch_type()\nexcept TypeError as e:\n    if 'get_input_batch_type' in str(e):\n        raise TypeError('Annotate process_batch(self, batch: List[T]) or override get_input_batch_type') from e\n    raise","preventionTips":["Always annotate the first process_batch parameter (e.g. List[T], pa.Table, pd.DataFrame)","Or override get_input_batch_type in the DoFn class","Lint DoFn classes with a custom flake8 plugin requiring return/param annotations on process_batch"],"tags":["python","apache-beam","type-annotation","typeerror"],"backgroundTag":"missing-required-argument","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}