acapy_agent.database_manager.wql_normalized package

Subpackages

acapy_agent.database_manager.wql_normalized.encoders package

Submodules

acapy_agent.database_manager.wql_normalized.query module

Askar WQL (Wallet Query Language) parsing and optimization.

class acapy_agent.database_manager.wql_normalized.query.AndQuery(subqueries: List[Query])[source]

Bases: Query

Logical AND of multiple clauses.

map(key_func, value_func)[source]: Perform the action.

optimise() → Query | None[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.EqQuery(key: str, value: str)[source]

Bases: Query

Equality comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.ExistQuery(keys: List[str])[source]

Bases: Query

Match any non-null field value of the given field names.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.GtQuery(key: str, value: str)[source]

Bases: Query

Greater-than comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.GteQuery(key: str, value: str)[source]

Bases: Query

Greater-than-or-equal comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.InQuery(key: str, values: List[str])[source]

Bases: Query

Match one of multiple field values in a set.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.LikeQuery(key: str, value: str)[source]

Bases: Query

SQL ‘LIKE’-compatible string comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.LtQuery(key: str, value: str)[source]

Bases: Query

Less-than comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.LteQuery(key: str, value: str)[source]

Bases: Query

Less-than-or-equal comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.NeqQuery(key: str, value: str)[source]

Bases: Query

Inequality comparison for a field value.

map(key_func, value_func)[source]: Perform the action.

optimise()[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.NotQuery(subquery: Query)[source]

Bases: Query

Negation of a clause.

map(key_func, value_func)[source]: Perform the action.

optimise() → Query | None[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.OrQuery(subqueries: List[Query])[source]

Bases: Query

Logical OR of multiple clauses.

map(key_func, value_func)[source]: Perform the action.

optimise() → Query | None[source]: Perform the action.

to_dict()[source]: Perform the action.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.query.Query[source]

Bases: object

Base class for all query types.

map(key_func: Callable[[str], str], value_func: Callable[[str, str], str]) → Query[source]: Transform keys and values in the query.

map_names(key_func: Callable[[str], str]) → Query[source]: Transform only the keys in the query.

map_values(value_func: Callable[[str, str], str]) → Query[source]: Transform only the values in the query.

optimise() → Query | None[source]: Optimize the query by simplifying its structure.

to_dict() → dict[source]: Convert the query to a JSON-compatible dictionary.

to_sql(table_columns: Set[str] | None = None) → Tuple[str, List[str | int | float]][source]

Convert the query to an SQL condition and parameters.

Parameters:

table_columns (Optional[Set[str]]) – Set of valid column names for validation.

Returns:

SQL condition string and: list of parameters.

Return type:

Tuple[str, List[Union[str, int, float]]]

acapy_agent.database_manager.wql_normalized.query.parse_operator(key: str, value: dict | list | str | None) → Query | None[source]: Parse an operator from a key-value pair.

acapy_agent.database_manager.wql_normalized.query.parse_query(query_dict: dict) → Query[source]: Parse a dictionary into a Query object.

acapy_agent.database_manager.wql_normalized.query.parse_single_operator(op_name: str, key: str, value: dict | list | str | None) → Query[source]: Parse a single operator from a key-value pair.

acapy_agent.database_manager.wql_normalized.query.query_from_json(json_value: dict | list | str | None) → Query[source]: Parse a JSON value (dict or list) into a Query object.

acapy_agent.database_manager.wql_normalized.query.query_from_str(json_str: str) → Query[source]: Parse a JSON string into a Query object.

acapy_agent.database_manager.wql_normalized.query.query_to_str(query: Query) → str[source]: Convert a Query object to a JSON string.

acapy_agent.database_manager.wql_normalized.tags module

Module docstring.

class acapy_agent.database_manager.wql_normalized.tags.CompareOp(*values)[source]

Bases: Enum

Class description.

Eq = '='

Gt = '>'

Gte = '>='

Like = 'LIKE'

Lt = '<'

Lte = '<='

Neq = '!='

as_sql_str()[source]: Perform the action.

as_sql_str_for_prefix()[source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.tags.ConjunctionOp(*values)[source]

Bases: Enum

Class description.

And = ' AND '

Or = ' OR '

as_sql_str()[source]: Perform the action.

negate()[source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.tags.TagName(value)[source]

Bases: object

Represents a tag name.

to_string()[source]: Perform the action.

class acapy_agent.database_manager.wql_normalized.tags.TagQuery(variant: str, data: TagQuery | List[TagQuery] | TagName | str | List[str])[source]

Bases: object

Class description.

static and_(subqueries: List[TagQuery])[source]: Perform the action.

static eq(name: TagName, value: str)[source]: Perform the action.

static exist(names: List[TagName])[source]: Perform the action.

static gt(name: TagName, value: str)[source]: Perform the action.

static gte(name: TagName, value: str)[source]: Perform the action.

static in_(name: TagName, values: List[str])[source]: Perform the action.

static like(name: TagName, value: str)[source]: Perform the action.

static lt(name: TagName, value: str)[source]: Perform the action.

static lte(name: TagName, value: str)[source]: Perform the action.

static neq(name: TagName, value: str)[source]: Perform the action.

static not_(subquery: TagQuery)[source]: Perform the action.

static or_(subqueries: List[TagQuery])[source]: Perform the action.

to_sql(table_columns: set | None = None) → Tuple[str, list][source]

Convert the TagQuery to an SQL condition and parameters for normalized tables.

Parameters:: table_columns (Optional[set]) – Set of valid column names for validation.
Returns:: SQL condition string and list of parameters.
Return type:: Tuple[str, list]
Raises:: ValueError – If an invalid column name is used or an unsupported query type is encountered.

to_wql_dict()[source]: Convert the TagQuery to a WQL-compatible dictionary.

to_wql_str()[source]: Convert the TagQuery to a WQL JSON string.

class acapy_agent.database_manager.wql_normalized.tags.TagQueryEncoder[source]

Bases: ABC

Class description.

encode_conj(op: ConjunctionOp, subqueries: List[TagQuery], negate: bool)[source]: Perform the action.

abstractmethod encode_conj_clause(op: ConjunctionOp, clauses: List[str]) → str[source]: Perform the action.

encode_exist(names: List[TagName], negate: bool)[source]: Perform the action.

abstractmethod encode_exist_clause(enc_name: str, negate: bool) → str[source]: Perform the action.

encode_in(name: TagName, values: List[str], negate: bool)[source]: Perform the action.

abstractmethod encode_in_clause(enc_name: str, enc_values: List[str], negate: bool) → str[source]: Perform the action.

abstractmethod encode_name(name: TagName) → str[source]: Perform the action.

encode_op(op: CompareOp, name: TagName, value: str, negate: bool)[source]: Perform the action.

abstractmethod encode_op_clause(op: CompareOp, enc_name: str, enc_value: str, negate: bool) → str[source]: Perform the action.

encode_query(query: TagQuery, negate: bool = False) → str[source]: Perform the action.

abstractmethod encode_value(value: str) → str[source]: Perform the action.

acapy_agent.database_manager.wql_normalized.tags.query_to_tagquery(q)[source]

Convert a Query object from query.py to a TagQuery object from tags.py.

Strips ‘~’ from keys as it is no longer used to determine tag type. NOTE: this is for backward compatibility as the caller will continue to provide the ~ character for plaintext.