> **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. # Loaders Reference The loaders' classes and types are in the `amp.loaders` package. You use most of them through `client.sql(…).load()`, which [Loaders](/sdks/python/loaders) explains, and the `DataLoader` base class to [write your own loader](/sdks/python/loaders/custom). ```python from amp.loaders import ( DataLoader, LabelJoinConfig, LoadConfig, LoaderRegistry, LoadMode, LoadResult, create_loader, get_available_loaders, get_loader_class, ) ``` ## `LoadResult` What a load returns: one for a whole load with `read_all=True`, or one per batch, per partition of a [parallel load](/sdks/python/parallel-streaming), or per reorg of a [stream](/sdks/python/streaming). | Field | Type | Description | | --------------------- | ----------------------------------------------------------------------------- | ----------------------------------------------------------------------- | | `success` | `bool` | Whether it loaded. A failure doesn't raise an exception: check this. | | `error` | `str` or `None` | Why it failed. | | `rows_loaded` | `int` | How many rows it loaded. | | `duration` | `float` | How long it took, in seconds. | | `ops_per_second` | `float` | `rows_loaded` divided by `duration`. | | `table_name` | `str` | The destination. | | `loader_type` | `str` | The loader's name, like `'deltalake'`, or `'parallel'` for a partition. | | `metadata` | `dict` | Details, which depend on the load (below). | | `is_reorg` | `bool` | Whether it reports a reorg rather than a load. | | `invalidation_ranges` | list of [`BlockRange`](/sdks/python/reference/streaming#blockrange) or `None` | For a reorg, the block ranges whose rows the loader deleted. | `str(result)` is a line like `✅ Loaded 100 rows to blocks in 0.02s`, `❌ Failed to load to blocks: …`, or `🔄 Reorg detected: 1 ranges invalidated`. `metadata` has these keys, among others: | Load | Keys | | ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `read_all=True` | `operation` (`'load_table'`), `batches_processed`, `rows_loaded`, `columns`, `table_size_bytes`, `throughput_rows_per_sec` | | A batch | `operation` (`'load_batch'`), `batch_size` (its rows), `schema_fields`, `throughput_rows_per_sec`, and what the loader adds | | A batch of a stream | `is_streaming`, `batch_count`, `block_ranges` (each `network`, `start`, and `end`), `ranges_complete`, and `operation`: `'skip_duplicate'` for a range already loaded | | A partition | `partition_id`, `batches_processed`, `partition_metadata` (`start_block`, `end_block`, `block_column`, `partition_size`), `worker_thread` | | A reorg | `operation` (`'reorg'`), `invalidation_count`, `reorg_number` | ## `LoadMode` An enum of how loaded rows join a table's existing rows: `LoadMode.APPEND` (`'append'`), `LoadMode.OVERWRITE` (`'overwrite'`), `LoadMode.UPSERT` (`'upsert'`), and `LoadMode.MERGE` (`'merge'`). `load()` takes their values, as strings; a `DataLoader`'s methods take the enum. See [Load modes](/sdks/python/loaders#load-modes), and each loader's page for the modes it supports. ## `LabelJoinConfig` ```python LabelJoinConfig(label_name: str, label_key_column: str, stream_key_column: str) ``` Which [labels](/sdks/python/labels) to add to each row: the label set's name, from `configure_label()`, the label file's column to match, and the results' column to match it with. See [Labels](/sdks/python/labels#labeljoinconfig). ## `LoadConfig` The load options, as `load()` passes them to a loader. You don't create one yourself. | Field | Type | Default | | ------------------ | ---------- | ----------------- | | `batch_size` | `int` | `10000` | | `mode` | `LoadMode` | `LoadMode.APPEND` | | `create_table` | `bool` | `True` | | `schema_evolution` | `bool` | `False` | | `max_retries` | `int` | `3` | | `retry_delay` | `float` | `1.0` | Loaders don't read `schema_evolution`, `max_retries`, or `retry_delay`: see [Load options](/sdks/python/loaders#load-options). ## `DataLoader` ```python class DataLoader(ABC, Generic[TConfig]): def __init__(self, config: dict, label_manager: LabelManager | None = None): ... ``` The base class of every loader. A loader takes its configuration as a dict, which the base class turns into the dataclass named in the class's declaration, like `class SQLiteLoader(DataLoader[SQLiteConfig])`. It leaves out the `state` and `resilience` keys, which it reads itself (see [Stream state](/sdks/python/streaming#stream-state) and [Retries](/sdks/python/loaders#retries)). An unknown key raises `ValueError: Invalid … configuration: …`. You can also use a loader without a `Client`, to load Arrow data you have. This one needs the [Delta Lake loader's](/sdks/python/loaders/delta-lake) `deltalake` package: ```python import pyarrow as pa from amp.loaders import LoadMode, create_loader table = pa.table({'symbol': ['USDT', 'USDC', 'WETH'], 'decimals': [6, 6, 18]}) with create_loader('deltalake', {'table_path': './tokens'}) as loader: result = loader.load_table(table, 'tokens', mode=LoadMode.APPEND) print(result) ``` ``` ✅ Loaded 3 rows to tokens in 0.01s ``` ### Class attributes | Attribute | Type | Default | Description | | ----------------------- | ----------------- | --------------------------------------- | ------------------------------------------------------------------ | | `SUPPORTED_MODES` | set of `LoadMode` | `{LoadMode.APPEND, LoadMode.OVERWRITE}` | The modes the loader accepts. Others fail with `Unsupported mode`. | | `REQUIRES_SCHEMA_MATCH` | `bool` | `True` | Describes the loader. The SDK doesn't read it. | | `SUPPORTS_TRANSACTIONS` | `bool` | `False` | Describes the loader. The SDK doesn't read it. | ### Attributes | Attribute | Type | Description | | ---------------------- | --------------------------------------------------------------------------- | -------------------------------------------------------------------- | | `config` | The loader's configuration dataclass | Its configuration, without `state` and `resilience`. | | `logger` | `logging.Logger` | A logger named after the class. | | `is_connected` | `bool` | Whether `connect()` succeeded and `disconnect()` hasn't run since. | | `label_manager` | `LabelManager` or `None` | The client's labels, for joins. | | `state_enabled` | `bool` | `state.enabled`. | | `state_store` | [`StreamStateStore`](/sdks/python/reference/streaming#stream-state-stores) | The batches loaded so far. In memory, unless the loader replaces it. | | `store_batch_id` | `bool` | `state.store_batch_id`: whether to add `_amp_batch_id`. | | `store_full_metadata` | `bool` | `state.store_full_metadata`: whether to add `_amp_block_ranges`. | | `retry_config` | [`RetryConfig`](/sdks/python/reference/streaming#retryconfig) | `resilience.retry`. | | `back_pressure_config` | [`BackPressureConfig`](/sdks/python/reference/streaming#backpressureconfig) | `resilience.back_pressure`. | ### Methods to implement | Method | Description | | ------------------------------------------------------ | ----------------------------------------------------------------------------------- | | `connect() -> None` | Opens the connection to the destination, and sets `self._is_connected = True`. | | `disconnect() -> None` | Closes it, and sets `self._is_connected = False`. | | `_load_batch_impl(batch, table_name, **kwargs) -> int` | Writes a `pyarrow.RecordBatch` to `table_name`, and returns how many rows it wrote. | ### Methods to override Each is optional. [Custom Loaders](/sdks/python/loaders/custom) shows them in a working loader. | Method | Called | Default | | ---------------------------------------------------------------------------------------------- | ------------------------------------------------------------------- | ----------------------------------------------- | | `_create_table_from_schema(schema, table_name) -> None` | Before the first batch, unless `create_table=False` | Not defined: the SDK logs a warning, and loads. | | `_clear_table(table_name) -> None` | For `overwrite`, before loading. Must accept a missing table. | Not defined: `overwrite` keeps existing rows. | | `_handle_reorg(invalidation_ranges, table_name, connection_name) -> None` | When a stream reports a reorg | Raises `NotImplementedError`. | | `load_batch_transactional(batch, table_name, connection_name, ranges, ranges_complete) -> int` | For each batch of a stream, instead of `load_batch()`, when defined | Not defined. | | `_get_required_config_fields() -> list[str]` | When the loader is created | `[]` | | `_validate_config() -> None` | When the loader is created, to check the configuration | Checks the required fields. | | `_get_loader_batch_metadata(batch, duration, **kwargs) -> dict` | After each batch, to add to its result's `metadata` | `{}` | | `_get_loader_table_metadata(table, duration, batch_count, **kwargs) -> dict` | After `load_table()`, to add to its result's `metadata` | `{}` | | `get_table_info(table_name) -> dict \| None` | By `table_exists()` | Raises `NotImplementedError`. | | `get_table_schema(table_name) -> pyarrow.Schema \| None` | By you | Raises `NotImplementedError`. | | `health_check() -> dict` | By you | `{'healthy': is_connected, 'loader_type': …}` | ### Methods to call | Method | Returns | Description | | --------------------------------------------------------- | ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `load_table(table, table_name, **kwargs)` | `LoadResult` | Loads a `pyarrow.Table`, one batch per chunk, split into batches of `batch_size` rows when larger. Takes the [load options](/sdks/python/loaders#load-options), with `mode` as a `LoadMode`. | | `load_batch(batch, table_name, **kwargs)` | `LoadResult` | Loads a `pyarrow.RecordBatch`, connecting first if needed, with [retries](/sdks/python/loaders#retries). Raises `RuntimeError` when it fails for good. | | `load_stream(batches, table_name, **kwargs)` | Iterator of `LoadResult` | Loads an iterator of record batches, one result per batch. | | `load_stream_continuous(responses, table_name, **kwargs)` | Iterator of `LoadResult` | Loads a stream's [`ResponseBatch`](/sdks/python/reference/streaming#responsebatch)es, handling reorgs. | | `table_exists(table_name)` | `bool` | Uses `get_table_info()`, or, without it, whether this loader created the table. | | `close()` | `None` | The same as `disconnect()`. | A loader is a context manager: `with loader:` connects, and disconnects at the end. ## `LoaderRegistry` The loaders the SDK knows, by name. It finds the built-in ones itself, and names each after its class, without `Loader`, in lowercase: `DeltaLakeLoader` is `deltalake`. | Method | Returns | Description | | ---------------------------------------------------------------- | ------------------ | -------------------------------------------------------------------------------------------------------------------- | | `LoaderRegistry.register(name, loader_class)` | `None` | Adds a loader, or replaces the one with that name. Raises `ValueError: Loader class … must inherit from DataLoader`. | | `LoaderRegistry.get_loader_class(name)` | `type[DataLoader]` | Raises `ValueError: Loader 'name' not found. Available loaders: […]`. | | `LoaderRegistry.create_loader(name, config, label_manager=None)` | `DataLoader` | Creates a loader, without connecting it. | | `LoaderRegistry.get_available_loaders()` | `list[str]` | The names it knows. PostgreSQL, Redis, LMDB, and Snowflake are left out without their packages. Others are listed. | `create_loader()`, `get_loader_class()`, and `get_available_loaders()` are also functions of `amp.loaders`, which call these.