Storage¶

Three orthogonal layers, each with a distinct responsibility:

Layer	What it owns	Module
Blobs	Content-addressed immutable bytes for uploaded images	`app.storage.blobs`
ImageSource	Logical reference to where bytes live (upload / local / S3)	`app.sources.*`
Materialization	Per-job realization of an `ImageSource` to a real path the backend can read	worker-only

Blob store¶

<blob_root>/<sha[:2]>/<sha>

Writes go through BlobStore.put_stream() — atomic via os.replace from a temp file. Reads stream chunked. Blob.refcount (in the DB) tracks how many entities (images, masks, model artifacts) reference the blob; lifecycle is GC’d when refcount → 0.

class app.storage.blobs.BlobStore(*args, **kwargs)[source]

Bases: Protocol

Sha256-keyed binary store. All methods are sha-addressed; the backend chooses where bytes physically live.

exists(sha)[source]

Parameters:: sha (str)
Return type:: bool

put_stream(reader, *, chunk_size=Ellipsis)[source]

Parameters:

reader (BinaryIO)
chunk_size (int)

Return type:

tuple[str, int]

put_bytes(data)[source]

Parameters:: data (bytes)
Return type:: tuple[str, int]

open(sha)[source]

Parameters:: sha (str)
Return type:: BinaryIO

aiter_chunks(sha, *, chunk_size=Ellipsis)[source]

Parameters:

sha (str)
chunk_size (int)

Return type:

AsyncIterator[bytes]

delete(sha)[source]

Parameters:: sha (str)
Return type:: None

local_path(sha)[source]

Return a local filesystem path for the blob’s bytes.

For filesystem backends this is the canonical storage path. For remote backends (S3) the bytes are downloaded into the local cache on first access; subsequent calls return the cached path. Callers that need to hand a real path to a native library (pycolmap, Pillow, OpenCV) should use this.

Parameters:: sha (str)
Return type:: Path

Image sources¶

Three implementations behind a single contract:

class app.sources.upload.UploadSource(kind: 'str' = 'upload', entries: 'list[UploadEntry]' = None)[source]

Bases: object

Parameters:

kind (str)
entries (list[UploadEntry])

kind: str = 'upload'

entries: list[UploadEntry] = None

fingerprint()[source]

Return type:: dict

materialize(into, blob_store=None)[source]

Parameters:

into (Path)
blob_store (BlobStore | None)

Return type:

list[MaterializedImage]

class app.sources.local.LocalPathSource(root: 'Path', kind: 'str' = 'local', recursive: 'bool' = True, extensions: 'tuple[str, ...]' = ('.jpg', '.jpeg', '.png', '.tif', '.tiff', '.bmp', '.webp', '.heic', '.heif'))[source]

Bases: object

Parameters:

root (Path)
kind (str)
recursive (bool)
extensions (tuple[str, ...])

root: Path

kind: str = 'local'

recursive: bool = True

extensions: tuple[str, ...] = ('.jpg', '.jpeg', '.png', '.tif', '.tiff', '.bmp', '.webp', '.heic', '.heif')

fingerprint()[source]

Return type:: dict

materialize(into=None)[source]

Parameters:: into (Path | None)
Return type:: list[MaterializedImage]

class app.sources.s3.S3Source(bucket: 'str', prefix: 'str' = '', kind: 'str' = 's3', extensions: 'tuple[str, ...]' = ('.jpg', '.jpeg', '.png', '.tif', '.tiff', '.bmp', '.webp', '.heic', '.heif'), region_name: 'str | None' = None, endpoint_url: 'str | None' = None, _client: 'Any | None' = None)[source]

Bases: object

Parameters:

bucket (str)
prefix (str)
kind (str)
extensions (tuple[str, ...])
region_name (str | None)
endpoint_url (str | None)
_client (Any | None)

bucket: str

prefix: str = ''

kind: str = 's3'

extensions: tuple[str, ...] = ('.jpg', '.jpeg', '.png', '.tif', '.tiff', '.bmp', '.webp', '.heic', '.heif')

region_name: str | None = None

endpoint_url: str | None = None

fingerprint()[source]

Return type:: dict

materialize(into=None)[source]

Parameters:: into (Path | None)
Return type:: list[MaterializedImage]

Local path: no copy, no symlink¶

LocalPathSource references the user’s directory directly — the backend gets pointed at the user’s path. To detect “user mutated their dir under us,” we record a fingerprint of every file: {path, size, mtime_ns, sample_hash(head/mid/tail 1MiB)}. Cheap, deterministic, fixed-cost regardless of file size.

S3: lazy download to a global LRU cache¶

S3Source.materialize() lists objects (filtered by extension), checks the LRU cache keyed by (bucket, key, etag), downloads any cache miss. Cache is shared across tenants because content is addressed by ETag, not by tenant prefix.

class app.storage.s3_cache.S3Cache(settings=None)[source]

Bases: object

Filesystem-backed LRU.

Atime-driven LRU on the entry’s manifest (.meta.json) so we don’t depend on filesystem atime, which is often disabled.

Parameters:: settings (Settings | None)

path_for(bucket, key, etag)[source]

Parameters:

bucket (str)
key (str)
etag (str)

Return type:

Path

lookup(bucket, key, etag)[source]

Parameters:

bucket (str)
key (str)
etag (str)

Return type:

CachedObject | None

insert(*, bucket, key, etag, src_bytes)[source]

Parameters:

bucket (str)
key (str)
etag (str)
src_bytes (bytes)

Return type:

CachedObject

total_bytes()[source]

Return type:: int

evict_to(*, max_bytes)[source]

Evict least-recently-used entries until under budget. Returns bytes freed.

Parameters:: max_bytes (int)
Return type:: int

clear()[source]

Return type:: None

LRU eviction:

cache = S3Cache()
cache.evict_to(max_bytes=10 * 1024**3)   # 10 GiB budget

Eviction order is by meta.json mtime (touched on every lookup()), so MRU entries survive.

Sealed snapshots¶

The worker’s only safe handoff to the API is a sealed snapshot dir. The protocol:

Write to snapshots/.tmp_{seq}/
Write .complete marker last
os.replace(tmp, snapshots/{seq}) — atomic rename
Update latest text file (also via tmp+rename)

Readers list snapshots/*/ and ignore any dir without a .complete file. The API never opens a non-sealed file, ever.

class app.storage.snapshots.SnapshotStore(root)[source]

Bases: object

Parameters:: root (Path)

seal(*, seq, source_dir, summary=None)[source]

Parameters:

seq (int)
source_dir (Path)
summary (dict | None)

Return type:

Path

list_sealed()[source]

Return type:: list[int]

latest()[source]

Return type:: int | None

path_for(seq)[source]

Parameters:: seq (int)
Return type:: Path

gc(*, keep_last=3)[source]

Parameters:: keep_last (int)
Return type:: list[int]

MappingInput checkpoints¶

For incremental SfM resume, the worker writes MappingInput.save() payloads under jobs/{job_id}/checkpoints/{seq}.pcmapin every N image registrations. On a re-run / resume, the worker calls pipeline.set_mapping_input(MappingInput.load(latest)) so the expensive setup work isn’t repeated.

MappingInput checkpoint storage helpers.

pycolmap.MappingInput.save/load writes a binary container with magic PCMAPINx00 (v1) holding the pre-mapping state needed to resume an incremental run. We persist these per-job under jobs/{job_id}/checkpoints/{seq}.pcmapin so the worker can reload from the latest checkpoint on resume without re-running upstream stages.

The pycolmap binding is required to read/write — for tests we expose a trivial adapter that round-trips a payload through a regular file so the storage code path can be exercised independently.

class app.storage.mapping_input.CheckpointRef(seq: 'int', path: 'Path', summary: 'dict')[source]

Bases: object

Parameters:

seq (int)
path (Path)
summary (dict)

seq: int

path: Path

summary: dict

app.storage.mapping_input.checkpoint_root(job_dir)[source]

Parameters:: job_dir (Path)
Return type:: Path

app.storage.mapping_input.write_checkpoint(job_dir, *, seq, payload, summary=None)[source]

Parameters:

job_dir (Path)
seq (int)
payload (bytes)
summary (dict | None)

Return type:

CheckpointRef

app.storage.mapping_input.list_checkpoints(job_dir)[source]

Parameters:: job_dir (Path)
Return type:: list[CheckpointRef]

app.storage.mapping_input.latest_checkpoint(job_dir)[source]

Parameters:: job_dir (Path)
Return type:: CheckpointRef | None

app.storage.mapping_input.gc_checkpoints(job_dir, *, keep_last=5)[source]

Parameters:

job_dir (Path)
keep_last (int)

Return type:

list[int]

GC¶

async app.storage.workspace.gc_completed_jobs(session, *, ttl_days, now=None, drop_db_rows=False)[source]

Sweep completed-and-old jobs. Returns a summary dict for tests.

Parameters:

session (AsyncSession)
ttl_days (int)
now (datetime | None)
drop_db_rows (bool)

Return type:

dict

Drops dense → snapshots → sparse per-job in that order, skips pinned jobs (Job.pinned=true), and never touches a job’s manifest.json. Reconstruction-level artifacts are NOT touched by job GC because reconstructions can outlive the job that produced them.