Acquisition

The acquisition module handles data fetching from external sources with built-in caching and a synchronous HTTP client. Connectors use RawCache for on-disk Parquet caching.

Architecture

Connector
    │
    ▼
┌─────────────┐         ┌──────────┐
│   Client    │────────▶│  Cache   │
│ (Rust core) │         │ (Parquet)│
└─────────────┘         └──────────┘

RawCache

The Parquet-based disk cache. All connectors share this cache layer.

RawCache(cache_dir=None)

from hermes.acquisition import RawCache
 
cache = RawCache()  # defaults to ~/.hermes_cache/raw/
cache = RawCache(cache_dir="/tmp/hermes_cache")

Methods

get(source, params, ttl=None) -> pl.DataFrame

Retrieve a cached DataFrame. Raises CacheMiss on miss, expiry, or corruption.

from hermes.acquisition import RawCache, CacheMiss
 
cache = RawCache()
try:
    df = cache.get("world_bank", {"country": "US"})
except CacheMiss:
    print("Not cached or expired")

put(source, params, df)

Store a DataFrame in the cache.

cache.put("world_bank", {"country": "US"}, df)

get_or_fetch(source, params, fetch_fn, force=False, ttl=None) -> pl.DataFrame

Returns cached data if available, otherwise calls fetch_fn, caches the result, and returns it.

import polars as pl
 
def fetch_data():
    return pl.DataFrame({"value": [1, 2, 3]})
 
df = cache.get_or_fetch(
    source="world_bank",
    params={"country": "US"},
    fetch_fn=fetch_data,
)

clear(older_than=None)

Delete cached files. Pass older_than as a timedelta to only remove old entries.

from datetime import timedelta
 
cache.clear()                          # clear everything
cache.clear(older_than=timedelta(days=7))  # clear files older than 7 days

stats() -> dict

Returns hit/miss statistics per source.

stats = cache.stats()
print(stats["total_files"])
print(stats["by_source"])
print(stats["hit_rate"])

exists(source, params, ttl=None) -> bool

Check if a cache entry exists without raising CacheMiss.

delete(source, params)

Remove a specific cache entry.

Cache Keys

Keys are SHA-256 hashes of source + JSON-serialized params, truncated to 16 characters. Each cached file has a .meta.json sidecar with source, params, timestamp, row count, and column names.

Client

Synchronous HTTP client backed by the Hermes Rust core. Transport-fault and 5xx retries are handled in Rust (idempotent methods only); this layer maps status codes and raw errors onto Hermes error types.

Client(base_url, timeout=30.0, max_retries=0, ...)

from hermes.acquisition import Client
 
with Client(base_url="https://api.worldbank.org/v2") as client:
    data = client.get("/countries/US/indicators/NY.GDP.MKTP.KD.ZG")

Methods

• get(url, **kwargs) — HTTP GET
• post(url, **kwargs) — HTTP POST
• put(url, **kwargs) — HTTP PUT
• delete(url, **kwargs) — HTTP DELETE
• stream(method, url, **kwargs) — Chunked streaming

Error Mapping

Using Connectors

The working path for acquiring data is to instantiate a connector directly. Every connector’s fetch() uses the shared RawCache unless cache=None is passed, and takes the same retries/timeout knobs:

from hermes.connectors import World_bank, FRED, Binance
 
wb = World_bank()
df = wb.fetch(country_code="US", indicator_code="NY.GDP.MKTP.KD.ZG")
print(df.head())

Registered connectors (exported from hermes.connectors):

Top-Level API

The top-level hr.fetch, hr.fetch_raw, hr.ingest, hr.sync, and hr.read wrappers are scaffolded — they currently return NotImplementedError. Use the connector classes directly until these land.