> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # Custom Loaders When the SDK has no [loader](/sdks/python/loaders) for the system you want Amp's data in, you can write one. A loader is a Python class that receives query results as Arrow record batches, groups of rows in Arrow's columnar format, and writes them to its destination. You write how to connect and how to write a batch. The SDK does the rest: it creates the loader for each load, retries failed batches, joins [labels](/sdks/python/labels), tracks the block ranges a [stream](/sdks/python/streaming) loaded, and reports each load's `LoadResult`. This page builds a loader for [SQLite](https://www.sqlite.org), a database in a single file, using only Python's standard library and PyArrow, which the SDK installs. ## Prerequisites * **Amp running, with the Quickstart's dataset.** The examples query `"acme/eth_mainnet"`, which the [Quickstart](/get-started/quickstart) deploys, through solo mode's query server at `grpc://localhost:1602`. * **The Python SDK**, installed as [Installation](/sdks/python#installation) shows. ## How the SDK uses a loader For each load, the SDK: 1. Looks up the connection's loader by the name you registered it under, and creates an instance of it with the connection's configuration. 2. Calls `connect()`. 3. For each batch: * Before the first batch for a table, calls `_create_table_from_schema()`, if the loader has it and the load's `create_table` option is on, which it is by default. * With `mode='overwrite'`, calls `_clear_table()`, if the loader has it: once before a `read_all=True` load, but before every batch of any other load, which then keeps only its last batch (see [Load modes](/sdks/python/loaders#load-modes)). * Calls `_load_batch_impl()` with the batch, and retries it if it fails with an error that looks temporary, like a timeout. When a stream reports a [reorg](/get-started/concepts/reorgs), it calls `_handle_reorg()` instead. 4. Calls `disconnect()`, even when the load fails. A new instance serves each load, and each worker of a [parallel load](/sdks/python/parallel-streaming), so keep no state on the instance that must outlive a load. ## Write the loader ```python [sqlite_loader.py] import sqlite3 from dataclasses import dataclass from typing import Any, Dict, List import pyarrow as pa from amp.loaders import DataLoader, LoaderRegistry, LoadMode from amp.streaming import BlockRange @dataclass class SQLiteConfig: path: str timeout: float = 5.0 class SQLiteLoader(DataLoader[SQLiteConfig]): SUPPORTED_MODES = {LoadMode.APPEND, LoadMode.OVERWRITE} def __init__(self, config: Dict[str, Any], label_manager=None) -> None: super().__init__(config, label_manager=label_manager) self._db = None def connect(self) -> None: self._db = sqlite3.connect(self.config.path, timeout=self.config.timeout) self._is_connected = True def disconnect(self) -> None: if self._db: self._db.close() self._db = None self._is_connected = False def _create_table_from_schema(self, schema: pa.Schema, table_name: str) -> None: columns = [f'"{field.name}" {self._sqlite_type(field.type)}' for field in schema] if '_amp_batch_id' not in schema.names: columns.append('"_amp_batch_id" TEXT') self._db.execute(f'CREATE TABLE IF NOT EXISTS "{table_name}" ({", ".join(columns)})') def _clear_table(self, table_name: str) -> None: exists = self._db.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (table_name,)) if exists.fetchone(): self._db.execute(f'DELETE FROM "{table_name}"') self._db.commit() def _load_batch_impl(self, batch: pa.RecordBatch, table_name: str, **kwargs) -> int: names = ', '.join(f'"{name}"' for name in batch.schema.names) placeholders = ', '.join('?' for _ in batch.schema.names) columns = [[self._sqlite_value(value) for value in column.to_pylist()] for column in batch.columns] self._db.executemany(f'INSERT INTO "{table_name}" ({names}) VALUES ({placeholders})', zip(*columns)) self._db.commit() return batch.num_rows def _handle_reorg(self, invalidation_ranges: List[BlockRange], table_name: str, connection_name: str) -> None: for block_range in invalidation_ranges: batches = self.state_store.invalidate_from_block( connection_name, table_name, block_range.network, block_range.start ) for batch_id in batches: self._db.execute( f'DELETE FROM "{table_name}" WHERE "_amp_batch_id" LIKE ?', (f'%{batch_id.unique_id}%',) ) self._db.commit() @staticmethod def _sqlite_type(arrow_type: pa.DataType) -> str: if pa.types.is_integer(arrow_type): return 'INTEGER' if pa.types.is_floating(arrow_type): return 'REAL' if pa.types.is_binary(arrow_type) or pa.types.is_fixed_size_binary(arrow_type): return 'BLOB' return 'TEXT' @staticmethod def _sqlite_value(value: Any) -> Any: if value is None or isinstance(value, (int, float, str, bytes)): return value if hasattr(value, 'isoformat'): return value.isoformat() return str(value) LoaderRegistry.register('sqlite', SQLiteLoader) ``` ### The configuration A loader's configuration is a [dataclass](https://docs.python.org/3/library/dataclasses.html): its fields are the options, with their types and defaults. Name it in the class's base, `DataLoader[SQLiteConfig]`, and the SDK builds it from the connection's dictionary, available as `self.config`: * A missing required field, or a key the dataclass doesn't have, fails the load with `Invalid SQLiteLoader configuration:` and the reason. * Two keys are the SDK's, not the loader's, and never reach the dataclass: `state`, where the loader records the block ranges it loaded, and `resilience`, how it retries. [Loaders](/sdks/python/loaders) documents both. * Validate further in the dataclass's `__post_init__()`, raising `ValueError` with a message that says what to fix. ### Required methods | Method | What it does | | ----------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `__init__(config, label_manager=None)` | Only if you need it, to set attributes. Call `super().__init__(config, label_manager=label_manager)` first: it parses the configuration. | | `connect()` | Opens the connection, then sets `self._is_connected = True`. The SDK calls it again if it's `False` when a batch arrives. | | `disconnect()` | Closes the connection, then sets `self._is_connected = False`. | | `_load_batch_impl(batch, table_name, **kwargs)` | Writes one `pyarrow.RecordBatch` to `table_name`, and returns how many rows it wrote. Raise an exception when it fails: the SDK reports it. `kwargs` holds the load's options, like `mode` and `create_table`, and any loader-specific ones passed to `load()`, so accept them even if you don't use them. | ### Optional methods and attributes | Method or attribute | What it does | | --------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `SUPPORTED_MODES` | The [load modes](/sdks/python/loaders) the loader accepts, a set of `LoadMode` values. The default is `{LoadMode.APPEND, LoadMode.OVERWRITE}`. A load in any other mode fails with `Unsupported mode`. | | `_create_table_from_schema(schema, table_name)` | Creates `table_name` from a `pyarrow.Schema`, if it doesn't exist. Make it safe to call on an existing table: every load calls it. Always add an `_amp_batch_id` text column (see [Reorgs](/sdks/python/loaders/custom#reorgs)). Without this method, the SDK logs a warning, and you must create tables yourself. | | `_clear_table(table_name)` | Deletes the table's rows, for `mode='overwrite'`. A `read_all=True` load calls it before creating the table, so do nothing when the table doesn't exist. | | `_handle_reorg(invalidation_ranges, table_name, connection_name)` | Deletes the rows of the blocks a reorg replaced (see [Reorgs](/sdks/python/loaders/custom#reorgs)). Without it, a stream fails at its first reorg. | | `load_batch_transactional(batch, table_name, connection_name, ranges, ranges_complete)` | If defined, streams call it instead of `_load_batch_impl()`, to check, write, and record a batch in one transaction, as the [PostgreSQL loader](/sdks/python/loaders/postgresql) does. It skips table creation, retries, and label joins, so it must handle them itself. Most loaders don't need it. | | `get_table_info(table_name)` | Returns a dictionary describing the table, or `None` if it doesn't exist. `table_exists()` uses it. | | `get_table_schema(table_name)` | Returns the table's columns as a `pyarrow.Schema`. | | `health_check()` | Returns a dictionary saying whether the destination is reachable. The default reports whether the loader is connected. | | `_get_loader_batch_metadata(batch, duration, **kwargs)` | Returns a dictionary to add to the `metadata` of each batch's `LoadResult`. | | `_get_loader_table_metadata(table, duration, batch_count, **kwargs)` | Returns a dictionary to add to the `metadata` of a `read_all=True` load's `LoadResult`. | The base class gives the loader these attributes: | Attribute | What it holds | | --------------------- | ------------------------------------------------------------------------------------------ | | `self.config` | The configuration dataclass. | | `self.logger` | A `logging.Logger` named after the class. | | `self.state_store` | The stream state: which block ranges the loader has loaded, for each connection and table. | | `self.store_batch_id` | Whether rows get an `_amp_batch_id`, from the `state` option. `True` by default. | | `self.label_manager` | The [labels](/sdks/python/labels) you configured on the client. | `REQUIRES_SCHEMA_MATCH` and `SUPPORTS_TRANSACTIONS`, which the built-in loaders set, describe the loader, but the SDK doesn't read them. ## Register and use it `LoaderRegistry.register()` gives the loader a name. Run it before the client loads anything, here by importing `sqlite_loader.py`. Then configure a connection that uses the name, and load into it: ```python [load_sqlite.py] import sqlite3 import sqlite_loader # noqa: F401 (registers the 'sqlite' loader) from amp import Client client = Client('grpc://localhost:1602') client.configure_connection('local_sqlite', 'sqlite', {'path': 'amp.db'}) result = client.sql('SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks').load( connection='local_sqlite', destination='blocks', read_all=True ) print(result) db = sqlite3.connect('amp.db') for row in db.execute('SELECT block_num, timestamp, hex(hash), gas_used FROM blocks ORDER BY block_num LIMIT 3'): print(row) ``` :::terminal ```bash python load_sqlite.py ``` ``` ✅ Loaded 100 rows to blocks in 0.00s (24000000, '2025-12-13T00:06:11+00:00', '738CC1716EA1F08ADAC2D4E2230AEDCEE2D5CD3F65D66D5D3597E05D710A3D50', 57860235) (24000001, '2025-12-13T00:06:23+00:00', '1353CF08DAA372D4F13D8A95721DBD406650C78A602C0BFA1867042275637C3C', 27660983) (24000002, '2025-12-13T00:06:35+00:00', 'B184E0E5844FE75B8CBF74B05A5AFA1662C2B67BFF13C597F2D56F8E40C19F9A', 17602635) ``` ::: Once registered, the loader appears in `client.get_available_loaders()`, after the built-in loaders. Its results' `loader_type` comes from the class's name, without `Loader`, in lowercase: `sqlite`. `configure_connection()` saves the connection in `~/.flight_sql_connections.json`, so later scripts can load into `local_sqlite` without configuring it again. They still have to register the loader first. ## Stream into it (Unstable) With `stream=True`, the loader keeps receiving batches as Amp extracts new blocks, until the stream ends. Each row gets an `_amp_batch_id`: ```python [stream_sqlite.py] import sqlite3 import sqlite_loader # noqa: F401 (registers the 'sqlite' loader) from amp import Client client = Client('grpc://localhost:1602') client.configure_connection('local_sqlite', 'sqlite', {'path': 'amp.db'}) results = client.sql('SELECT block_num, hash, gas_used FROM "acme/eth_mainnet".blocks').load( connection='local_sqlite', destination='streamed_blocks', stream=True, with_reorg_detection=False ) for result in results: print(result) db = sqlite3.connect('amp.db') print(db.execute('SELECT count(*), min(block_num), max(block_num), _amp_batch_id FROM streamed_blocks').fetchone()) ``` :::terminal ```bash python stream_sqlite.py ``` ``` Failed to load checkpoint, starting from beginning: 'SQLiteLoader' object has no attribute 'checkpoint_store' ✅ Loaded 100 rows to streamed_blocks in 0.00s ✅ Loaded 0 rows to streamed_blocks in 0.00s (100, 24000000, 24000099, '7dce8312375e3370') ``` ::: Amp sends each block range as a batch of rows, then an empty batch that marks the range complete, hence the second result. The stream then ends 30 seconds after it started, as every stream does, even on a dataset that follows its chain (see [Streams end after 30 seconds](/sdks/python/streaming#streams-end-after-30-seconds)). `with_reorg_detection=False` and the first line come from issues in the SDK's streaming, which every loader shares: see [Known issues](/sdks/python/streaming#known-issues). ## Reorgs When a [reorg](/get-started/concepts/reorgs) replaces blocks, a loader must remove the rows it loaded from the old ones. The SDK gives it what it needs: 1. Before a streamed batch reaches `_load_batch_impl()`, the SDK adds an `_amp_batch_id` column. Its value identifies the block range the rows came from: 16 hexadecimal characters, derived from the network, the range's first and last blocks, the last block's hash, and the hash of the block before the first. A batch that spans several networks gets several, separated by `|`. Store the column with the rows. 2. Once a range is complete, the SDK records its identifier in `self.state_store`. 3. When a stream reports a reorg, the SDK calls `_handle_reorg()` with the replaced ranges, as `BlockRange` objects with a `network`, a `start`, and an `end`. For each, `self.state_store.invalidate_from_block()` returns the identifiers of the batches that end at or after that block, and forgets them. Delete the rows whose `_amp_batch_id` contains one, as the example does with `LIKE`. Create the `_amp_batch_id` column even when the schema doesn't have it, as `_create_table_from_schema()` does above. A parallel load creates the table from a sample of the query's results, before its workers add the column to every row. You can test `_handle_reorg()` without a reorg, by giving the loader a stream you build yourself, with `ResponseBatch.data_batch()` and `ResponseBatch.reorg_batch()`: ```python [test_reorg.py] import sqlite3 import pyarrow as pa from amp.streaming import BatchMetadata, BlockRange, ResponseBatch from sqlite_loader import SQLiteLoader def blocks(start, end, block_hash): data = pa.record_batch({'block_num': pa.array(range(start, end + 1), type=pa.uint64())}) metadata = BatchMetadata( ranges=[BlockRange(network='mainnet', start=start, end=end, hash=block_hash)], ranges_complete=True ) return ResponseBatch.data_batch(data, metadata) stream = [ blocks(100, 101, '0xaaa'), blocks(102, 103, '0xbbb'), ResponseBatch.reorg_batch([BlockRange(network='mainnet', start=102, end=103)]), blocks(102, 103, '0xccc'), ] with SQLiteLoader({'path': 'reorg_test.db'}) as loader: for result in loader.load_stream_continuous(iter(stream), 'blocks', connection_name='test'): print(result) db = sqlite3.connect('reorg_test.db') print(db.execute('SELECT block_num, _amp_batch_id FROM blocks ORDER BY block_num').fetchall()) ``` :::terminal ```bash python test_reorg.py ``` ``` ✅ Loaded 2 rows to blocks in 0.00s ✅ Loaded 2 rows to blocks in 0.00s Reorg detected on mainnet: blocks 102-103 invalidated 🔄 Reorg detected: 1 ranges invalidated ✅ Loaded 2 rows to blocks in 0.00s [(100, 'c1406207f76f4761'), (101, 'c1406207f76f4761'), (102, '1db7e0f139af20da'), (103, '1db7e0f139af20da')] ``` ::: Blocks 102 and 103 appear once, with the identifier of their new batch: `_handle_reorg()` deleted the rows of the old one. ## Troubleshooting ### `Invalid SQLiteLoader configuration: 'typing.TypeVar' object is not callable` The class doesn't name its configuration dataclass. Declare it as `class SQLiteLoader(DataLoader[SQLiteConfig])`. ### `Invalid SQLiteLoader configuration: … got an unexpected keyword argument 'paht'` The connection's configuration has a key the dataclass doesn't. Fix its spelling, or add the field. ### `Loader 'sqlit' not found. Available loaders: […]` No loader is registered under that name. Check the name in `configure_connection()`, and that the script registers the loader, by importing its module, before it loads. ### `Loader class must inherit from DataLoader` `LoaderRegistry.register()` only takes subclasses of `DataLoader`. ### `Unsupported mode LoadMode.UPSERT. Supported modes: {…}` The load asked for a mode that isn't in `SUPPORTED_MODES`. Use one that is, or support it and add it. ### `table … has no column named _amp_batch_id` The table was created without the `_amp_batch_id` column, then a stream or a parallel load added it to the rows. Add the column in `_create_table_from_schema()`, and in the existing table. ### `no such table` with `mode='overwrite'` `_clear_table()` ran before the table existed. Make it do nothing when the table doesn't exist. ### `SQLiteLoader does not implement _handle_reorg(). Streaming with reorg detection requires implementing this method.` A stream reported a reorg, and the loader can't handle it. Implement `_handle_reorg()`.