Creating Custom Probes
=====================

metapod is extensible. Add your own probes for internal systems, custom APIs,
or niche data sources.

The BaseProbe interface
-----------------------

Every probe implements three methods:

.. code-block:: python

   from core.base_probe import BaseProbe
   from core.registry import register

   @register
   class MySourceTables(BaseProbe):
       @staticmethod
       def probe_name() -> str:
           """Fully qualified name: source.category.leaf"""
           return "mysource.metadata.tables"

       @staticmethod
       def description() -> str:
           """One-line description shown in probe catalog."""
           return "Extract table inventory from MySource API"

       def execute(self, connection, schema: str, **kwargs) -> list[dict]:
           """Run the probe. Return list of record dicts."""
           # connection = whatever your connection module returns
           # schema = the --schema CLI argument
           response = connection.get(f"/api/tables?schema={schema}")
           return [
               {
                   "table_name": t["name"],
                   "row_count": t["rows"],
                   "created": t["created_at"],
               }
               for t in response.json()
           ]

The ``@register`` decorator adds the probe to the global registry. It will
appear in ``metapod list-probes`` and can be matched by glob patterns.

Connection module
-----------------

Each source needs a connection module at ``probes/<source>/connection.py``:

.. code-block:: python

   # probes/mysource/connection.py
   import httpx
   from core.config import SourceConfig

   def connect(config: SourceConfig) -> httpx.Client:
       """Create a connection to MySource."""
       return httpx.Client(
           base_url=config.host,
           headers={"Authorization": f"Bearer {config.token}"},
           timeout=30,
       )

   def source_id(config: SourceConfig) -> str:
       """Return a human-readable source identifier."""
       return f"mysource://{config.host}"

Register the connection in ``core/connections.py``:

.. code-block:: python

   CONNECTION_REGISTRY["mysource"] = ("probes.mysource.connection", "connect", "source_id")

File structure
--------------

.. code-block:: text

   probes/
     mysource/
       __init__.py          # empty
       connection.py        # connect() + source_id()
       metadata/
         __init__.py        # empty
         tables.py          # @register class MySourceTables(BaseProbe)
         columns.py         # @register class MySourceColumns(BaseProbe)

ProbeResult envelope
--------------------

``BaseProbe.run()`` wraps your ``execute()`` with timing, error handling,
and the standard envelope:

.. code-block:: python

   result = probe.run(connection, schema, source_id="mysource://host")
   # result.probe       = "mysource.metadata.tables"
   # result.source_type = "mysource"
   # result.source_id   = "mysource://host"
   # result.timestamp   = "2026-03-29T12:00:00+00:00"
   # result.records     = [{...}, {...}]
   # result.record_count = 2
   # result.duration_ms  = 45
   # result.error        = None

If ``execute()`` raises, ``result.error`` contains the exception message and
``result.records`` is empty.

Plugin system (entry points)
----------------------------

External packages can register probes without modifying metapod source code.

In your package's ``pyproject.toml``:

.. code-block:: toml

   [project.entry-points."metapod.probes"]
   mysource = "my_metapod_plugin.probes"

metapod discovers entry points at startup and registers all ``BaseProbe``
subclasses found in the specified module.

Testing your probe
------------------

.. code-block:: python

   # tests/test_mysource.py
   from unittest.mock import MagicMock
   from probes.mysource.metadata.tables import MySourceTables

   def test_returns_tables():
       mock_conn = MagicMock()
       mock_conn.get.return_value.json.return_value = [
           {"name": "users", "rows": 100, "created_at": "2026-01-01"},
       ]
       probe = MySourceTables()
       result = probe.run(mock_conn, "default")
       assert result.record_count == 1
       assert result.records[0]["table_name"] == "users"

Run with:

.. code-block:: bash

   python3 -m pytest tests/test_mysource.py -v