harp_apps.storage.types

class IBlobStorage[source]

Bases: Protocol

__init__(*args, **kwargs)
async delete(blob_id)[source]
Parameters:

blob_id (str)

async exists(blob_id)[source]
Parameters:

blob_id (str)

Return type:

bool

async force_put(blob)[source]
Parameters:

blob (Blob)

Return type:

Blob

forget(blob_ids)[source]

Discard anything cached about these blobs, which have been removed by other means.

A backend that caches “this blob exists” has to be told when something deletes rows without going through delete, or it will keep answering from a belief that is no longer true. Backends holding no such cache may ignore this.

Parameters:

blob_ids (Iterable[str])

Return type:

None

async get(blob_id)[source]
Parameters:

blob_id (str)

async put(blob)[source]
Parameters:

blob (Blob)

Return type:

Blob

type: str
class IStorage[source]

Bases: Protocol

__init__(*args, **kwargs)
async create_users_once_ready(users)[source]

Create users.

Parameters:

users (Iterable[str])

async finalize()[source]

Coroutine function to finalize the instance. Should release resources, close files etc… The caller should make sure to call finalize on all cases where the instance is not needed anymore to get a clean shutdown.

async get_facet_meta(name)[source]

Retrieve a facet’s metadata, by name.

async get_transaction(id, *, username)[source]

Find a transaction, by id.

Parameters:
async get_transaction_list(*, username, with_messages=False, filters=None, page=1, cursor='', text_search='')[source]

Find transactions, using optional filters, for example to be displayed in the dashboard.

Parameters:
async get_usage()[source]

Get storage usage.

async initialize()[source]

Coroutine function to initialize the instance.

async set_user_flag(*, transaction_id, username, flag, value=True)[source]

Sets or unsets a user flag on a transaction.

Parameters:

  • transaction_id (str)

  • username (str)

  • flag (int)

async transactions_grouped_by_time_bucket(*, endpoint, time_bucket, start_datetime)[source]
Parameters:
Return type:

list[TransactionsGroupedByTimeBucket]

class TransactionsGroupedByTimeBucket[source]

Bases: TypedDict

cached: int
count: int
datetime: datetime
errors: int
meanDuration: float
meanTpdex: float

Submodules