Skip to content

aea.helpers.storage.generic_storage

This module contains the storage implementation.

AsyncCollection Objects

class AsyncCollection()

Async collection.

__init__

def __init__(storage_backend: AbstractStorageBackend,
             collection_name: str) -> None

Init collection object.

Arguments:

  • storage_backend: storage backed to use.
  • collection_name: str

put

async def put(object_id: str, object_body: JSON_TYPES) -> None

Put object into collection.

Arguments:

  • object_id: str object id
  • object_body: python dict, json compatible.

Returns:

None

get

async def get(object_id: str) -> Optional[JSON_TYPES]

Get object from the collection.

Arguments:

  • object_id: str object id

Returns:

dict if object exists in collection otherwise None

remove

async def remove(object_id: str) -> None

Remove object from the collection.

Arguments:

  • object_id: str object id

Returns:

None

find

async def find(field: str, equals: EQUALS_TYPE) -> List[OBJECT_ID_AND_BODY]

Get objects from the collection by filtering by field value.

Arguments:

  • field: field name to search: example "parent.field"
  • equals: value field should be equal to

Returns:

None

list

async def list() -> List[OBJECT_ID_AND_BODY]

List all objects with keys from the collection.

Returns:

Tuple of objects keys, bodies.

SyncCollection Objects

class SyncCollection()

Async collection.

__init__

def __init__(async_collection_coro: Coroutine,
             loop: asyncio.AbstractEventLoop) -> None

Init collection object.

Arguments:

  • async_collection_coro: coroutine returns async collection.
  • loop: abstract event loop where storage is running.

put

def put(object_id: str, object_body: JSON_TYPES) -> None

Put object into collection.

Arguments:

  • object_id: str object id
  • object_body: python dict, json compatible.

Returns:

None

get

def get(object_id: str) -> Optional[JSON_TYPES]

Get object from the collection.

Arguments:

  • object_id: str object id

Returns:

dict if object exists in collection otherwise None

remove

def remove(object_id: str) -> None

Remove object from the collection.

Arguments:

  • object_id: str object id

Returns:

None

find

def find(field: str, equals: EQUALS_TYPE) -> List[OBJECT_ID_AND_BODY]

Get objects from the collection by filtering by field value.

Arguments:

  • field: field name to search: example "parent.field"
  • equals: value field should be equal to

Returns:

List of object bodies

list

def list() -> List[OBJECT_ID_AND_BODY]

List all objects with keys from the collection.

Returns:

Tuple of objects keys, bodies.

Storage Objects

class Storage(Runnable)

Generic storage.

__init__

def __init__(storage_uri: str,
             loop: Optional[asyncio.AbstractEventLoop] = None,
             threaded: bool = False) -> None

Init storage.

Arguments:

  • storage_uri: configuration string for storage.
  • loop: asyncio event loop to use.
  • threaded: bool. start in thread if True.

wait_connected

async def wait_connected() -> None

Wait generic storage is connected.

is_connected

@property
def is_connected() -> bool

Get running state of the storage.

run

async def run() -> None

Connect storage.

get_collection

async def get_collection(collection_name: str) -> AsyncCollection

Get async collection.

get_sync_collection

def get_sync_collection(collection_name: str) -> SyncCollection

Get sync collection.

__repr__

def __repr__() -> str

Get string representation of the storage.