acapy_agent.messaging.models package

Common code for messaging models.

Submodules

acapy_agent.messaging.models.base module

Base classes for Models and Schemas.

class acapy_agent.messaging.models.base.BaseModel[source]

Bases: ABC

Base model that provides convenience methods.

class Meta[source]

Bases: object

BaseModel meta data.

schema_class = None
property Schema: Type[BaseModelSchema]

Accessor for the model’s schema class.

Returns:

The schema class

classmethod deserialize(obj) → ModelType[source]
classmethod deserialize(obj, *, unknown: str | None = None) → ModelType
classmethod deserialize(obj, *, none2none: typing_extensions.Literal.False, unknown: str | None = None) → ModelType
classmethod deserialize(obj, *, none2none: typing_extensions.Literal.True, unknown: str | None = None) → ModelType | None

Convert from JSON representation to a model instance.

Parameters:

  • obj – The dict to load into a model instance

  • unknown – Behaviour for unknown attributes

  • none2none – Deserialize None to None

Returns:

A model instance for this data

classmethod from_json(json_repr: str | bytes, unknown: str | None = None)[source]

Parse a JSON string into a model instance.

Parameters:

  • json_repr – JSON string

  • unknown – Behaviour for unknown attributes

Returns:

A model instance representation of this JSON

classmethod serde(obj: BaseModel | Mapping | None) → SerDe | None[source]

Return serialized, deserialized representations of input object.

serialize(*, as_string: typing_extensions.Literal.True, unknown: str | None = None) → str[source]
serialize(*, unknown: str | None = None) → dict

Create a JSON-compatible dict representation of the model instance.

Parameters:

  • as_string – Return a string of JSON instead of a dict

  • unknown – Behaviour for unknown attributes

Returns:

A dict representation of this model, or a JSON string if as_string is True

to_json(unknown: str | None = None) → str[source]

Create a JSON representation of the model instance.

Returns:

A JSON representation of this message

validate(unknown: str | None = None)[source]

Validate a constructed model.

exception acapy_agent.messaging.models.base.BaseModelError(*args, error_code: str | None = None, **kwargs)[source]

Bases: BaseError

Base exception class for base model errors.

class acapy_agent.messaging.models.base.BaseModelSchema(*args: Any, **kwargs: Any)[source]

Bases: Schema

BaseModel schema.

class Meta[source]

Bases: object

BaseModelSchema metadata.

model_class = None
skip_values = [None]
property Model: type

Accessor for the schema’s model class.

Returns:

The model class

make_model(data: dict, **kwargs)

Return model instance after loading.

Returns:

A model instance

remove_skipped_values(data, **kwargs)

Remove values that are are marked to skip.

Returns:

Returns this modified data

skip_dump_only(data, **kwargs)

Skip fields that are only expected during serialization.

Parameters:

  • data – The incoming data to clean

  • kwargs – Arbitrary keyword arguments

Returns:

The modified data

class acapy_agent.messaging.models.base.SerDe(ser, de)

Bases: tuple

de

Alias for field number 1

ser

Alias for field number 0

acapy_agent.messaging.models.base.resolve_class(the_cls, relative_cls: type | None = None) → type[source]

Resolve a class.

Parameters:

  • the_cls – The class to resolve

  • relative_cls – Relative class to resolve from

Returns:

The resolved class

Raises:

ClassNotFoundError – If the class could not be loaded

acapy_agent.messaging.models.base.resolve_meta_property(obj, prop_name: str, defval=None)[source]

Resolve a meta property.

Parameters:

  • obj – The object to resolve the property from

  • prop_name – The property to resolve

  • defval – The default value

Returns:

The meta property

acapy_agent.messaging.models.base_record module

Classes for BaseStorage-based record management.

class acapy_agent.messaging.models.base_record.BaseExchangeRecord(id: str | None = None, state: str | None = None, *, trace: bool = False, **kwargs)[source]

Bases: BaseRecord

Represents a base record with event tracing capability.

class acapy_agent.messaging.models.base_record.BaseExchangeSchema(*args: Any, **kwargs: Any)[source]

Bases: BaseRecordSchema

Base schema for exchange records.

class Meta[source]

Bases: object

BaseExchangeSchema metadata.

model_class

alias of BaseExchangeRecord

class acapy_agent.messaging.models.base_record.BaseRecord(id: str | None = None, state: str | None = None, *, created_at: str | datetime | None = None, updated_at: str | datetime | None = None, new_with_id: bool = False)[source]

Bases: BaseModel

Represents a single storage record.

DEFAULT_CACHE_TTL = 60
EVENT_NAMESPACE: str = 'acapy::record'
LOG_STATE_FLAG = None
class Meta[source]

Bases: object

BaseRecord metadata.

RECORD_ID_NAME = 'id'
RECORD_TOPIC: str | None = None
RECORD_TYPE = None
STATE_DELETED = 'deleted'
TAG_NAMES = {'state'}
async classmethod clear_cached_key(session: ProfileSession, cache_key: str)[source]

Shortcut method to clear a cached key value, if any.

Parameters:

  • session – The profile session to use

  • cache_key – The unique cache identifier

async delete_record(session: ProfileSession)[source]

Remove the stored record.

Parameters:

session – The profile session to use

async emit_event(session: ProfileSession, payload: Any | None = None)[source]

Emit an event.

Parameters:

  • session – The profile session to use

  • payload – The event payload

classmethod from_storage(record_id: str, record: Mapping[str, Any])[source]

Initialize a record from its stored representation.

Parameters:

  • record_id – The unique record identifier

  • record – The stored representation

classmethod get_attributes_by_prefix(prefix: str, walk_mro: bool = True)[source]

List all values for attributes with common prefix.

Parameters:

  • prefix – Common prefix to look for

  • walk_mro – Walk MRO to find attributes inherited from superclasses

async classmethod get_cached_key(session: ProfileSession, cache_key: str)[source]

Shortcut method to fetch a cached key value.

Parameters:

  • session – The profile session to use

  • cache_key – The unique cache identifier

classmethod get_tag_map() → Mapping[str, str][source]

Accessor for the set of defined tags.

classmethod log_state(msg: str, params: dict | None = None, settings: BaseSettings | None = None, override: bool = False)[source]

Print a message with increased visibility (for testing).

async post_save(session: ProfileSession, new_record: bool, last_state: str | None, event: bool | None = None)[source]

Perform post-save actions.

Parameters:

  • session – The profile session to use

  • new_record – Flag indicating if the record was just created

  • last_state – The previous state value

  • event – Flag to override whether the event is sent

classmethod prefix_tag_filter(tag_filter: dict)[source]

Prefix unencrypted tags used in the tag filter.

async classmethod query(session: ProfileSession, tag_filter: dict | None = None, *, limit: int | None = None, offset: int | None = None, order_by: str | None = None, descending: bool = False, post_filter_positive: dict | None = None, post_filter_negative: dict | None = None, alt: bool = False) → Sequence[RecordType][source]

Query stored records.

Parameters:

  • session – The profile session to use

  • tag_filter – An optional dictionary of tag filter clauses

  • limit – The maximum number of records to retrieve

  • offset – The offset to start retrieving records from

  • order_by – An optional field by which to order the records.

  • descending – Whether to order the records in descending order.

  • post_filter_positive – Additional value filters to apply matching positively

  • post_filter_negative – Additional value filters to apply matching negatively

  • alt – set to match any (positive=True) value or miss all (positive=False) values in post_filter

property record_tags: dict

Accessor to define implementation-specific tags.

property record_value: dict

Accessor to define custom properties for the JSON record value.

async classmethod retrieve_by_id(session: ProfileSession, record_id: str, *, for_update=False) → RecordType[source]

Retrieve a stored record by ID.

Parameters:

  • session – The profile session to use

  • record_id – The ID of the record to find

  • for_update – Whether to lock the record for update

async classmethod retrieve_by_tag_filter(session: ProfileSession, tag_filter: dict, post_filter: dict | None = None, *, for_update=False) → RecordType[source]

Retrieve a record by tag filter.

Parameters:

  • cls – The record class

  • session – The profile session to use

  • tag_filter – The filter dictionary to apply

  • post_filter – Additional value filters to apply matching positively, with sequence values specifying alternatives to match (hit any)

  • for_update – Whether to lock the record for update

async save(session: ProfileSession, *, reason: str | None = None, log_params: Mapping[str, Any] = None, log_override: bool = False, event: bool | None = None) → str[source]

Persist the record to storage.

Parameters:

  • session – The profile session to use

  • reason – A reason to add to the log

  • log_params – Additional parameters to log

  • log_override – Override configured logging regimen, print to stderr instead

  • event – Flag to override whether the event is sent

async classmethod set_cached_key(session: ProfileSession, cache_key: str, value: Any, ttl=None)[source]

Shortcut method to set a cached key value.

Parameters:

  • session – The profile session to use

  • cache_key – The unique cache identifier

  • value – The value to cache

  • ttl – The cache ttl

property storage_record: StorageRecord

Accessor for a StorageRecord representing this record.

classmethod strip_tag_prefix(tags: dict)[source]

Strip tilde from unencrypted tag names.

property tags: dict

Accessor for the record tags generated for this record.

property value: dict

Accessor for the JSON record value generated for this record.

class acapy_agent.messaging.models.base_record.BaseRecordSchema(*args: Any, **kwargs: Any)[source]

Bases: BaseModelSchema

Schema to allow serialization/deserialization of base records.

class Meta[source]

Bases: object

BaseRecordSchema metadata.

model_class = None
acapy_agent.messaging.models.base_record.match_post_filter(record: dict, post_filter: dict, positive: bool = True, alt: bool = False) → bool[source]

Determine if a record value matches the post-filter.

Parameters:

  • record – record to check

  • post_filter – filter to apply (empty or None filter matches everything)

  • positive – whether matching all filter criteria positively or negatively

  • alt – set to match any (positive=True) value or miss all (positive=False) values in post_filter

acapy_agent.messaging.models.openapi module

Base class for OpenAPI artifact schema.

class acapy_agent.messaging.models.openapi.OpenAPISchema(*args: Any, **kwargs: Any)[source]

Bases: Schema

Schema for OpenAPI artifacts: excluding unknown fields, not raising exception.

class Meta[source]

Bases: object

OpenAPISchema metadata.

model_class = None

acapy_agent.messaging.models.paginated_query module

Class for paginated query parameters.

class acapy_agent.messaging.models.paginated_query.PaginatedQuerySchema(*args: Any, **kwargs: Any)[source]

Bases: OpenAPISchema

Parameters for paginated queries.

acapy_agent.messaging.models.paginated_query.get_paginated_query_params(request: aiohttp.web.BaseRequest) → Tuple[int, int, str, bool][source]

Read the limit, offset, order_by, and descending query parameters from a request.

Parameters:

request – aiohttp request object.

Returns:

  • limit (int): The number of results to return, defaulting to DEFAULT_PAGE_SIZE.

  • offset (int): The offset for pagination, defaulting to 0.

  • order_by (str): The field by which to order results, defaulting to “id”.

  • descending (bool): Order results in descending order; defaults to False.

Return type:

A tuple containing