Storage

Persist simulation results with file-system or S3 backends.

Overview

Asgard provides two storage layers:

Core storage (FileStorage, S3Storage) — low-level persistence for any Serialisable object
Monitor storage (AsgardStorage) — high-level run/result management used by the CLI and monitor dashboard

Both share the same directory layout:

storage_path/
├── runs/
│   └── {uuid}.json
├── results/
│   └── {uuid}.json
└── files/
    └── {uuid}_{field}.npy

File Storage

For local development and single-machine use.

from gimle.asgard.core.data.file import FileStorage

storage = FileStorage({"base_path": "./storage"})

Objects are stored as {base_path}/{obj_type}/{name}.json. Directories are created automatically on save.

S3 Storage

For shared or cloud deployments.

from gimle.asgard.core.data.s3 import S3Storage

storage = S3Storage({
    "bucket_name": "my-asgard-bucket",
    "base_path": "experiments/2026-02",
})

AWS credentials are read from the environment by default (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION). You can also pass them explicitly:

storage = S3Storage({
    "bucket_name": "my-bucket",
    "base_path": "runs",
    "aws_access_key_id": "AKIA...",
    "aws_secret_access_key": "...",
    "region_name": "eu-west-1",
})

Storage Methods

Both backends implement the same interface:

Method	Description
`path()`	Return the base path.
`list_objects(obj_type)`	List all object names of a given type.
`save(obj, name, obj_type)`	Save a `Serialisable` object.
`load(name, obj_type)`	Load raw JSON string for an object.
`delete(name, obj_type)`	Delete an object.

Caching

Wrap any backend with LocalCache to avoid repeated reads:

from gimle.asgard.core.data.local_cache import LocalCache

cached = LocalCache(storage)
cached.list_objects("run")   # hits disk/S3
cached.list_objects("run")   # cache hit

The cache is updated automatically on save() and delete().

Monitor Storage

AsgardStorage is the high-level interface used by the CLI example and monitor commands. It organises results into runs (a simulation session) and results (individual outputs within a run).

from gimle.asgard.monitor.storage import AsgardStorage

storage = AsgardStorage("./storage")

Creating Runs and Results

# Create a run
run = storage.create_run(
    name="Exponential ODE",
    equation="int(f, x) = f",
    circuit="trace(composition(register(x), split))",
)

# Add a result to the run
storage.add_result(
    run,
    result_type="Solution1D",
    data={"coefficients": [1.0, 1.0, 0.5, 0.167]},
    display={"title": "exp(x) approximation"},
)

One-Liner Convenience

run = storage.save_simulation_result(
    name="Quick test",
    result_type="Solution1D",
    data={"coefficients": [1.0, 1.0, 0.5]},
    equation="int(f, x) = f",
)

Methods

Method	Description
`create_run(name, equation, circuit, metadata)`	Create a new simulation run.
`load_run(run_id)`	Load a run by UUID.
`list_runs()`	List all runs, newest first.
`delete_run(run_id)`	Delete a run and all its results.
`add_result(run, result_type, data, display)`	Add a result to a run.
`load_result(result_id)`	Load a result by UUID.
`delete_result(result_id)`	Delete a result.
`save_simulation_result(name, result_type, data, ...)`	Create run + result in one call.
`get_run_with_results(run_id)`	Load run and all results together.

CLI Integration

# Run examples and save to monitor storage
uv run asgard example -c basic --monitor

# Specify a custom storage directory
uv run asgard example -c basic --storage /data/asgard

# Launch the monitor dashboard
uv run asgard monitor --storage /data/asgard --port 8080

File vs S3

Aspect	FileStorage	S3Storage
Setup	Directory path	Bucket + credentials
Latency	Milliseconds	100-500 ms
Concurrency	Single machine	Multi-region safe
Cost	Free	Per-request
Use case	Development	Shared / cloud

Custom Serialisable Objects

Any class inheriting from Serialisable can be stored:

from gimle.asgard.core.data.serialisable import Serialisable

class MyResult(Serialisable):
    def __init__(self, values):
        super().__init__()
        self.values = values

    def to_json_impl(self):
        return {"values": self.values}

    @classmethod
    def from_json_impl(cls, data):
        return cls(values=data["values"])

The Serialisable base class adds created_at and updated_at timestamps automatically and registers the class for deserialization via SerialisableRegistry.