# How to Extend Semantica with Custom Ingestors: A Developer's Guide

> Learn how to extend Semantica with custom ingestors. This developer's guide covers subclassing Ingestor, implementing ingest methods, and registering your ingestor for seamless integration.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-18

---

**Extending Semantica with custom ingestors requires subclassing the abstract `Ingestor` base class (or the concrete `DBIngestor` for relational sources), implementing the `ingest_table()` and `ingest_query()` methods, and registering the class in [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py) to enable discovery via the factory function.**

The open-source Semantica framework provides a modular ingestion layer for building knowledge graphs from diverse data sources. While the platform ships with built-in connectors for common databases and APIs, its plugin-based architecture makes it straightforward to extend Semantica with custom ingestors for proprietary or specialized systems. This guide demonstrates the implementation pattern using the actual source code structure from the `semantica-agi/semantica` repository.

## Understanding the Ingestor Architecture

Semantica’s data ingestion layer is built around a small set of core abstractions that enforce consistent interfaces across all data sources.

### The Base Ingestor Class

All ingestors inherit from the abstract base class defined in [`semantica/ingest/api_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/api_ingestor.py). This class establishes the public API contract that the rest of the platform expects, including methods like `ingest_table()`, `ingest_query()`, and `ingest_directory()`. These methods must return either a **list of `Document` objects** or a **pandas `DataFrame`** for tabular sources.

### The Database Ingestor Abstraction

For relational data sources, Semantica provides `DBIngestor` in [`semantica/ingest/db_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/db_ingestor.py). This concrete class implements generic helper methods—most notably `_rows_to_documents()`—that handle connection management, schema introspection, and the conversion of raw row dictionaries into the unified `Document` model. Subclassing `DBIngestor` allows you to reuse this boilerplate while only implementing source-specific connection logic.

## Step-by-Step Guide to Creating a Custom Ingestor

Follow this four-step workflow to add a new data source to the framework.

### 1. Subclass the Base Ingestor

Create a new module under `semantica/ingest/` and inherit from the appropriate base class. For relational databases, extend `DBIngestor`; for other sources, extend the generic `Ingestor`.

```python
from semantica.ingest.db_ingestor import DBIngestor
from semantica.ingest.api_ingestor import Document

class MySQLIngestor(DBIngestor):
    """Ingests data from a MySQL database."""
    def __init__(self, host: str, user: str, password: str, database: str, **kwargs):
        super().__init__(**kwargs)
        self._conn_cfg = {
            "host": host,
            "user": user,
            "password": password,
            "database": database,
        }

```

### 2. Implement Required Methods

You must implement at minimum `ingest_table()` and/or `ingest_query()`. These methods handle data retrieval and must return standardized `Document` objects.

```python
    def _get_connection(self):
        """Create a new MySQL connection on demand."""
        import mysql.connector
        return mysql.connector.connect(**self._conn_cfg)

    def ingest_table(self, table_name: str, limit: int | None = None) -> list[Document]:
        """Pull an entire table and return a list of Document objects."""
        conn = self._get_connection()
        cursor = conn.cursor(dictionary=True)
        sql = f"SELECT * FROM `{table_name}`"
        if limit:
            sql += f" LIMIT {limit}"
        cursor.execute(sql)
        rows = cursor.fetchall()
        conn.close()
        # Convert rows → Document using inherited helper

        return self._rows_to_documents(
            rows, 
            source=f"{self._conn_cfg['database']}.{table_name}"
        )

    def ingest_query(self, query: str, params: dict | None = None) -> list[Document]:
        """Execute an arbitrary SELECT query and return Documents."""
        conn = self._get_connection()
        cursor = conn.cursor(dictionary=True)
        cursor.execute(query, params or {})
        rows = cursor.fetchall()
        conn.close()
        return self._rows_to_documents(rows, source="custom_query")

```

### 3. Register in the Ingestor Registry

Add your class to the `INGESTOR_REGISTRY` dictionary in [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py). This maps a short name (e.g., `"mysql"`) to your class, enabling the factory function `from_name()` (also exposed as `get_ingestor()`) to instantiate your ingestor dynamically.

```python

# semantica/ingest/registry.py

INGESTOR_REGISTRY = {
    # built-in ingestors …

    "mysql": MySQLIngestor,   # ← newly added entry

}

```

### 4. Expose in Package Init

Import the class in [`semantica/ingest/__init__.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/__init__.py) so users can access it directly from the package root.

```python

# semantica/ingest/__init__.py

from .mysql_ingestor import MySQLIngestor

__all__ = [
    # …other ingestors…

    "MySQLIngestor",
]

```

## How the Factory Pattern Resolves Ingestors

When a user calls `get_ingestor("mysql", host="...", ...)`, Semantica consults the `INGESTOR_REGISTRY` in [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py) to resolve the name, constructs the class with the provided keyword arguments, and returns the instance. This factory approach ensures that custom ingestors behave exactly like native ones—returning standardized `Document` lists that the downstream pipeline (splitting, semantic extraction, and knowledge graph construction) can process without modification.

## Summary

- **Extend the correct base class**: Use `DBIngestor` from [`semantica/ingest/db_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/db_ingestor.py) for relational sources, or the abstract `Ingestor` from [`semantica/ingest/api_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/api_ingestor.py) for other types.
- **Return standardized types**: `ingest_table()` and `ingest_query()` must return lists of `Document` objects (or pandas DataFrames).
- **Register statically**: Add your class to `INGESTOR_REGISTRY` in [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py) to enable factory-based discovery.
- **Expose publicly**: Import the class in [`semantica/ingest/__init__.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/__init__.py) for clean public APIs.
- **Leverage helpers**: Reuse `_rows_to_documents()` and connection pooling from `DBIngestor` to minimize boilerplate.

## Frequently Asked Questions

### What base class should I use when extending Semantica with custom ingestors for databases?

Subclass `DBIngestor` from [`semantica/ingest/db_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/db_ingestor.py). This class provides `_rows_to_documents()` and other helpers for connection handling and schema introspection, allowing you to focus only on source-specific connection logic rather than reimplementing the row-to-Document conversion.

### Can I register a custom ingestor without modifying [`registry.py`](https://github.com/semantica-agi/semantica/blob/main/registry.py)?

Yes, because `INGESTOR_REGISTRY` is a runtime dictionary, you can dynamically add entries by importing the registry in your application code and adding your class directly. However, static registration in [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py) is the recommended pattern for packaged extensions that you intend to share or maintain long-term.

### What return type is required for `ingest_table()` and `ingest_query()`?

Both methods must return a **list of `Document` objects** defined in [`semantica/ingest/api_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/api_ingestor.py), or a pandas `DataFrame` for tabular sources. The `Document` model standardizes text content, metadata, and provenance information so that downstream components like the semantic extractor and graph builder can process data uniformly regardless of the original source.

### How does the factory function locate my custom ingestor?

The `from_name()` function (exposed as `get_ingestor()`) looks up the provided string key in the `INGESTOR_REGISTRY` dictionary located at [`semantica/ingest/registry.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/registry.py). If your class is properly registered under that key, the factory instantiates it with any additional keyword arguments and returns the configured ingestor ready for use.