aea.helpers.storage.generic_
storage
This module contains the storage implementation.
AsyncCollection Objects
class AsyncCollection()
Async collection.
__
init__
| __init__(storage_backend: AbstractStorageBackend, collection_name: str)
Init collection object.
Arguments:
storage_backend
: storage backed to use.collection_name
: srt
put
| async put(object_id: str, object_body: JSON_TYPES) -> None
Put object into collection.
Arguments:
object_id
: str object idobject_body
: python dict, json compatible.
Returns:
None
get
| async 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 remove(object_id: str) -> None
Remove object from the collection.
Arguments:
object_id
: str object id
Returns:
None
find
| async 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 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__
| __init__(async_collection_coro, loop: asyncio.AbstractEventLoop)
Init collection object.
Arguments:
async_collection_coro
: coroutine returns async collection.loop
: abstract event loop where storage is running.
put
| put(object_id: str, object_body: JSON_TYPES) -> None
Put object into collection.
Arguments:
object_id
: str object idobject_body
: python dict, json compatible.
Returns:
None
get
| 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
| remove(object_id: str) -> None
Remove object from the collection.
Arguments:
object_id
: str object id
Returns:
None
find
| 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
| 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__
| __init__(storage_uri: str, loop: asyncio.AbstractEventLoop = None, threaded: bool = False) -> None
Init stortage.
Arguments:
storage_uri
: configuration string for storage.loop
: asyncio event loop to use.threaded
: bool. start in thread if True.
Returns:
None
wait_
connected
| async wait_connected() -> None
Wait generic storage is connected.
is_
connected
| @property
| is_connected() -> bool
Get running state of the storage.
run
| async run()
Connect storage.
get_
collection
| async get_collection(collection_name: str) -> AsyncCollection
Get async collection.
get_
sync_
collection
| get_sync_collection(collection_name: str) -> SyncCollection
Get sync collection.
__
repr__
| __repr__() -> str
Get string representation of the storage.