app.services

Tenant-scoped CRUD + transaction + DAG construction. Routes call into here; never the other way around.

Project CRUD.

async app.services.project_service.create_project(session, *, tenant_id, name, description=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• name (str)

• description (str | None)

Return type:

Project

async app.services.project_service.get_project(session, *, tenant_id, project_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

Return type:

Project

async app.services.project_service.list_projects(session, *, tenant_id, page_size=50, page_token=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• page_size (int)

• page_token (str | None)

Return type:

tuple[list[Project], str | None]

async app.services.project_service.patch_project(session, *, tenant_id, project_id, updates)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• updates (dict)

Return type:

Project

async app.services.project_service.delete_project(session, *, tenant_id, project_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

Return type:

None

Dataset CRUD + manifest_hash recompute + ImageSource creation.

async app.services.dataset_service.create_image_source(session, *, tenant_id, source)

Persist an ImageSource row from a request-side source spec.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• source (Annotated[UploadSourceSpec | LocalSourceSpec | S3SourceSpec, FieldInfo(annotation=NoneType, required=True, discriminator='kind')])

Return type:

ImageSource

async app.services.dataset_service.create_dataset(session, *, tenant_id, project_id, source_id, name, camera_model='SIMPLE_RADIAL', intrinsics_mode='single_camera', is_spherical=False, rig_config=None, respect_exif_orientation=False)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• source_id (str)

• name (str)

• camera_model (str)

• intrinsics_mode (str)

• is_spherical (bool)

• rig_config (dict | None)

• respect_exif_orientation (bool)

Return type:

Dataset

async app.services.dataset_service.get_dataset(session, *, tenant_id, dataset_id, project_id=None)

Load a Dataset by id, scoped to tenant_id.

Pass project_id to additionally enforce that the row belongs to that project — raises ValidationError on mismatch. Routes nested under /projects/{pid}/datasets/{did} should pass it; top-level reads (/v1/datasets/{did}/...) leave it None.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• project_id (str | None)

Return type:

Dataset

async app.services.dataset_service.list_datasets(session, *, tenant_id, project_id, page_size=100, page_token=None)

AIP-158 keyset pagination on dataset_id ascending.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• page_size (int)

• page_token (str | None)

Return type:

tuple[list[Dataset], str | None]

async app.services.dataset_service.delete_dataset(session, *, tenant_id, dataset_id)

Cascade-delete a dataset and its images. Blob refcounts are decremented for upload-sourced images. Caller is responsible for workspace cleanup if any.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

Return type:

None

async app.services.dataset_service.patch_dataset(session, *, tenant_id, dataset_id, updates)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• updates (dict)

Return type:

Dataset

async app.services.dataset_service.recompute_manifest_hash(session, *, dataset_id)

Parameters:

• session (AsyncSession)

• dataset_id (str)

Return type:

str

Image CRUD inside a dataset.

async app.services.image_service.add_image(session, *, tenant_id, dataset, name, content_sha, source_kind, rel_path=None, byte_size=None, width=None, height=None, exif=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset (Dataset)

• name (str)

• content_sha (str)

• source_kind (str)

• rel_path (str | None)

• byte_size (int | None)

• width (int | None)

• height (int | None)

• exif (dict[str, Any] | None)

Return type:

Image

async app.services.image_service.list_images(session, *, tenant_id, dataset_id, page_size=100, page_token=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• page_size (int)

• page_token (str | None)

Return type:

tuple[list[Image], str | None]

async app.services.image_service.get_image(session, *, tenant_id, image_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• image_id (str)

Return type:

Image

async app.services.image_service.set_pose_prior(session, *, tenant_id, image_id, prior)

Attach (or clear) a PosePrior on an image. Pass None to clear.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• image_id (str)

• prior (dict[str, Any] | None)

Return type:

Image

async app.services.image_service.list_pose_priors(session, *, tenant_id, dataset_id)

Return (image, prior_dict) for every image in the dataset that carries a non-null pose_prior_json. Order: by image name.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

Return type:

list[tuple[Image, dict[str, Any]]]

async app.services.image_service.delete_image(session, *, tenant_id, dataset_id, name)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• name (str)

Return type:

None

async app.services.image_service.delete_image_by_id(session, *, tenant_id, image_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• image_id (str)

Return type:

None

Chunked upload state management.

async app.services.upload_service.init_upload(session, *, tenant_id, expected_size, content_type, expected_sha, idempotency_key, settings=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• expected_size (int)

• content_type (str | None)

• expected_sha (str | None)

• idempotency_key (str | None)

• settings (Settings | None)

Return type:

Upload

async app.services.upload_service.get_upload(session, *, tenant_id, upload_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• upload_id (str)

Return type:

Upload

async app.services.upload_service.append_chunk(session, *, tenant_id, upload_id, offset, data)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• upload_id (str)

• offset (int)

• data (bytes)

Return type:

Upload

async app.services.upload_service.finalize_upload(session, *, tenant_id, upload_id, client_sha=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• upload_id (str)

• client_sha (str | None)

Return type:

Upload

async app.services.upload_service.gc_expired_uploads(session, *, now=None)

Parameters:

• session (AsyncSession)

• now (datetime | None)

Return type:

int

Job CRUD + cache lookup + DAG persistence.

The orchestrator owns DAG building (in app.orchestrator.dag). This service persists Jobs and Tasks to the DB and is responsible for cache short-circuit logic.

async app.services.job_service.create_job(session, *, tenant_id, project_id, recipe, spec)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• recipe (str)

• spec (dict[str, Any] | None)

Return type:

Job

async app.services.job_service.get_job(session, *, tenant_id, job_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• job_id (str)

Return type:

Job

async app.services.job_service.list_jobs(session, *, tenant_id, page_size=50, page_token=None, status=None)

AIP-158 keyset pagination on job_id descending (most recent submissions first). status filters to a single lifecycle state when set — the canonical cheap-filter exposed in lieu of a full AIP-160 grammar.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• page_size (int)

• page_token (str | None)

• status (str | None)

Return type:

tuple[list[Job], str | None]

async app.services.job_service.lookup_cached_task(session, *, tenant_id, cache_key)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• cache_key (str)

Return type:

Task | None

async app.services.job_service.materialize_dag(session, *, tenant_id, job_id, runtime_version_id, nodes)

Persist task nodes to the DB. Looks up cache hits per task; if a successful Task with the same cache_key exists, copies its outputs_ref_json and marks the new Task succeeded immediately.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• job_id (str)

• runtime_version_id (str)

• nodes (Iterable[TaskNode])

Return type:

list[Task]

async app.services.job_service.cancel_job(session, *, tenant_id, job_id, force)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• job_id (str)

• force (bool)

Return type:

Job

Reconstruction + SubModel CRUD + Snapshot reads.

async app.services.reconstruction_service.get_reconstruction(session, *, tenant_id, recon_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

Return type:

Reconstruction

async app.services.reconstruction_service.list_submodels(session, *, tenant_id, recon_id, page_size=100, page_token=None)

AIP-158 keyset pagination on submodel_id ascending; idx determines display order but submodel_id is the cursor key (stable even when components are added / removed mid-iteration).

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• page_size (int)

• page_token (str | None)

Return type:

tuple[list[SubModel], str | None]

async app.services.reconstruction_service.get_submodel(session, *, tenant_id, submodel_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• submodel_id (str)

Return type:

SubModel

async app.services.reconstruction_service.mark_reconstruction_status(session, *, tenant_id, recon_id, status)

Update a reconstruction lifecycle state if the row still exists.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• status (str)

Return type:

None

async app.services.reconstruction_service.record_mapping_result(session, *, tenant_id, recon_id, models, snapshot_seq, snapshot_path)

Replace SubModel rows with the components emitted by a map task.

Backends return one summary per disconnected mapping component. The snapshot writer seals those component files under the same sequence; this function makes the resource layer reflect that output.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• models (list[dict[str, Any]])

• snapshot_seq (int | None)

• snapshot_path (str | None)

Return type:

None

app.services.reconstruction_service.list_snapshot_seqs(paths, tenant_id, project_id, recon_id)

Parameters:

• paths (Paths)

• tenant_id (str)

• project_id (str)

• recon_id (str)

Return type:

list[int]

app.services.reconstruction_service.snapshot_dir(paths, tenant_id, project_id, recon_id, seq)

Parameters:

• paths (Paths)

• tenant_id (str)

• project_id (str)

• recon_id (str)

• seq (int)

Return type:

Path

Build the Job -> Task DAG for SfM stage calls and named recipes.

The HTTP layer no longer needs to pass image_root / image_list / database_path — they are derived here from the dataset’s source and the persisted Image rows. This keeps the API surface clean (the client knows about its dataset, not about worker-side filesystem layout) and ensures the same materialization logic is used for every stage.

A recipe (incremental | global | hierarchical | spherical) is sugar over the per-stage builders: it strings extract → match → verify → map into one DAG so per-stage caching short-circuits as soon as any prefix is reused.

async app.services.sfm_stage_service.ensure_reconstruction(session, *, tenant_id, dataset, spec)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset (Dataset)

• spec (dict[str, Any])

Return type:

Reconstruction

async app.services.sfm_stage_service.derive_materialization(session, *, tenant_id, dataset)

Return a manifest the worker uses to realize the dataset on disk.

Shape:

{
  "kind": "upload" | "local" | "s3",
  "image_list": [str],            # ordered image names
  "image_root": str | None,       # filesystem path (None for upload)
  "blob_shas": {name: sha},       # only for upload
  "bucket": str, "prefix": str,   # only for s3
}

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset (Dataset)

Return type:

dict[str, Any]

app.services.sfm_stage_service.reconstruction_database_path(tenant_id, project_id, recon_id)

Parameters:

• tenant_id (str)

• project_id (str)

• recon_id (str)

Return type:

str

app.services.sfm_stage_service.validate_features_config(spec, *, project_id=None)

Parameters:

• spec (dict[str, Any])

• project_id (str | None)

Return type:

None

app.services.sfm_stage_service.validate_matches_config(spec, *, project_id=None)

Parameters:

• spec (dict[str, Any])

• project_id (str | None)

Return type:

None

app.services.sfm_stage_service.validate_verify_config(spec, *, project_id=None)

Parameters:

• spec (dict[str, Any])

• project_id (str | None)

Return type:

None

app.services.sfm_stage_service.validate_mapping_config(spec, *, project_id=None)

Parameters:

• spec (dict[str, Any])

• project_id (str | None)

Return type:

None

app.services.sfm_stage_service.validate_recipe_stage_configs(*, features_spec, matches_spec, verify_spec, pipeline_spec, project_id=None)

Parameters:

• features_spec (dict[str, Any])

• matches_spec (dict[str, Any])

• verify_spec (dict[str, Any])

• pipeline_spec (dict[str, Any])

• project_id (str | None)

Return type:

None

async app.services.sfm_stage_service.submit_features(session, *, tenant_id, dataset_id, spec, input_artifacts=None, inline=False)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• spec (dict[str, Any])

• input_artifacts (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_matches(session, *, tenant_id, dataset_id, spec, input_artifacts=None, inline=False)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• spec (dict[str, Any])

• input_artifacts (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_render_cubemap(session, *, tenant_id, dataset_id, face_size=None, spec=None, inline=False)

Render every spherical panorama into 6 cubemap faces.

Refuses if the dataset isn’t is_spherical=true. The output is a directory under the dataset’s workspace; the user can register it as a new local dataset for downstream pinhole-only pipelines.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• face_size (int | None)

• spec (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_project_images(session, *, tenant_id, dataset_id, spec, recipe='project_images', inline=False)

Submit a portable image projection job over a dataset.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• spec (dict[str, Any])

• recipe (str)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_video_frames(session, *, tenant_id, project_id, video_path, fps=2.0, max_frames=1000, inline=False)

Extract keyframes from a worker-local video file.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• video_path (str)

• fps (float)

• max_frames (int)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_kapture_import(session, *, tenant_id, project_id, archive_path, inline=False)

Parse a Kapture archive (extracted directory) into sensors and records lists in the task result. The client follows up with a POST /v1/projects/{pid}/datasets of kind=``local`` pointing at the returned image_root.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• project_id (str)

• archive_path (str)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_merge_recons(session, *, tenant_id, target_recon_id, source_recon_ids, sim3_aligners=None, inline=False)

Merge several reconstructions into target_recon_id.

All sources MUST belong to the same project as the target.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• target_recon_id (str)

• source_recon_ids (list[str])

• sim3_aligners (list[dict[str, Any]] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_to_cubemap(session, *, tenant_id, recon_id, inline=False)

Convert a spherical reconstruction to a cubemap rig (worker job).

Refuses if the reconstruction’s dataset isn’t marked is_spherical — this avoids running the converter against a pinhole reconstruction (which produces nonsense).

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_georegister(session, *, tenant_id, recon_id, sim3, inline=False)

Apply a Sim(3) georegistration transform to a reconstruction.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• sim3 (dict[str, Any])

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_localize(session, *, tenant_id, recon_id, blob_sha, spec=None, inline=False)

Localize a single query image (blob_sha) against recon_id.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• recon_id (str)

• blob_sha (str)

• spec (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_vlad_index(session, *, tenant_id, dataset_id, spec=None, inline=False)

Build a VLAD descriptor index for the dataset (worker job).

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• spec (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.submit_verify(session, *, tenant_id, dataset_id, spec, input_artifacts=None, inline=False)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

• spec (dict[str, Any])

• input_artifacts (dict[str, Any] | None)

• inline (bool)

Return type:

tuple[str, list[Any]]

async app.services.sfm_stage_service.collect_pose_priors_by_name(session, *, tenant_id, dataset_id)

Return {image_name: PosePrior dict} for every image in the dataset that has a non-null pose_prior_json. Keyed by name so the worker can correlate with pycolmap’s image-name primary key.

Parameters:

• session (AsyncSession)

• tenant_id (str)

• dataset_id (str)

Return type:

dict[str, dict[str, Any]]

app.services.sfm_stage_service.build_recipe_dag(*, project_id, dataset_id, recon_id, materialization, database_path, features_spec, matches_spec, verify_spec, pipeline_spec, pose_priors=None, input_artifacts=None)

Stitch extract → match → verify → map into one DAG. Each TaskNode is hashed with the same shape as a single-stage submission, so a recipe that re-uses an already-computed extract+match prefix short-circuits to the cached results.

pose_priors are optional per-image priors (keyed by image name); when present they’re forwarded into the map task’s inputs so the worker can wire them into pycolmap’s MappingInput.

Parameters:

• project_id (str)

• dataset_id (str)

• recon_id (str)

• materialization (dict[str, Any])

• database_path (str)

• features_spec (dict[str, Any])

• matches_spec (dict[str, Any])

• verify_spec (dict[str, Any])

• pipeline_spec (dict[str, Any])

• pose_priors (dict[str, dict[str, Any]] | None)

• input_artifacts (dict[str, dict[str, Any]] | None)

Return type:

list[TaskNode]

API key issuance + tenant resolution.

async app.services.api_key_service.issue_key(session, *, tenant_id, name=None)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• name (str | None)

Return type:

tuple[str, ApiKey]

async app.services.api_key_service.resolve_tenant(session, *, key)

Parameters:

• session (AsyncSession)

• key (str)

Return type:

str

Quota enforcement hooks (NOOP unless auth_mode=api_key).

async app.services.quota_service.get_or_create_quota(session, *, tenant_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

Return type:

TenantQuota

async app.services.quota_service.check_storage(session, *, tenant_id, additional)

Parameters:

• session (AsyncSession)

• tenant_id (str)

• additional (int)

Return type:

None

async app.services.quota_service.check_gpu_seconds(session, *, tenant_id)

Parameters:

• session (AsyncSession)

• tenant_id (str)

Return type:

None

runtime_version row management.

async app.services.runtime_version_service.ensure_runtime_version(session, settings=None)

Parameters:

• session (AsyncSession)

• settings (Settings | None)

Return type:

RuntimeVersion