# Data storage

```{toctree}
:hidden:
:maxdepth: 1
file-system-adapter
```

This section explains how class instances of the [](../data-model.md) 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:

- [](#StorageAdapter.get_reader) → [](#StorageReader): read entry point
- [](#StorageAdapter.get_writer) → [](#StorageWriter): write entry point

Each level in the chain manages a specific scope:

- **Storage** level ([StorageReader](#StorageReader) / [StorageWriter](#StorageWriter)): can manage multiple providers
- **Provider** level ([ProviderReader](#ProviderReader) / [ProviderWriter](#ProviderWriter)): data of a single provider
- **Dataset** level ([DatasetReader](#DatasetReader) / [DatasetWriter](#DatasetWriter)): data of a single dataset

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

### Reader chain

```{mermaid}
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](#StorageReader) provides access to multiple providers:

- [](#StorageReader.get_provider_reader) — get a reader for a specific provider
- [](#StorageReader.iter_provider_codes) — iterate over all provider codes
- [](#StorageReader.iter_provider_readers) — iterate over all provider readers

[ProviderReader](#ProviderReader) provides access to data of a single provider:

- [](#ProviderReader.read_provider_metadata) — read provider metadata
- [](#ProviderReader.read_category_tree) — read the category tree
- [](#ProviderReader.read_dataset_family) — read a dataset family
- [](#ProviderReader.get_dataset_reader) — get a reader for a specific dataset
- [](#ProviderReader.iter_dataset_codes) — iterate over dataset codes
- [](#ProviderReader.iter_dataset_readers) — iterate over dataset readers
- [](#ProviderReader.iter_dataset_families) — iterate over dataset families
- [](#ProviderReader.iter_provider_snapshots) — iterate over provider snapshots
- [](#ProviderReader.at_snapshot) — get a reader fixed at a specific snapshot
- [](#ProviderReader.has_dataset) / [](#ProviderReader.has_category_tree) — existence checks
- [](#ProviderReader.count_datasets) / [](#ProviderReader.compute_dataset_hash) — utilities

[DatasetReader](#DatasetReader) provides access to data of a single dataset:

- [](#DatasetReader.read_dataset_metadata) — read dataset metadata
- [](#DatasetReader.read_series) — read a single series (metadata + observations)
- [](#DatasetReader.iter_series) — iterate over all series
- [](#DatasetReader.iter_series_metadata) — iterate over series metadata only
- [](#DatasetReader.iter_series_codes) — iterate over series codes
- [](#DatasetReader.iter_provider_snapshots) — iterate over snapshots touching this dataset

### Writer chain

```{mermaid}
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](#StorageWriter) manages providers at the storage level:

- [](#StorageWriter.get_provider_writer) — get a writer for a specific provider
- [](#StorageWriter.delete_provider) — delete an entire provider

[ProviderWriter](#ProviderWriter) writes data for a single provider:

- [](#ProviderWriter.write_provider_metadata) — write provider metadata
- [](#ProviderWriter.write_category_tree) — write the category tree
- [](#ProviderWriter.get_dataset_writer) — get a writer for a specific dataset
- [](#ProviderWriter.create_session) — create an atomic write session
- [](#ProviderWriter.delete_dataset) / [](#ProviderWriter.delete_category_tree) — deletion
- [](#ProviderWriter.delete_provider_metadata) / [](#ProviderWriter.delete_dataset_family) — deletion

[DatasetWriter](#DatasetWriter) writes data for a single dataset:

- [](#DatasetWriter.write_dataset_metadata) — write dataset metadata
- [](#DatasetWriter.write_series) — write series (metadata + observations)

### 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](#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:

```python
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:

```python
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:

```python
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.

```python
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-adapter.md) page.