aries_cloudagent.messaging.models package

Submodules

aries_cloudagent.messaging.models.base module

Base classes for Models and Schemas.

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

Bases: abc.ABC

Base model that provides convenience methods.

class Meta[source]

Bases: object

BaseModel meta data.

schema_class = None
Schema

Accessor for the model’s schema class.

Returns:The schema class
classmethod deserialize(obj)[source]

Convert from JSON representation to a model instance.

Parameters:obj – The dict to load into a model instance
Returns:A model instance for this data
classmethod from_json(json_repr: Union[str, bytes])[source]

Parse a JSON string into a model instance.

Parameters:json_repr – JSON string
Returns:A model instance representation of this JSON
serialize(as_string=False) → dict[source]

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() → str[source]

Create a JSON representation of the model instance.

Returns:A JSON representation of this message
exception aries_cloudagent.messaging.models.base.BaseModelError(*args, error_code: str = None, **kwargs)[source]

Bases: aries_cloudagent.core.error.BaseError

Base exception class for base model errors.

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

Bases: sphinx.ext.autodoc.importer._MockObject

BaseModel schema.

class Meta[source]

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)[source]

Return model instance after loading.

Returns:A model instance
remove_skipped_values(data, **kwargs)[source]

Remove values that are are marked to skip.

Returns:Returns this modified data
skip_dump_only(data, **kwargs)[source]

Skip fields that are only expected during serialization.

Parameters:data – The incoming data to clean
Returns:The modified data
aries_cloudagent.messaging.models.base.resolve_class(the_cls, relative_cls: type = None)[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

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

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)[source]

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)[source]

Bases: aries_cloudagent.messaging.models.base_record.BaseRecordSchema

Base schema for exchange records.

class Meta[source]

Bases: object

BaseExchangeSchema metadata.

model_class

alias of BaseExchangeRecord

trace = <fields.Boolean(default=False, attribute=None, validate=None, required=False, load_only=False, dump_only=False, missing=<marshmallow.missing>, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid boolean.'})>
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)[source]

Bases: aries_cloudagent.messaging.models.base.BaseModel

Represents a single storage record.

CACHE_ENABLED = False
CACHE_TTL = 60
LOG_STATE_FLAG = None
class Meta[source]

Bases: object

BaseRecord metadata.

RECORD_ID_NAME = 'id'
RECORD_TYPE = None
TAG_NAMES = {'state'}
WEBHOOK_TOPIC = None
classmethod cache_key(record_id: str, record_type: str = None)[source]

Assemble a cache key.

Parameters:
  • record_id – The record identifier
  • The cache type identifier, defaulting to RECORD_TYPE (record_type) –
clear_cached(context: aries_cloudagent.config.injection_context.InjectionContext)[source]

Clear the cached value of this record, if any.

classmethod clear_cached_key(context: aries_cloudagent.config.injection_context.InjectionContext, cache_key: str)[source]

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

Parameters:
  • context – The injection context to use
  • cache_key – The unique cache identifier
delete_record(context: aries_cloudagent.config.injection_context.InjectionContext)[source]

Remove the stored record.

Parameters:context – The injection context to use
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_cached_key(context: aries_cloudagent.config.injection_context.InjectionContext, cache_key: str)[source]

Shortcut method to fetch a cached key value.

Parameters:
  • context – The injection context 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(context: aries_cloudagent.config.injection_context.InjectionContext, msg: str, params: dict = None, override: bool = False)[source]

Print a message with increased visibility (for testing).

post_save(context: aries_cloudagent.config.injection_context.InjectionContext, new_record: bool, last_state: str, webhook: bool = None)[source]

Perform post-save actions.

Parameters:
  • context – The injection context to use
  • new_record – Flag indicating if the record was just created
  • last_state – The previous state value
  • webhook – Adjust whether the webhook is called
classmethod prefix_tag_filter(tag_filter: dict)[source]

Prefix unencrypted tags used in the tag filter.

classmethod query(context: aries_cloudagent.config.injection_context.InjectionContext, tag_filter: dict = None, post_filter_positive: dict = None, post_filter_negative: dict = None) → Sequence[aries_cloudagent.messaging.models.base_record.BaseRecord][source]

Query stored records.

Parameters:
  • context – The injection context 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
record_tags

Accessor to define implementation-specific tags.

record_value

Accessor to define custom properties for the JSON record value.

classmethod retrieve_by_id(context: aries_cloudagent.config.injection_context.InjectionContext, record_id: str, cached: bool = True) → aries_cloudagent.messaging.models.base_record.BaseRecord[source]

Retrieve a stored record by ID.

Parameters:
  • context – The injection context to use
  • record_id – The ID of the record to find
  • cached – Whether to check the cache for this record
classmethod retrieve_by_tag_filter(context: aries_cloudagent.config.injection_context.InjectionContext, tag_filter: dict, post_filter: dict = None) → aries_cloudagent.messaging.models.base_record.BaseRecord[source]

Retrieve a record by tag filter.

Parameters:
  • context – The injection context to use
  • tag_filter – The filter dictionary to apply
  • post_filter – Additional value filters to apply after retrieval
save(context: aries_cloudagent.config.injection_context.InjectionContext, *, reason: str = None, log_params: Mapping[str, Any] = None, log_override: bool = False, webhook: bool = None) → str[source]

Persist the record to storage.

Parameters:
  • context – The injection context to use
  • reason – A reason to add to the log
  • log_params – Additional parameters to log
  • webhook – Flag to override whether the webhook is sent
send_webhook(context: aries_cloudagent.config.injection_context.InjectionContext, payload: Any, topic: str = None)[source]

Send a standard webhook.

Parameters:
  • context – The injection context to use
  • payload – The webhook payload
  • topic – The webhook topic, defaulting to WEBHOOK_TOPIC
classmethod set_cached_key(context: aries_cloudagent.config.injection_context.InjectionContext, cache_key: str, value: Any, ttl=None)[source]

Shortcut method to set a cached key value.

Parameters:
  • context – The injection context 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)[source]

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.

webhook_payload

Return a JSON-serialized version of the record for the webhook.

webhook_topic

Return the webhook topic value.

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

Bases: aries_cloudagent.messaging.models.base.BaseModelSchema

Schema to allow serialization/deserialization of base records.

class Meta[source]

Bases: object

BaseRecordSchema metadata.

created_at = <fields.String(default=<marshmallow.missing>, attribute=None, validate=<aries_cloudagent.messaging.valid.IndyISO8601DateTime object>, required=False, load_only=False, dump_only=False, missing=<marshmallow.missing>, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid string.', 'invalid_utf8': 'Not a valid utf-8 string.'})>
state = <fields.String(default=<marshmallow.missing>, attribute=None, validate=None, required=False, load_only=False, dump_only=False, missing=<marshmallow.missing>, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid string.', 'invalid_utf8': 'Not a valid utf-8 string.'})>
updated_at = <fields.String(default=<marshmallow.missing>, attribute=None, validate=<aries_cloudagent.messaging.valid.IndyISO8601DateTime object>, required=False, load_only=False, dump_only=False, missing=<marshmallow.missing>, allow_none=False, error_messages={'required': 'Missing data for required field.', 'null': 'Field may not be null.', 'validator_failed': 'Invalid value.', 'invalid': 'Not a valid string.', 'invalid_utf8': 'Not a valid utf-8 string.'})>
aries_cloudagent.messaging.models.base_record.match_post_filter(record: dict, post_filter: dict, positive: bool = True) → 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