Data storage

This section explains how class instances of the Data model living in Python memory can be stored.

The storage layer is organized around a set of protocols that define contracts for reading and writing data model entities. This protocol-based design allows swapping the underlying storage implementation without changing the fetcher code.

StorageAdapter interface

The StorageAdapter protocol is the top-level entry point. It exposes two methods that return the reader and writer chains:

Each level in the chain manages a specific scope:

This chain allows reading or writing any entity of the data model, from provider metadata down to individual series observations.

Reader chain

        flowchart TD
    A[StorageAdapter] -->|"get_reader()"| B[StorageReader]
    B -->|"get_provider_reader(provider_code)"| C[ProviderReader]
    C -->|"get_dataset_reader(dataset_code)"| D[DatasetReader]
    C -->|"iter_dataset_readers()"| D
    D -->|"read_series(series_code)"| F[Series]
    D -->|"iter_series()"| F
    D -->|"iter_series_metadata()"| G[SeriesMetadata]

StorageReader provides access to multiple providers:

ProviderReader provides access to data of a single provider:

DatasetReader provides access to data of a single dataset:

Writer chain

        flowchart TD
    A[StorageAdapter] -->|"get_writer()"| B[StorageWriter]
    B -->|"get_provider_writer(provider_code)"| C[ProviderWriter]
    C -->|"get_dataset_writer(dataset_code)"| D[DatasetWriter]
    C -->|"create_session(session_id)"| S[ProviderWriterSession]
    C -->|"write_provider_metadata()"| M[ProviderMetadata]
    C -->|"write_category_tree()"| T[CategoryTree]
    D -->|"write_dataset_metadata()"| DM[DatasetMetadata]
    D -->|"write_series(series_iter)"| SI[SeriesIterator]
    S -->|"commit()"| OK[✓]

StorageWriter manages providers at the storage level:

ProviderWriter writes data for a single provider:

DatasetWriter writes data for a single dataset:

Decoupling from the adapter

The convert part of fetchers is designed to use the StorageAdapter protocol without knowing which concrete adapter is in use, so that the fetcher is not coupled to any specific storage backend.

To achieve this, instead of instantiating any concrete adapter directly, the converter process will read the storage URI from the STORAGE_URI environment variable at runtime, and the corresponding adapter will be instantiated and used. Since any adapter implements the same set of protocols, the source code of the fetcher will remain valid.

Provider Writer Sessions

ProviderWriterSession allows applying write operations to a provider atomically. It acts as a context manager: all changes are written to a temporary session directory, and only committed to the target storage when ProviderWriterSession.commit is called. If an exception occurs or commit is not called, all changes are rolled back.

This ensures that a reader never sees a partially-written state.

In practice, the BaseConverter class automatically creates and manages sessions — the fetcher’s converter only needs to implement _start_dataset_converter() and call the write_* methods on the DatasetWriter it receives:

from dbnomics_toolbox.fetcher_utils import BaseConverter


class MyConverter(BaseConverter):
    def _start_dataset_converter(self, dataset_converter, *, dataset_writer):
        # BaseConverter has already created the session and obtained the DatasetWriter.
        # Just write your data — commit/rollback is handled automatically.
        dataset_writer.write_dataset_metadata(metadata)
        dataset_writer.write_series(series_iter)

If you need to write data manually, you can still use sessions directly:

provider_writer = storage_adapter.get_writer().get_provider_writer(parse_provider_code("INSEE"))

with provider_writer.create_session("2025-01-01T12:00:00") as session:
    session_writer = session.provider_writer
    session_writer.write_provider_metadata(provider_metadata)
    session_writer.write_category_tree(category_tree)

    for dataset_code in dataset_codes:
        dataset_writer = session_writer.get_dataset_writer(dataset_code)
        dataset_writer.write_dataset_metadata(dataset_metadata)
        dataset_writer.write_series(all_series)

    session.commit()

Provider Snapshots

The storage layer tracks the history of changes to provider data through ProviderSnapshot objects.

A snapshot represents the state of a provider at a point in time, identified by a ProviderSnapshotId and containing metadata such as the creation date, author, and commit message.

Snapshots allow reading data as it existed at a specific point in time:

from dbnomics_toolbox.model.identifiers import parse_provider_snapshot_id, parse_dataset_code

provider_reader = storage_adapter.get_reader().get_provider_reader(parse_provider_code("INSEE"))

# Iterate over all snapshots (newest first)
for snapshot in provider_reader.iter_provider_snapshots():
    print(f"Snapshot {snapshot.id} at {snapshot.created_at}: {snapshot.message}")

# Read data at a specific snapshot
snapshot_reader = provider_reader.at_snapshot(parse_provider_snapshot_id("abc123"))
snapshot_metadata = snapshot_reader.read_provider_metadata()

# Iterate over snapshots touching a specific dataset
dataset_reader = provider_reader.get_dataset_reader(parse_dataset_code("IPC-2015"))
for snapshot in dataset_reader.iter_provider_snapshots():
    print(f"Dataset IPC-2015 changed in snapshot {snapshot.id}")

Examples

The following example creates a storage adapter and demonstrates several common use cases.

from dbnomics_toolbox.model import DatasetMetadata, Observation, ProviderMetadata, Series
from dbnomics_toolbox.model.identifiers import parse_provider_code, parse_dataset_code
from dbnomics_toolbox.storage.adapters.factories import create_storage_adapter_from_uri
from dbnomics_toolbox.storage.uris import Uri

# Instantiate a storage adapter from a storage URI.
storage_adapter = create_storage_adapter_from_uri(Uri.parse("filesystem:///converted-data"))

# --- Use case 1: read / write provider metadata ---
provider_reader = storage_adapter.get_reader().get_provider_reader(parse_provider_code("INSEE"))
provider_metadata = provider_reader.read_provider_metadata()

provider_writer = storage_adapter.get_writer().get_provider_writer(parse_provider_code("INSEE"))
provider_writer.write_provider_metadata(
    ProviderMetadata.create(
        code="INSEE",
        name="Institut national de la statistique et des études économiques",
        region="FR",
        terms_of_use="https://www.insee.fr/fr/information/2381863",
        website="https://www.insee.fr/",
    )
)

# --- Use case 2: iterate over datasets and their series ---
for dataset_reader in provider_reader.iter_dataset_readers():
    dataset_metadata = dataset_reader.read_dataset_metadata()
    print(f"Dataset: {dataset_metadata.code} ({dataset_metadata.name})")

    for series in dataset_reader.iter_series():
        print(f"  Series: {series.metadata.code} ({len(series.observations)} observations)")

# --- Use case 3: write a dataset with its series ---
dataset_writer = (
    storage_adapter
    .get_writer()
    .get_provider_writer(parse_provider_code("INSEE"))
    .get_dataset_writer(parse_dataset_code("IPC-2015"))
)

dataset_writer.write_dataset_metadata(
    DatasetMetadata.create(
        code="IPC-2015",
        name="Indice des prix à la consommation - Base 2015",
    )
)

dataset_writer.write_series([
    Series.create(
        code="A.IPC.INDICE.ENSEMBLE",
        dimensions={"frequency": "A"},
        observations=[
            Observation.create(period="2000", value="19"),
            Observation.create(period="2001", value="22"),
        ],
    ),
])

Note: here the storage URI is parsed manually for demonstration purpose, but in the converter part of the fetcher, the storage instantiation is done automatically.

Storage adapters

The adapters are concrete implementations of the storage protocols.

File-system adapter

As of now, only the FsStorageAdapter adapter is available, backed by the local file-system.

It is documented in the File-system storage adapter page.