{"record":{"id":"b207edea96e6da40","repo":"apache/beam","slug":"notimplementederror-type-self-base","errorCode":null,"errorMessage":"NotImplementedError(type(self))","messagePattern":"NotImplementedError\\(type\\(self\\)\\)","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/ml/rag/ingestion/base.py","lineNumber":60,"sourceCode":"    ...     def create_write_transform(self):\n    ...         return beam.io.WriteToBigQuery(\n    ...             table=self.table\n    ...         )\n  \"\"\"\n  @abstractmethod\n  def create_write_transform(self) -> beam.PTransform[EmbeddableItem, Any]:\n    \"\"\"Creates a PTransform that writes embeddings to the vector database.\n\n    Returns:\n        A PTransform that accepts PCollection[EmbeddableItem]\n        and writes the embeddings\n        and metadata to the configured vector database.\n        The transform should handle:\n        - Converting EmbeddableItem format to database schema\n        - Setting up database connection/client\n        - Writing with appropriate batching/error handling\n    \"\"\"\n    raise NotImplementedError(type(self))\n\n\nclass VectorDatabaseWriteTransform(beam.PTransform):\n  \"\"\"A PTransform for writing embedded chunks to vector databases.\n  \n  This transform uses a VectorDatabaseWriteConfig to write chunks with\n  embeddings to vector database. It handles validating the config and applying\n  the database-specific write transform.\n\n  Example usage:\n    >>> config = BigQueryVectorConfig(\n    ...     table='project.dataset.embeddings',\n    ...     embedding_column='embedding'\n    ... )\n    >>>\n    >>> with beam.Pipeline() as p:\n    ...     items = p | beam.Create([...])  # PCollection[EmbeddableItem]\n    ...     items | VectorDatabaseWriteTransform(config)","sourceCodeStart":42,"sourceCodeEnd":78,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/ml/rag/ingestion/base.py#L42-L78","documentation":"VectorDatabaseWriteConfig.create_write_transform is an abstract-style hook: subclasses must return a beam.PTransform that writes EmbeddableItems to the target vector DB. The base class raises NotImplementedError(type(self)) when a caller uses a config subclass that never implemented this method.","triggerScenarios":"Instantiating a custom VectorDatabaseWriteConfig subclass (or the bare base class) and passing it to VectorDatabaseWriteTransform, whose expand() calls create_write_transform().","commonSituations":"Writing a custom DB integration but forgetting to override create_write_transform; using the abstract base directly in tests; a subclass that only overrode other hooks.","solutions":["Override create_write_transform in your config subclass and return a beam.PTransform that writes to your database","Use one of the built-in configs (e.g., BigQueryVectorWriterConfig) instead of the base class","Never instantiate VectorDatabaseWriteConfig directly"],"exampleFix":"// before\nclass MyConfig(VectorDatabaseWriteConfig):\n    pass  # no create_write_transform\n// after\nclass MyConfig(VectorDatabaseWriteConfig):\n    def create_write_transform(self):\n        return _WriteToMyVectorDb(self)","handlingStrategy":"type-guard","validationCode":"import inspect\nassert not inspect.isabstract(type(cfg)) and type(cfg).create_write_transform is not VectorDatabaseWriteConfig.create_write_transform","typeGuard":"def has_write_transform(cfg) -> bool:\n    return type(cfg).create_write_transform is not VectorDatabaseWriteConfig.create_write_transform","tryCatchPattern":"try:\n    transform = cfg.create_write_transform()\nexcept NotImplementedError:\n    transform = default_bigquery_write_transform()","preventionTips":["Never instantiate the base VectorDatabaseWriteConfig directly","Override create_write_transform in every custom config subclass","Add a smoke test that builds the write transform for each config you ship"],"tags":["python","apache-beam","rag","abstract-method","vector-db"],"backgroundTag":"abstract-method-not-implemented","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}