Skip to content

aea.mail.base

Mail module abstract base classes.

URI Objects

class URI()

URI following RFC3986.

__init__

def __init__(uri_raw: str) -> None

Initialize the URI.

Must follow: https://tools.ietf.org/html/rfc3986.html

Arguments:

  • uri_raw: the raw form uri

scheme

@property
def scheme() -> str

Get the scheme.

netloc

@property
def netloc() -> str

Get the netloc.

path

@property
def path() -> str

Get the path.

params

@property
def params() -> str

Get the params.

query

@property
def query() -> str

Get the query.

fragment

@property
def fragment() -> str

Get the fragment.

username

@property
def username() -> Optional[str]

Get the username.

password

@property
def password() -> Optional[str]

Get the password.

host

@property
def host() -> Optional[str]

Get the host.

port

@property
def port() -> Optional[int]

Get the port.

__str__

def __str__() -> str

Get string representation.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

EnvelopeContext Objects

class EnvelopeContext()

Contains context information of an envelope.

__init__

def __init__(connection_id: Optional[PublicId] = None,
             uri: Optional[URI] = None) -> None

Initialize the envelope context.

Arguments:

  • connection_id: the connection id used for routing the outgoing envelope in the multiplexer.
  • uri: the URI sent with the envelope.

uri

@property
def uri() -> Optional[URI]

Get the URI.

connection_id

@property
def connection_id() -> Optional[PublicId]

Get the connection id to route the envelope.

connection_id

@connection_id.setter
def connection_id(connection_id: PublicId) -> None

Set the 'via' connection id.

__str__

def __str__() -> str

Get the string representation.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

AEAConnectionError Objects

class AEAConnectionError(Exception)

Exception class for connection errors.

Empty Objects

class Empty(Exception)

Exception for when the inbox is empty.

EnvelopeSerializer Objects

class EnvelopeSerializer(ABC)

Abstract class to specify the serialization layer for the envelope.

encode

@abstractmethod
def encode(envelope: "Envelope") -> bytes

Encode the envelope.

Arguments:

  • envelope: the envelope to encode

Returns:

the encoded envelope

decode

@abstractmethod
def decode(envelope_bytes: bytes) -> "Envelope"

Decode the envelope.

Arguments:

  • envelope_bytes: the encoded envelope

Returns:

the envelope

ProtobufEnvelopeSerializer Objects

class ProtobufEnvelopeSerializer(EnvelopeSerializer)

Envelope serializer using Protobuf.

encode

def encode(envelope: "Envelope") -> bytes

Encode the envelope.

Arguments:

  • envelope: the envelope to encode

Returns:

the encoded envelope

decode

def decode(envelope_bytes: bytes) -> "Envelope"

Decode the envelope.

The default serializer doesn't decode the message field.

Arguments:

  • envelope_bytes: the encoded envelope

Returns:

the envelope

Envelope Objects

class Envelope()

The top level message class for agent to agent communication.

__init__

def __init__(to: Address,
             sender: Address,
             message: Union[Message, bytes],
             context: Optional[EnvelopeContext] = None,
             protocol_specification_id: Optional[PublicId] = None) -> None

Initialize a Message object.

Arguments:

  • to: the address of the receiver.
  • sender: the address of the sender.
  • message: the protocol-specific message.
  • context: the optional envelope context.
  • protocol_specification_id: the protocol specification id (wire id).

to

@property
def to() -> Address

Get address of receiver.

to

@to.setter
def to(to: Address) -> None

Set address of receiver.

sender

@property
def sender() -> Address

Get address of sender.

sender

@sender.setter
def sender(sender: Address) -> None

Set address of sender.

protocol_specification_id

@property
def protocol_specification_id() -> PublicId

Get protocol_specification_id.

message

@property
def message() -> Union[Message, bytes]

Get the protocol-specific message.

message

@message.setter
def message(message: Union[Message, bytes]) -> None

Set the protocol-specific message.

message_bytes

@property
def message_bytes() -> bytes

Get the protocol-specific message.

context

@property
def context() -> Optional[EnvelopeContext]

Get the envelope context.

to_as_public_id

@property
def to_as_public_id() -> Optional[PublicId]

Get to as public id.

is_sender_public_id

@property
def is_sender_public_id() -> bool

Check if sender is a public id.

is_to_public_id

@property
def is_to_public_id() -> bool

Check if to is a public id.

is_component_to_component_message

@property
def is_component_to_component_message() -> bool

Whether or not the message contained is component to component.

__eq__

def __eq__(other: Any) -> bool

Compare with another object.

encode

def encode(serializer: Optional[EnvelopeSerializer] = None) -> bytes

Encode the envelope.

Arguments:

  • serializer: the serializer that implements the encoding procedure.

Returns:

the encoded envelope.

decode

@classmethod
def decode(cls,
           envelope_bytes: bytes,
           serializer: Optional[EnvelopeSerializer] = None) -> "Envelope"

Decode the envelope.

Arguments:

  • envelope_bytes: the bytes to be decoded.
  • serializer: the serializer that implements the decoding procedure.

Returns:

the decoded envelope.

__str__

def __str__() -> str

Get the string representation of an envelope.