{"record":{"id":"84377b85377a32d0","repo":"apache/beam","slug":"unable-to-extract-pvalue-inputs-from-s-either-s-does-not","errorCode":null,"errorMessage":"Unable to extract PValue inputs from %s; either %s does not accept inputs of this format, or it does not properly override _extract_input_pvalues","messagePattern":"Unable to extract PValue inputs from (.+?); either (.+?) does not accept inputs of this format, or it does not properly override _extract_input_pvalues","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/pipeline.py","lineNumber":779,"sourceCode":"            '\"auto_unique_labels\" to automatically generate unique '\n            'transform labels. Note \"auto_unique_labels\" '\n            'could cause data loss when updating a pipeline or '\n            'reloading the job state. This is not recommended for '\n            'streaming jobs.' % full_label)\n    self.applied_labels.add(full_label)\n\n    if pvalueish is None:\n      full_label = self._current_transform().full_label\n      raise TypeCheckError(\n          f'Transform \"{full_label}\" was applied to the output of '\n          f'an object of type None.')\n\n    pvalueish, inputs = transform._extract_input_pvalues(pvalueish)\n    try:\n      if not isinstance(inputs, dict):\n        inputs = {str(ix): input for (ix, input) in enumerate(inputs)}\n    except TypeError:\n      raise NotImplementedError(\n          'Unable to extract PValue inputs from %s; either %s does not accept '\n          'inputs of this format, or it does not properly override '\n          '_extract_input_pvalues' % (pvalueish, transform))\n    for t, leaf_input in inputs.items():\n      if not isinstance(leaf_input, pvalue.PValue) or not isinstance(t, str):\n        raise NotImplementedError(\n            '%s does not properly override _extract_input_pvalues, '\n            'returned %s from %s' % (transform, inputs, pvalueish))\n\n    current = AppliedPTransform(\n        self._current_transform(),\n        transform,\n        full_label,\n        inputs,\n        None,\n        annotations=self._current_annotations())\n    self._current_transform().add_part(current)\n","sourceCodeStart":761,"sourceCodeEnd":797,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/pipeline.py#L761-L797","documentation":"Beam asks the transform to decompose its input via _extract_input_pvalues. If that call raises TypeError (the transform can't iterate/consume the input format), _apply_internal raises NotImplementedError saying it cannot extract PValue inputs from the given value — either the input format is unsupported or the custom transform didn't properly override _extract_input_pvalues.","triggerScenarios":"Passing a non-PValue container (list of values, dict of raw data, plain iterable) to pipeline.apply for a transform that doesn't accept/extract it; a custom PTransform lacking a correct _extract_input_pvalues override for unusual inputs (e.g. tuple inputs).","commonSituations":"Passing raw Python collections instead of PCollections into composite transforms; custom PTransforms handling tuple/dict inputs without overriding _extract_input_pvalues; feeding side-input-style values as main inputs.","solutions":["Pass a PCollection (or PTransform-accepted PValue structure) as the input.","If your transform accepts tuples/dicts of PCollections, override _extract_input_pvalues to return (pvalueish, dict-of-leaf-pvalues).","Unwrap raw lists/dicts into PCollections with beam.Create before applying.","Check which part of the input is not a PValue; the message names the pvalueish and transform."],"exampleFix":"// before\nclass MyT(PTransform):\n  def expand(self, pcolls):  # pcolls is a tuple, no _extract_input_pvalues override\n    ...\np.apply(MyT(), (pc1, pc2))\n// after\nclass MyT(PTransform):\n  def _extract_input_pvalues(self, pvalueish):\n    return pvalueish, {'a': pvalueish[0], 'b': pvalueish[1]}\n  def expand(self, pcolls):\n    ...\np.apply(MyT(), (pc1, pc2))","handlingStrategy":"validation","validationCode":"from apache_beam.pvalue import PValue\ndef ensure_pvalue_inputs(x):\n    if isinstance(x, PValue):\n        return x\n    if isinstance(x, (tuple, list)):\n        assert all(isinstance(i, PValue) for i in x), 'non-PValue member in input'\n    else:\n        raise TypeError('input must be a PValue or tuple/list of PValues')\n    return x","typeGuard":null,"tryCatchPattern":"try:\n    out = pipeline.apply(t, pvalueish)\nexcept NotImplementedError as e:\n    if 'extract PValue inputs' in str(e):\n        raise ValueError(f'{t} cannot consume {pvalueish!r}; wrap raw data with beam.Create') from e","preventionTips":["Feed PCollections, not raw collections, into transforms","Override _extract_input_pvalues for composites with tuple/dict inputs","Use beam.Create for in-memory data"],"tags":["python","apache-beam","not-implemented","pvalue-input"],"backgroundTag":"method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}