swh.vault.backend module

class swh.vault.backend.VaultDB(**config)

Bases: object

PostgreSQL backend for the Software Heritage Vault.

current_version = 4

get_db()

put_db(db)

class swh.vault.backend.VaultBackend(**config)

Bases: VaultDB

Backend for the Software Heritage Vault.

progress(bundle_type: str, swhid: CoreSWHID, raise_notfound: bool = True) → Dict[str, Any] | None

create_task(bundle_type: str, swhid: CoreSWHID, sticky: bool = False)

Create and send a cooking task

add_notif_email(bundle_type: str, swhid: CoreSWHID, email: str)

Add an e-mail address to notify when a given bundle is ready

put_bundle(bundle_type: str, swhid: CoreSWHID, bundle) → bool

cook(bundle_type: str, swhid: CoreSWHID, *, sticky: bool = False, email: str | None = None) → Dict[str, Any]

batch_cook(batch: List[Tuple[str, str]]) → Dict[str, int]

batch_progress(batch_id: int) → Dict[str, Any]

is_available(bundle_type: str, swhid: CoreSWHID)

Check whether a bundle is available for retrieval

fetch(bundle_type: str, swhid: CoreSWHID, raise_notfound=True) → bytes | None

Retrieve a bundle from the cache

download_url(bundle_type: str, swhid: CoreSWHID, content_disposition: str | None = None, expiry: timedelta | None = None, raise_notfound=True) → str | None

Obtain a bundle direct download link from the cache if supported

update_access_ts(bundle_type: str, swhid: CoreSWHID)

Update the last access timestamp of a bundle

set_status(bundle_type: str, swhid: CoreSWHID, status: str) → bool

set_progress(bundle_type: str, swhid: CoreSWHID, progress: str) → bool

send_notif(bundle_type: str, swhid: CoreSWHID) → bool

send_notification(n_id: int | None, email: str, bundle_type: str, swhid: CoreSWHID, status: str, progress_msg: str | None = None) → None

Send the notification of a bundle to a specific e-mail

cache_expire_oldest(n=1, by='last_access') → None

Expire the n oldest bundles

cache_expire_until(date, by='last_access') → None

Expire all the bundles until a certain date