aries_cloudagent.messaging.models package

Common code for messaging models.

Submodules

aries_cloudagent.messaging.models.base module

Base classes for Models and Schemas.

class aries_cloudagent.messaging.models.base.BaseModel

Bases: abc.ABC

Base model that provides convenience methods.

class Meta

Bases: object

BaseModel meta data.

schema_class = None

Schema

Accessor for the model’s schema class.

Returns:The schema class

classmethod deserialize(obj, unknown: str = None, none2none: str = False)

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: Union[str, bytes], unknown: str = None)

Parse a JSON string into a model instance.

Parameters:json_repr – JSON string Returns:A model instance representation of this JSON

classmethod serde(obj: Union[BaseModel, Mapping[KT, VT_co]]) → aries_cloudagent.messaging.models.base.SerDe

Return serialized, deserialized representations of input object.

serialize(as_string=False, unknown: str = None) → dict

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

Parameters:as_string – Return a string of JSON instead of a dict Returns:A dict representation of this model, or a JSON string if as_string is True

to_json(unknown: str = None) → str

Create a JSON representation of the model instance.

Returns:A JSON representation of this message

validate(unknown: str = None)

Validate a constructed model.

exception aries_cloudagent.messaging.models.base.BaseModelError(*args, error_code: str = None, **kwargs)

Bases: aries_cloudagent.core.error.BaseError

Base exception class for base model errors.

class aries_cloudagent.messaging.models.base.BaseModelSchema(*args, **kwargs)

Bases: sphinx.ext.autodoc.importer._MockObject

BaseModel schema.

class Meta

Bases: object

BaseModelSchema metadata.

model_class = None

ordered = True

skip_values = [None]

Model

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 Returns:The modified data

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

Bases: tuple

de

Alias for field number 1

ser

Alias for field number 0

aries_cloudagent.messaging.models.base.resolve_class(the_cls, relative_cls: type = None)

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

aries_cloudagent.messaging.models.base.resolve_meta_property(obj, prop_name: str, defval=None)

Resolve a meta property.

Parameters:

• prop_name – The property to resolve
• defval – The default value

Returns:

The meta property

aries_cloudagent.messaging.models.base_record module

Classes for BaseStorage-based record management.

class aries_cloudagent.messaging.models.base_record.BaseExchangeRecord(id: str = None, state: str = None, *, trace: bool = False, **kwargs)

Bases: aries_cloudagent.messaging.models.base_record.BaseRecord

Represents a base record with event tracing capability.

class aries_cloudagent.messaging.models.base_record.BaseExchangeSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.base_record.BaseRecordSchema

Base schema for exchange records.

class Meta

Bases: object

BaseExchangeSchema metadata.

model_class

alias of BaseExchangeRecord

trace

Used by autodoc_mock_imports.

class aries_cloudagent.messaging.models.base_record.BaseRecord(id: str = None, state: str = None, *, created_at: Union[str, datetime.datetime] = None, updated_at: Union[str, datetime.datetime] = None)

Bases: aries_cloudagent.messaging.models.base.BaseModel

Represents a single storage record.

DEFAULT_CACHE_TTL = 60

EVENT_NAMESPACE = 'acapy::record'

LOG_STATE_FLAG = None

class Meta

Bases: object

BaseRecord metadata.

RECORD_ID_NAME = 'id'

RECORD_TOPIC = None

RECORD_TYPE = None

TAG_NAMES = {'state'}

classmethod clear_cached_key(session: aries_cloudagent.core.profile.ProfileSession, cache_key: str)

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

Parameters:

• session – The profile session to use
• cache_key – The unique cache identifier

delete_record(session: aries_cloudagent.core.profile.ProfileSession)

Remove the stored record.

Parameters:session – The profile session to use

emit_event(session: aries_cloudagent.core.profile.ProfileSession, payload: Any = None)

Emit an event.

Parameters:

• session – The profile session to use
• payload – The event payload

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

Initialize a record from its stored representation.

Parameters:

• record_id – The unique record identifier
• record – The stored representation

classmethod get_cached_key(session: aries_cloudagent.core.profile.ProfileSession, cache_key: str)

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]

Accessor for the set of defined tags.

classmethod log_state(msg: str, params: dict = None, settings: aries_cloudagent.config.base.BaseSettings = None, override: bool = False)

Print a message with increased visibility (for testing).

post_save(session: aries_cloudagent.core.profile.ProfileSession, new_record: bool, last_state: Optional[str], event: bool = None)

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)

Prefix unencrypted tags used in the tag filter.

classmethod query(session: aries_cloudagent.core.profile.ProfileSession, tag_filter: dict = None, *, post_filter_positive: dict = None, post_filter_negative: dict = None, alt: bool = False) → Sequence[aries_cloudagent.messaging.models.base_record.BaseRecord]

Query stored records.

Parameters:

• session – The profile session to use
• tag_filter – An optional dictionary of tag filter clauses
• 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

record_tags

Accessor to define implementation-specific tags.

record_value

Accessor to define custom properties for the JSON record value.

classmethod retrieve_by_id(session: aries_cloudagent.core.profile.ProfileSession, record_id: str) → aries_cloudagent.messaging.models.base_record.BaseRecord

Retrieve a stored record by ID.

Parameters:

• session – The profile session to use
• record_id – The ID of the record to find

classmethod retrieve_by_tag_filter(session: aries_cloudagent.core.profile.ProfileSession, tag_filter: dict, post_filter: dict = None) → aries_cloudagent.messaging.models.base_record.BaseRecord

Retrieve a record by tag filter.

Parameters:

• 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)

save(session: aries_cloudagent.core.profile.ProfileSession, *, reason: str = None, log_params: Mapping[str, Any] = None, log_override: bool = False, event: bool = None) → str

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
• override – Override configured logging regimen, print to stderr instead
• event – Flag to override whether the event is sent

classmethod set_cached_key(session: aries_cloudagent.core.profile.ProfileSession, cache_key: str, value: Any, ttl=None)

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

storage_record

Accessor for a StorageRecord representing this record.

classmethod strip_tag_prefix(tags: dict)

Strip tilde from unencrypted tag names.

tags

Accessor for the record tags generated for this record.

value

Accessor for the JSON record value generated for this record.

class aries_cloudagent.messaging.models.base_record.BaseRecordSchema(*args, **kwargs)

Bases: aries_cloudagent.messaging.models.base.BaseModelSchema

Schema to allow serialization/deserialization of base records.

class Meta

Bases: object

BaseRecordSchema metadata.

model_class = None

created_at

Used by autodoc_mock_imports.

state

Used by autodoc_mock_imports.

updated_at

Used by autodoc_mock_imports.

aries_cloudagent.messaging.models.base_record.match_post_filter(record: dict, post_filter: dict, positive: bool = True, alt: bool = False) → bool

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

aries_cloudagent.messaging.models.openapi module

Base class for OpenAPI artifact schema.

class aries_cloudagent.messaging.models.openapi.OpenAPISchema(*args, **kwargs)

Bases: sphinx.ext.autodoc.importer._MockObject

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

class Meta

Bases: object

OpenAPISchema metadata.

model_class = None

unknown

Used by autodoc_mock_imports.