acapy_agent.database_manager.wql_nosql.encoders package

Submodules

acapy_agent.database_manager.wql_nosql.encoders.encoder_factory module

Module docstring.

acapy_agent.database_manager.wql_nosql.encoders.encoder_factory.get_encoder(db_type: str, enc_name, enc_value)[source]

Returns an encoder object based on the database type.

Parameters:

  • db_type (str) – The type of database (e.g., ‘sqlite’, ‘postgresql’, ‘mongodb’).

  • enc_name (callable) – Function to encode tag names.

  • enc_value (callable) – Function to encode tag values.

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_nosql.encoders.mongo_encoder module

Module docstring.

class acapy_agent.database_manager.wql_nosql.encoders.mongo_encoder.MongoTagEncoder(enc_name, enc_value)[source]

Bases: TagQueryEncoder

MongoDB query encoder for tag-based queries.

encode_conj_clause(op: ConjunctionOp, clauses: List[Dict]) → Dict[source]

Encode a conjunction (AND/OR) clause.

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

Encode an EXISTS clause.

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

Encode an IN clause for multiple values.

encode_name(name: TagName) → str[source]

Encode a tag name using the provided encoding function.

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

Encode a comparison operation clause with low branching.

encode_value(value: str) → Any[source]

Encode a tag value using the provided encoding function.

acapy_agent.database_manager.wql_nosql.encoders.sqlite_encoder module

Module docstring.

class acapy_agent.database_manager.wql_nosql.encoders.sqlite_encoder.SqliteTagEncoder(enc_name, enc_value)[source]

Bases: TagQueryEncoder

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

Uses ‘?’ placeholders for parameters.

encode_conj_clause(op: ConjunctionOp, clauses: List[str]) → str[source]

Encode a conjunction clause (AND/OR) for SQLite.

Parameters:

  • op (ConjunctionOp) – The conjunction operator.

  • clauses (List[str]) – List of SQL clause strings to combine.

Returns:

Combined SQL clause string.

Return type:

str

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

Encode an ‘EXISTS’ clause for tag existence in SQLite.

Parameters:

  • enc_name (bytes) – Encoded tag name.

  • negate (bool) – Whether to negate the existence check.

Returns:

SQL clause string.

Return type:

str

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

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

Parameters:

  • enc_name (bytes) – Encoded tag name.

  • enc_values (List[bytes]) – List of encoded tag values.

  • negate (bool) – Whether to use ‘NOT IN’ instead of ‘IN’.

Returns:

SQL clause string.

Return type:

str

encode_name(name: TagName) → bytes[source]

Encode the tag name using the provided enc_name function.

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

Encode a comparison operation clause for SQLite.

Parameters:

  • op (CompareOp) – The comparison operator.

  • enc_name (bytes) – Encoded tag name.

  • enc_value (bytes) – Encoded tag value.

  • negate (bool) – Whether to negate the clause.

Returns:

SQL clause string.

Return type:

str

encode_value(value: str) → bytes[source]

Encode the tag value using the provided enc_value function.