acapy_agent.database_manager package

Database manager module with automatic backend registration.

Subpackages

Submodules

acapy_agent.database_manager.category_registry module

Category registry for database managers.

acapy_agent.database_manager.category_registry.get_release(release_number: str, db_type: str = 'sqlite') → Tuple[dict, dict, dict][source]: Retrieve handlers and schemas for a given release number and database type.

acapy_agent.database_manager.category_registry.load_release(release_number: str) → dict[source]: Load the release configuration from its module.

acapy_agent.database_manager.category_registry.load_schema(category: str, version: str) → dict[source]: Load schema from the versioned file.

acapy_agent.database_manager.db_errors module

Unified database error helpers across Askar and DBStore.

Provides a common tuple of exception types and grouped error-code sets so business logic can remain agnostic to the underlying storage backend.

class acapy_agent.database_manager.db_errors.DBCode[source]

Bases: object

Unified code groups: use in membership checks.

Example

try:

repo.save(record)

except DBError as err:

if err.code not in DBCode.DUPLICATE:: raise

DUPLICATE: FrozenSet[aries_askar.AskarErrorCode | DBStoreErrorCode] = frozenset({DBStoreErrorCode.DUPLICATE, aries_askar.AskarErrorCode.DUPLICATE})

NOT_FOUND: FrozenSet[aries_askar.AskarErrorCode | DBStoreErrorCode] = frozenset({DBStoreErrorCode.NOT_FOUND, aries_askar.AskarErrorCode.NOT_FOUND})

acapy_agent.database_manager.db_types module

Database types and data structures for database manager.

class acapy_agent.database_manager.db_types.Entry(category: str, name: str, value: str | bytes, tags: dict)[source]

Bases: object

A single result from a store query.

property category: str: Get the entry category.

keys()[source]: Get the entry keys.

property name: str: Get the entry name.

property raw_value: memoryview: Get the entry value as raw bytes.

property tags: dict: Get the entry tags.

property value: str: Get the entry value.

property value_json: dict: Get the entry value parsed as JSON.

class acapy_agent.database_manager.db_types.EntryList(entries: Sequence[Entry], length: int = None)[source]

Bases: object

A list of Entry objects.

property handle: Get dummy handle for compatibility.

class acapy_agent.database_manager.db_types.EntryOperation(*values)[source]

Bases: Enum

Enumeration of database entry operations.

INSERT = 0

REMOVE = 2

REPLACE = 1

class acapy_agent.database_manager.db_types.IterEntryList(list: EntryList)[source]

Bases: object

Iterator for EntryList.

class acapy_agent.database_manager.db_types.KeyAlg(*values)[source]

Bases: Enum

Enumeration of supported key algorithms.

A128CBC_HS256 = 'a128cbchs256'

A128GCM = 'a128gcm'

A128KW = 'a128kw'

A256CBC_HS512 = 'a256cbchs512'

A256GCM = 'a256gcm'

A256KW = 'a256kw'

BLS12_381_G1 = 'bls12381g1'

BLS12_381_G1G2 = 'bls12381g1g2'

BLS12_381_G2 = 'bls12381g2'

C20P = 'c20p'

ED25519 = 'ed25519'

K256 = 'k256'

P256 = 'p256'

X25519 = 'x25519'

XC20P = 'xc20p'

classmethod from_key_alg(alg: str) → KeyAlg | None[source]: Get KeyAlg instance from the algorithm identifier.

class acapy_agent.database_manager.db_types.SeedMethod(*values)[source]

Bases: Enum

Enumeration of supported seed methods.

BlsKeyGen = 'bls_keygen'

classmethod from_seed_method(method: str) → SeedMethod | None[source]: Get SeedMethod instance from the method identifier.

acapy_agent.database_manager.dbstore module

Database store module for managing different database backends.

class acapy_agent.database_manager.dbstore.DBOpenSession(db: AbstractDatabaseStore, profile: str | None, is_txn: bool, release_number: str)[source]

Bases: object

Database open session class.

property is_transaction: bool: Perform the action.

class acapy_agent.database_manager.dbstore.DBStore(db: AbstractDatabaseStore, uri: str, release_number: str = 'release_0')[source]

Bases: object

Database store class.

async close(*, remove: bool = False) → bool[source]: Perform the action.

async create_profile(name: str = None) → str[source]: Perform the action.

classmethod generate_raw_key(seed: str | bytes | None = None) → str[source]: Perform the action.

async get_profile_name() → str[source]: Perform the action.

property handle: Perform the action.

async initialize() → None[source]: Initialize the database store.

async classmethod open(uri: str, key_method: str = None, pass_key: str = None, *, profile: str = None, schema_migration: bool | None = None, target_schema_release_number: str | None = None, config: dict | None = None) → DBStore[source]: Perform the action.

async classmethod provision(uri: str, key_method: str = None, pass_key: str = None, *, profile: str = None, recreate: bool = False, release_number: str = 'release_0', schema_config: str | None = None, config: dict | None = None) → DBStore[source]: Provision a new database store with specified release and schema.

async rekey(key_method: str = None, pass_key: str = None) → None[source]: Perform the action.

property release_number: str: Perform the action.

async classmethod remove(uri: str, release_number: str = 'release_0', config: dict | None = None) → bool[source]: Remove the database store.

async remove_profile(name: str) → bool[source]: Perform the action.

scan(category: str, tag_filter: str | dict = None, offset: int = None, limit: int = None, profile: str = None, order_by: str | None = None, descending: bool = False) → Scan[source]: Scan the database for entries matching the criteria.

scan_keyset(category: str, tag_filter: str | dict = None, last_id: int | None = None, limit: int = None, profile: str = None, order_by: str | None = None, descending: bool = False) → ScanKeyset[source]: Scan the database using keyset pagination.

session(profile: str = None) → DBOpenSession[source]: Perform the action.

transaction(profile: str = None) → DBOpenSession[source]: Perform the action.

property uri: str: Perform the action.

class acapy_agent.database_manager.dbstore.DBStoreSession(db_session: AbstractDatabaseSession, is_txn: bool)[source]

Bases: object

Database store session class.

async close() → None[source]: Perform the action.

async commit() → None[source]: Perform the action.

async count(category: str, tag_filter: str | dict = None) → int[source]: Perform the action.

async fetch(category: str, name: str, *, for_update: bool = False) → Entry | None[source]: Perform the action.

async fetch_all(category: str, tag_filter: str | dict = None, limit: int = None, for_update: bool = False, order_by: str | None = None, descending: bool = False) → EntryList[source]: Perform the action.

property handle: Get a unique identifier for the session.

async insert(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None) → None[source]: Perform the action.

property is_transaction: bool: Check if the session is a transaction.

async remove(category: str, name: str) → None[source]: Perform the action.

async remove_all(category: str, tag_filter: str | dict = None) → int[source]: Perform the action.

async replace(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None) → None[source]: Perform the action.

async rollback() → None[source]: Perform the action.

class acapy_agent.database_manager.dbstore.Scan(store: DBStore, profile: str | None, category: str | bytes, tag_filter: str | dict = None, offset: int = None, limit: int = None, order_by: str | None = None, descending: bool = False)[source]

Bases: AsyncIterator

Async iterator for database scanning.

class acapy_agent.database_manager.dbstore.ScanKeyset(store: DBStore, profile: str | None, category: str | bytes, tag_filter: str | dict = None, last_id: int | None = None, limit: int = None, order_by: str | None = None, descending: bool = False)[source]

Bases: AsyncIterator

Keyset-based scan iterator.

async fetch_all() → Sequence[Entry][source]: Perform the action.

acapy_agent.database_manager.dbstore.register_backend(db_type: str, backend: DatabaseBackend)[source]: Register a backend for a given database type.

acapy_agent.database_manager.error module

Error classes for database management.

exception acapy_agent.database_manager.error.DBStoreError(code: DBStoreErrorCode, message: str, extra: str = None)[source]

Bases: Exception

Database store error.

class acapy_agent.database_manager.error.DBStoreErrorCode(*values)[source]

Bases: IntEnum

Error codes for database store operations.

BACKEND = 1

BUSY = 2

CUSTOM = 100

DUPLICATE = 3

ENCRYPTION = 4

INPUT = 5

NOT_FOUND = 6

SUCCESS = 0

UNEXPECTED = 7

UNSUPPORTED = 8

WRAPPER = 99

acapy_agent.database_manager.interfaces module

Module docstring.

class acapy_agent.database_manager.interfaces.AbstractDatabaseSession[source]

Bases: ABC

Abstract base class for database session implementations.

abstractmethod async close()[source]: Close the session.

abstractmethod async commit()[source]: Commit the transaction.

abstractmethod async count(category: str, tag_filter: str | dict = None) → int[source]: Count entries.

abstractmethod async fetch(category: str, name: str, for_update: bool = False) → Entry | None[source]: Fetch a single entry.

abstractmethod async fetch_all(category: str, tag_filter: str | dict = None, limit: int = None, for_update: bool = False) → Sequence[Entry][source]: Fetch all matching entries.

abstractmethod async insert(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None)[source]: Insert a new entry.

abstractmethod async remove(category: str, name: str)[source]: Remove an entry.

abstractmethod async remove_all(category: str, tag_filter: str | dict = None) → int[source]: Remove all matching entries.

abstractmethod async replace(category: str, name: str, value: str | bytes = None, tags: dict = None, expiry_ms: int = None, value_json=None)[source]: Replace an existing entry.

abstractmethod async rollback()[source]: Rollback the transaction.

class acapy_agent.database_manager.interfaces.AbstractDatabaseStore[source]

Bases: ABC

Abstract base class for database store implementations.

abstractmethod async close(remove: bool = False) → bool[source]: Close the database store.

abstractmethod async create_profile(name: str = None) → str[source]: Create a new profile.

abstractmethod async get_profile_name() → str[source]: Get the profile name.

abstractmethod async rekey(key_method: str = None, pass_key: str = None)[source]: Re-key the database.

abstractmethod async remove_profile(name: str) → bool[source]: Remove a profile.

abstractmethod scan(profile: str | None, category: str, tag_filter: str | dict = None, offset: int = None, limit: int = None) → Generator[Entry, None, None][source]: Scan database entries.

abstractmethod session(profile: str = None, release_number: str = 'release_0') → AbstractDatabaseSession[source]

Create a new database session with the specified release number.

Parameters:

profile – Optional profile name.
release_number – Release number to use (e.g., ‘release_0’). Defaults to ‘release_0’.

Returns:

The session instance.

Return type:

AbstractDatabaseSession

abstractmethod transaction(profile: str = None, release_number: str = 'release_0') → AbstractDatabaseSession[source]

Create a new database transaction with the specified release number.

Parameters:

profile – Optional profile name.
release_number – Release number to use (e.g., ‘release_0’). Defaults to ‘release_0’.

Returns:

The transaction instance.

Return type:

AbstractDatabaseSession

class acapy_agent.database_manager.interfaces.DatabaseBackend[source]

Bases: ABC

Abstract base class for database backends.

abstractmethod open(uri, key_method, pass_key, profile, release_number: str = 'release_0')[source]

Open an existing database with the specified release number.

Parameters:

uri – The database URI.
key_method – Optional key method for encryption.
pass_key – Optional encryption key.
profile – Optional profile name.
release_number – Release number to use (e.g., ‘release_0’). Defaults to ‘release_0’.

abstractmethod provision(uri, key_method, pass_key, profile, recreate, release_number: str = 'release_0')[source]

Provision a new database with the specified release number.

Parameters:

uri – The database URI.
key_method – Optional key method for encryption.
pass_key – Optional encryption key.
profile – Optional profile name.
recreate – If True, recreate the database.
release_number – Release number to use (e.g., ‘release_0’). Defaults to ‘release_0’.

abstractmethod remove(uri, release_number: str = 'release_0')[source]

Remove the database.

Parameters:

uri – The database URI.
release_number – Release number to use (e.g., ‘release_0’). Defaults to ‘release_0’.

abstractmethod translate_error(exception)[source]: Translate backend-specific exceptions to DBStoreError.

acapy_agent.database_manager.key module

Handling of Key instances.

class acapy_agent.database_manager.key.Key(handle: Any)[source]

Bases: object

An active key or keypair instance.

aead_decrypt(ciphertext: bytes = None, *, nonce: bytes | None = None, aad: bytes | None = None) → bytes[source]: Return placeholder decrypted data.

aead_encrypt(plaintext: bytes = None, *, nonce: bytes | None = None, aad: bytes | None = None) → str[source]: Return a placeholder for encrypted data.

aead_params() → str[source]: Return a placeholder for AEAD parameters.

aead_random_nonce() → bytes[source]: Return placeholder nonce since bindings is unavailable.

property algorithm: KeyAlg: Return a placeholder algorithm since bindings is unavailable.

convert_key(alg: str | KeyAlg) → Key[source]: Raise an error as key conversion requires bindings.

property ephemeral: bool: Return a placeholder ephemeral flag since bindings is unavailable.

classmethod from_jwk(jwk: dict | str | bytes) → Key[source]: Raise an error as JWK-based key creation requires bindings.

classmethod from_public_bytes(alg: str | KeyAlg, public: bytes) → Key[source]: Raise an error as public-based key creation requires bindings.

classmethod from_secret_bytes(alg: str | KeyAlg, secret: bytes) → Key[source]: Raise an error as secret-based key creation requires bindings.

classmethod from_seed(alg: str | KeyAlg, seed: str | bytes, *, method: str | SeedMethod = None) → Key[source]: Raise an error as seed-based key creation requires bindings.

classmethod generate(alg: str | KeyAlg, *, ephemeral: bool = False) → Key[source]: Raise an error as key generation requires bindings.

get_jwk_public() → str[source]: Return placeholder public JWK since bindings is unavailable.

get_jwk_secret() → bytes[source]: Return placeholder secret JWK since bindings is unavailable.

get_jwk_thumbprint() → str[source]: Return placeholder JWK thumbprint since bindings is unavailable.

get_public_bytes() → bytes[source]: Return placeholder public bytes since bindings is unavailable.

get_secret_bytes() → bytes[source]: Return placeholder secret bytes since bindings is unavailable.

property handle: Any: Accessor for the key handle.

key_exchange(alg: str | KeyAlg, pk: Key) → Key[source]: Raise an error as key exchange requires bindings.

sign_message(message: str | bytes, sig_type: str = None) → bytes[source]: Raise an error as signing requires bindings.

unwrap_key() → Key[source]: Return a placeholder Key instance.

verify_signature(message: str | bytes, signature: bytes, sig_type: str = None) → bool[source]: Raise an error as verification requires bindings.

wrap_key(other_key=None) → str[source]: Return a placeholder for wrapped key.