acapy_agent.database_manager.wql_normalized.encoders package

Submodules

acapy_agent.database_manager.wql_normalized.encoders.encoder_factory module

Module docstring.

acapy_agent.database_manager.wql_normalized.encoders.encoder_factory.get_encoder(db_type: str, enc_name, enc_value, normalized: bool = False, tags_table: str = 'items_tags')[source]

Returns an encoder object based on the database type.

Parameters:

db_type (str) – The type of database (e.g., ‘sqlite’, ‘postgresql’, ‘mongodb’, ‘mssql’).
enc_name (callable) – Function to encode tag names.
enc_value (callable) – Function to encode tag values.
normalized (bool) – Flag to indicate if the encoder should use normalized mode (default: False).
tags_table (str) – Name of the tags table for non-normalized mode (default: ‘items_tags’). Ignored in normalized mode.

Returns:

An instance of the appropriate encoder class.

Return type:

TagQueryEncoder

Raises:

ValueError – If the database type is not supported.

acapy_agent.database_manager.wql_normalized.encoders.postgres_encoder module

Module docstring.

class acapy_agent.database_manager.wql_normalized.encoders.postgres_encoder.PostgresTagEncoder(enc_name, enc_value, normalized: bool = False, table_alias: str = 't', tags_table: str = 'items_tags')[source]

Bases: TagQueryEncoder

PostgreSQL tag query encoder.

encode_conj(op: ConjunctionOp, subqueries: List[TagQuery], negate: bool)[source]: Encode a conjunction operation.

encode_conj_clause(op: ConjunctionOp, clauses: List[str]) → str[source]: Encode a conjunction clause (AND/OR) for PostgreSQL.

encode_exist(names: List[TagName], negate: bool)[source]: Encode an existence check.

encode_exist_clause(enc_name: str, negate: bool) → str[source]

Encode an ‘EXISTS’ clause for tag or column existence in PostgreSQL.

Uses %s placeholders for psycopg 3.2.9 compatibility.

encode_in(name: TagName, values: List[str], negate: bool)[source]: Encode an IN operation.

encode_in_clause(enc_name: str, enc_values: List[str], negate: bool) → str[source]

Encode an ‘IN’ clause for multiple values in PostgreSQL.

Uses %s placeholders for psycopg 3.2.9 compatibility.

encode_name(name: TagName) → str[source]: Encode the tag name using the provided enc_name function.

encode_op(op: CompareOp, name: TagName, value: str, negate: bool)[source]: Encode a comparison operation.

encode_op_clause(op: CompareOp, enc_name: str, enc_value: str, negate: bool) → str[source]

Encode a comparison operation clause for PostgreSQL.

In normalized mode, generates direct column comparisons (e.g., “t.column = %s”). In non-normalized mode, generates subqueries using the configured tags table (e.g., “i.id IN (SELECT item_id FROM tags_table …)”). Uses %s placeholders for psycopg 3.2.9 compatibility.

encode_query(query: TagQuery, negate: bool = False, top_level: bool = True) → Tuple[str, List[str]] | str[source]

Encode the query and reset arguments list only at top level.

Parameters:

query (TagQuery) – The query to encode.
negate (bool) – Whether to negate the query.
top_level (bool) – Whether this is a top-level query.

Returns:

SQL clause and list of parameters for: top-level queries, or SQL clause string for subqueries.

Return type:

Tuple[str, List[str]] | str

encode_value(value: str) → str[source]: Encode the tag value using the provided enc_value function.

acapy_agent.database_manager.wql_normalized.encoders.sqlite_encoder module

Module docstring.

class acapy_agent.database_manager.wql_normalized.encoders.sqlite_encoder.SqliteTagEncoder(enc_name, enc_value, normalized: bool = False, table_alias: str = 't', tags_table: str = 'items_tags')[source]

Bases: TagQueryEncoder

Encoder for generating SQLite-compatible SQL queries from TagQuery objects.

Uses ‘?’ placeholders for parameters. Supports both normalized and non-normalized modes with a configurable tags table for non-normalized mode.

encode_conj(op: ConjunctionOp, subqueries: List[TagQuery], negate: bool)[source]: Encode a conjunction (AND/OR) of subqueries.

encode_conj_clause(op: ConjunctionOp, clauses: List[str]) → str[source]: Encode a conjunction clause (AND/OR) for SQLite.

encode_exist(names: List[TagName], negate: bool)[source]: Encode an EXISTS operation for tag names.

encode_exist_clause(enc_name: str, negate: bool) → str[source]: Encode an ‘EXISTS’ clause for tag or column existence in SQLite.

encode_in(name: TagName, values: List[str], negate: bool)[source]: Encode an IN operation with multiple values.

encode_in_clause(enc_name: str, enc_values: List[str], negate: bool) → str[source]: Encode an ‘IN’ clause for multiple values in SQLite.

encode_name(name: TagName) → str[source]: Encode the tag name using the provided enc_name function.

encode_op(op: CompareOp, name: TagName, value: str, negate: bool)[source]: Encode a comparison operation.

encode_op_clause(op: CompareOp, enc_name: str, enc_value: str, negate: bool) → str[source]

Encode a comparison operation clause for SQLite.

In normalized mode, generates direct column comparisons (e.g., “t.column = ?”). In non-normalized mode, generates subqueries using the configured tags table (e.g., “i.id IN (SELECT item_id FROM tags_table …)”).

encode_query(query: TagQuery, negate: bool = False, top_level: bool = True) → str[source]: Encode the query and reset arguments list only at top level.

encode_value(value: str) → str[source]: Encode the tag value using the provided enc_value function.