Custom Connectors
A custom connector is a Python package that contains the configuration classes and read or write code for an external system. Installing the package adds that system as a source for publishers or a destination for subscribers.
AI agents connected through the MCP Server can also write custom connectors.
File structure
The following code uses an example for an sqlite connector.
Create a directory named sqlite_connector with the following files. The Python files belong in src/tabsdata_sqlite, which is the package users will import.
pyproject.toml defines the package, its dependencies, and the entry points Tabsdata uses to find the connector.
error.py defines the errors reported when configuration is invalid or a read or write fails.
__init__.py defines the fields used in connection documents and Python functions.
_plugin.py implements the reads and writes, and registers the classes they use.
Download the complete connector directory using the button beside sqlite_connector, or copy the files below. Highlighted lines show the names, settings, and database code to adapt when building a connector for another system.
pyproject.toml
pyproject.toml defines the package name, dependencies, and entry points Tabsdata uses to load the connector.
- Set the
name,version, and `description - Add python dependencies into
dependencies
2.
[build-system]
requires = ["setuptools>=69"]
build-backend = "setuptools.build_meta"
[project]
name = "tabsdata-conn-sqlite"
version = "0.1.0"
description = "Tabsdata source and destination connector for SQLite databases."
requires-python = ">=3.12"
dependencies = [
"tabsdata>=2.1,<2.2",
]
[tool.setuptools.packages.find]
where = ["src"]
[project.entry-points."tabsdatak.connectors"]
src_sqlite = "tabsdata_sqlite._plugin:SQLITE_SRC"
dest_sqlite = "tabsdata_sqlite._plugin:SQLITE_DEST"
error.py
SqliteRuntimeException extends ConnException for failed reads and writes. SqliteErrorCode then associates each error with an exception type and message.
Codes SQLITE_1 through SQLITE_4 validate the configuration classes in __init__.py. The plugins use SQLITE_5 for query failures, SQLITE_6 for a table count mismatch, and SQLITE_7 for write failures. Message placeholders such as {table} receive their values through .exception().
from tabsdatak.conn.common.error import ConnException, ConnInitException
from tabsdatak.error import ErrorCode, ErrorDef
class SqliteRuntimeException(ConnException):
"""A SQLite read or write failed."""
class SqliteErrorCode(ErrorCode):
SQLITE_1 = ErrorDef(ConnInitException, "SqliteSrcConn validation failed")
SQLITE_2 = ErrorDef(ConnInitException, "SqliteDestConn validation failed")
SQLITE_3 = ErrorDef(ConnInitException, "SqliteSrc validation failed")
SQLITE_4 = ErrorDef(ConnInitException, "SqliteDest validation failed")
SQLITE_5 = ErrorDef(SqliteRuntimeException, "SQLite query {index} failed")
SQLITE_6 = ErrorDef(
SqliteRuntimeException,
"SQLite received {slots} table(s) for {tables} target table(s)",
)
SQLITE_7 = ErrorDef(SqliteRuntimeException, "SQLite write to {table} failed")
__init__.py
__init__.py defines the classes users import from tabsdata_sqlite. StrOrSecretSpec accepts a string or secret, and resolve reads its value. TableName limits table names to letters, numbers, and underscores, starting with a letter or underscore.
SqliteSrcConn and SqliteDestConn define the database path supplied by the Collection's connection document. Both extend Conn and require a path field, which becomes spec.path in the document.
SqliteSrc defines the queries supplied by a Publisher. SqliteDest defines the destination tables and write mode supplied by a Subscriber. Query results and returned data follow the order of queries and tables, respectively. The write mode accepts append or replace and defaults to replace.
The @_api.dataclass decorator validates these settings using the specified error code. Its kw_only=True argument requires named arguments, such as SqliteSrcConn(path="/data/shop.db").
from typing import Annotated, Literal
from pydantic import StringConstraints
import tabsdatak._api as _api
from tabsdatak.api import Dest, Secret, Src, StrOrSecretSpec
from tabsdatak.spi import Conn
from tabsdata_sqlite.error import SqliteErrorCode
TableName = Annotated[str, StringConstraints(pattern=r"^[A-Za-z_][A-Za-z0-9_]*$")]
def resolve(spec: StrOrSecretSpec) -> str:
return spec.value() if isinstance(spec, Secret) else spec
@_api.dataclass(SqliteErrorCode.SQLITE_1, kw_only=True)
class SqliteSrcConn(Conn):
"""Connection to the SQLite database a publisher reads."""
path: StrOrSecretSpec
@_api.dataclass(SqliteErrorCode.SQLITE_2, kw_only=True)
class SqliteDestConn(Conn):
"""Connection to the SQLite database a subscriber writes."""
path: StrOrSecretSpec
@_api.dataclass(SqliteErrorCode.SQLITE_3, kw_only=True)
class SqliteSrc(Src):
"""Queries a publisher runs against SQLite."""
queries: list[str]
@_api.dataclass(SqliteErrorCode.SQLITE_4, kw_only=True)
class SqliteDest(Dest):
"""SQLite tables that receive a subscriber's returned data."""
tables: list[TableName]
if_table_exists: Literal["append", "replace"] = "replace"
The connection supplies the database path, while the Publisher or Subscriber supplies what to read or write. Tabsdata passes both objects to the plugin during execution.
class SqliteSrcConn(Conn):
path: StrOrSecretSpec
kind: connectionDef
apiVersion: '1.0'
type: tabsdata_sqlite:SqliteSrcConn
spec:
path: str:/data/shop.db
class SqliteSrc(Src):
queries: list[str]
source=SqliteSrc(
queries=[
"SELECT * FROM customers",
"SELECT * FROM orders",
],
)
conn: SqliteSrcConnsrc: SqliteSrcconn.path # "/data/shop.db"
src.queries # ["SELECT * FROM customers", "SELECT * FROM orders"]
The path field belongs to SqliteSrcConn. Its value is set in the collection's connection document and passed to the plugin as conn.path.
_plugin.py
The plugins receive the connection and function settings from __init__.py. The shared connect helper resolves the database path and opens SQLite.
SqliteSrcPlugin.read_in runs the Publisher's queries and writes each result to a Parquet file in ctx.work_dir. It returns the files in query order, with an inner list for each result. If a query fails, it raises SQLITE_5 with the query index and original exception.
SqliteDestPlugin.write_out matches the Subscriber's returned files to dest.tables and skips any None file reference. It reads the remaining files and either replaces the destination tables or appends rows, according to dest.if_table_exists. The explicit BEGIN includes table creation and replacement in the transaction, so failed writes roll back those changes too.
The database path is resolved on the machine running the Function. The SQLite file must be accessible there, including when Functions run on separate workers.
SQLITE_SRC and SQLITE_DEST register the connection classes, function settings, and plugins together. Their names must match the entry points in pyproject.toml.
import sqlite3
from pathlib import Path
import polars as pl
from tabsdatak.spi import (
DestContext,
DestDef,
DestPlugin,
SrcDef,
SrcPlugin,
SrcPluginCtx,
TableFileInput,
TableFileSpec,
TableMode,
)
from tabsdata_sqlite import (
SqliteDest,
SqliteDestConn,
SqliteSrc,
SqliteSrcConn,
resolve,
)
from tabsdata_sqlite.error import SqliteErrorCode
def connect(conn: SqliteSrcConn | SqliteDestConn) -> sqlite3.Connection:
return sqlite3.connect(resolve(conn.path))
class SqliteSrcPlugin(SrcPlugin[SqliteSrcConn, SqliteSrc]):
def read_in(
self,
ctx: SrcPluginCtx,
conn: SqliteSrcConn,
src: SqliteSrc,
) -> list[list[TableFileInput]]:
con = connect(conn)
try:
slots = []
for index, query in enumerate(src.queries):
try:
frame = pl.read_database(query, con)
except Exception as e:
raise SqliteErrorCode.SQLITE_5.exception(cause=e, index=index)
parquet = Path(ctx.work_dir) / f"{index}.parquet"
frame.write_parquet(parquet)
slots.append([TableFileInput(file=parquet)])
return slots
finally:
con.close()
class SqliteDestPlugin(DestPlugin[SqliteDestConn, SqliteDest]):
def write_out(
self,
ctx: DestContext,
conn: SqliteDestConn,
dest: SqliteDest,
tables: list[TableFileSpec],
) -> None:
if len(tables) != len(dest.tables):
raise SqliteErrorCode.SQLITE_6.exception(
slots=len(tables), tables=len(dest.tables)
)
con = connect(conn)
try:
with con:
con.execute("BEGIN")
for table, parquet in zip(dest.tables, tables):
if parquet is None:
continue
try:
frame = pl.read_parquet(parquet)
columns = ", ".join(
'"' + c.replace('"', '""') + '"' for c in frame.columns
)
if dest.if_table_exists == "replace":
con.execute(f'DROP TABLE IF EXISTS "{table}"')
con.execute(
f'CREATE TABLE IF NOT EXISTS "{table}" ({columns})'
)
marks = ", ".join("?" for _ in frame.columns)
con.executemany(
f'INSERT INTO "{table}" ({columns}) VALUES ({marks})',
frame.iter_rows(),
)
except Exception as e:
raise SqliteErrorCode.SQLITE_7.exception(cause=e, table=table)
finally:
con.close()
SQLITE_SRC = SrcDef(
conn=SqliteSrcConn,
type_="sqlite-sql-in",
system="SQLite",
src_version="v1",
src=SqliteSrc,
table_mode=TableMode.SINGLE,
cardinality=lambda src: len(src.queries),
plugin=SqliteSrcPlugin,
explorer=None,
icon=None,
)
SQLITE_DEST = DestDef(
conn=SqliteDestConn,
type_="sqlite-sql-out",
system="SQLite",
dest_version="v1",
dest=SqliteDest,
cardinality=lambda dest: len(dest.tables),
plugin=SqliteDestPlugin,
explorer=None,
icon=None,
)
type_ is the registration identifier and must be unique across installed connectors. It differs from the connection document's type, which names the Python connection class.
cardinality gives the number of query results or destination tables. TableMode.SINGLE means each source result contains a single file. This connector has no explorer or icon, so both are set to None.
Installing a connector
The connector package must be available where tdk runs to validate connections and register functions. It must also be installed in the server's function environment to execute those functions.
From the directory containing sqlite_connector, run:
pip install ./sqlite_connector
Make the package available to the server, for example through a Git repository. Replace <org> and <repo> with the repository that contains sqlite_connector, then add the package to a requirements file:
tabsdata-conn-sqlite @ git+https://github.com/<org>/<repo>.git#subdirectory=sqlite_connector
Update the function environment and restart the instance. The environment update stops the instance.
tdkserver venv update --name fn --requirements connectors.txt
tdkserver start
The requirements file replaces the environment's current list of complementary packages. Include every complementary package the server still needs.
Using the connector
Create connection documents for the source and destination. The type field selects a connection class from the connection classes, and spec.path supplies the required path field. The str: prefix represents a literal string in the connection document.
kind: connectionDef
apiVersion: '1.0'
type: tabsdata_sqlite:SqliteSrcConn
spec:
path: str:/data/shop.db
kind: connectionDef
apiVersion: '1.0'
type: tabsdata_sqlite:SqliteDestConn
spec:
path: str:/data/warehouse.db
The source database must already contain the customers and orders tables used by the publisher. The destination database is created on first use if it does not exist; its parent directory must exist and be writable.
Attach the source connection to shop in the sources group, and the destination connection to warehouse in the destinations group:
tdk collection create --name shop --group sources --conn-file conn-sqlite-in.yaml
tdk collection create --name warehouse --group destinations --conn-file conn-sqlite-out.yaml
Create pub_shop.py and set queries, the field defined on SqliteSrc in the function settings. The source plugin reads the query results before pub_shop runs. The function returns those results into the customers and orders Tabsdata tables.
from tabsdatak.api import publisher, TableFrameSpec
from tabsdata_sqlite import SqliteSrc
@publisher(
source=SqliteSrc(queries=["SELECT * FROM customers", "SELECT * FROM orders"]),
output_tables=["customers", "orders"],
)
def pub_shop(
customers: TableFrameSpec, orders: TableFrameSpec
) -> tuple[TableFrameSpec, TableFrameSpec]:
return customers, orders
Register the publisher in shop. Its connection comes from the shop collection created in step 6a.
tdk fn register --coll shop --path pub_shop.py::pub_shop
Create sub_warehouse.py and set tables and if_table_exists, the fields defined on SqliteDest. The subscriber reads shop/customers and shop/orders, then returns the data to the destination plugin.
Using "replace" drops and recreates the named SQLite tables on each write. Use "append" to keep existing rows and insert the returned rows.
from tabsdatak.api import subscriber, TableFrameSpec
from tabsdata_sqlite import SqliteDest
@subscriber(
destination=SqliteDest(tables=["customers", "orders"], if_table_exists="replace"),
input_tables=["shop/customers", "shop/orders"],
)
def sub_warehouse(
customers: TableFrameSpec, orders: TableFrameSpec
) -> tuple[TableFrameSpec, TableFrameSpec]:
return customers, orders
Register the subscriber in warehouse so it uses that collection's destination connection.
tdk fn register --coll warehouse --path sub_warehouse.py::sub_warehouse