acapy_agent.database_manager.databases.sqlite_normalized.handlers package

Module docstring.

Subpackages

Submodules

acapy_agent.database_manager.databases.sqlite_normalized.handlers.base_handler module

Module docstring.

class acapy_agent.database_manager.databases.sqlite_normalized.handlers.base_handler.BaseHandler(category: str)[source]

Bases: ABC

Abstract base class for handlers managing CRUD/query operations for a category.

abstractmethod count(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Count the number of entries matching the specified criteria.

abstractmethod fetch(cursor: Cursor, profile_id: int, category: str, name: str, tag_filter: str | dict, for_update: bool) → Entry | None[source]

Fetch a single entry by its name.

abstractmethod fetch_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict, limit: int, for_update: bool) → Sequence[Entry][source]

Fetch all entries matching the specified criteria.

abstractmethod get_sql_clause(tag_query: TagQuery) → Tuple[str, List[Any]][source]

Translate a TagQuery into an SQL clause and corresponding parameters.

abstractmethod insert(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Insert a new entry into the database.

abstractmethod remove(cursor: Cursor, profile_id: int, category: str, name: str) → None[source]

Remove an entry identified by its name.

abstractmethod remove_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Remove all entries matching the specified criteria.

abstractmethod replace(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Replace an existing entry in the database.

abstractmethod scan(cursor: Cursor, profile_id: int, category: str, tag_query: TagQuery | None, offset: int, limit: int) → Generator[Entry, None, None][source]

Scan the database for entries matching the criteria.

acapy_agent.database_manager.databases.sqlite_normalized.handlers.generic_handler module

Module docstring.

class acapy_agent.database_manager.databases.sqlite_normalized.handlers.generic_handler.GenericHandler(category: str = 'default', tags_table_name: str | None = None)[source]

Bases: BaseHandler

Handler for generic categories using items and a configurable tags table.

ALLOWED_ORDER_BY_COLUMNS = {'id', 'name', 'value'}
count(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Count entries matching criteria in the database.

fetch(cursor: Cursor, profile_id: int, category: str, name: str, tag_filter: str | dict, for_update: bool) → Entry | None[source]

Fetch a single entry from the database.

fetch_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict, limit: int, for_update: bool, order_by: str | None = None, descending: bool = False) → Sequence[Entry][source]

Fetch all entries matching criteria from the database.

get_sql_clause(tag_query: TagQuery) → Tuple[str, List[Any]][source]

Translate a TagQuery into an SQL clause and corresponding parameters.

insert(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Insert an entry into the database.

remove(cursor: Cursor, profile_id: int, category: str, name: str) → None[source]

Remove a single entry from the database.

remove_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Remove all entries matching criteria from the database.

replace(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Replace an existing entry in the database.

scan(cursor: Cursor, profile_id: int, category: str, tag_query: TagQuery | None, offset: int | None, limit: int | None, order_by: str | None = None, descending: bool = False) → Generator[Entry, None, None][source]

Scan entries with pagination from the database.

scan_keyset(cursor: Cursor, profile_id: int, category: str, tag_query: TagQuery | None, last_id: int | None, limit: int, order_by: str | None = None, descending: bool = False) → Generator[Entry, None, None][source]

Scan entries using keyset pagination from the database.

acapy_agent.database_manager.databases.sqlite_normalized.handlers.normalized_handler module

Module docstring.

class acapy_agent.database_manager.databases.sqlite_normalized.handlers.normalized_handler.NormalizedHandler(category: str, columns: List[str], table_name: str | None = None)[source]

Bases: BaseHandler

Handler for normalized categories using specific tables.

count(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Count the number of entries matching the specified criteria.

fetch(cursor: Cursor, profile_id: int, category: str, name: str, tag_filter: str | dict, for_update: bool) → Entry | None[source]

Fetch a single entry by its name.

fetch_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict, limit: int, for_update: bool, order_by: str | None = None, descending: bool = False) → Sequence[Entry][source]

Fetch all entries matching the specified criteria with ordering.

get_sql_clause(tag_query: TagQuery) → Tuple[str, List[Any]][source]

Translate a TagQuery into an SQL clause for the normalized table.

insert(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Insert an entry into the database.

remove(cursor: Cursor, profile_id: int, category: str, name: str) → None[source]

Remove an entry identified by its name.

remove_all(cursor: Cursor, profile_id: int, category: str, tag_filter: str | dict) → int[source]

Remove all entries matching the specified criteria.

replace(cursor: Cursor, profile_id: int, category: str, name: str, value: str | bytes, tags: dict, expiry_ms: int) → None[source]

Replace an existing entry in the database.

scan(cursor: Cursor, profile_id: int, category: str, tag_query: TagQuery | None, offset: int, limit: int, order_by: str | None = None, descending: bool = False) → Generator[Entry, None, None][source]

Scan the database for entries matching the criteria.

scan_keyset(cursor: Cursor, profile_id: int, category: str, tag_query: TagQuery | None, last_id: int | None, limit: int, order_by: str | None = None, descending: bool = False) → Generator[Entry, None, None][source]

Scan the database using keyset pagination based on the last seen item ID.

acapy_agent.database_manager.databases.sqlite_normalized.handlers.normalized_handler.deserialize_tags(tags: dict) → dict[source]

Deserialize tags, converting JSON strings and handling booleans.

acapy_agent.database_manager.databases.sqlite_normalized.handlers.normalized_handler.is_valid_json(value: str) → bool[source]

Check if a string is valid JSON.

acapy_agent.database_manager.databases.sqlite_normalized.handlers.normalized_handler.serialize_json_with_bool_strings(data: Any) → str[source]

Serialize data to JSON, converting booleans to string ‘true’/’false’ and ‘~’.