Skip to main content

agora.common.memory

ProtocolMemory Objects​

class ProtocolMemory()

Manages protocol-related memory, including registration and retrieval of protocols and their implementations.

__init__​

def __init__(storage: Storage, **kwargs)

Initializes ProtocolMemory with the given storage and additional keyword arguments.

Arguments:

  • storage Storage - The storage backend to use for managing protocols.
  • **kwargs - Additional keyword arguments, with their default values.

protocol_ids​

def protocol_ids() -> List[str]

Returns a list of registered protocol IDs.

Returns:

  • List[str] - A list containing all registered protocol identifiers.

is_known​

def is_known(protocol_id: str) -> bool

Checks if a protocol ID is known (registered).

Arguments:

  • protocol_id str - The protocol identifier to check.

Returns:

  • bool - True if the protocol is registered, False otherwise.

register_new_protocol​

def register_new_protocol(protocol_id: str,
                          protocol_document: str,
                          sources: List[str],
                          metadata: dict,
                          implementation: Optional[str] = None,
                          **kwargs)

Registers a new protocol with the specified details.

Arguments:

  • protocol_id str - The identifier of the new protocol.
  • protocol_document str - The document describing the protocol.
  • sources List[str] - A list of sources where the protocol is referenced.
  • metadata dict - Additional metadata related to the protocol.
  • implementation Optional[str], optional - The implementation code associated with the protocol. Defaults to None.
  • **kwargs - Additional keyword arguments to store with the protocol.

Raises:

  • StorageError - If the protocol is already registered.

get_protocol​

def get_protocol(protocol_id: str) -> Optional[Protocol]

Retrieves a Protocol object based on the protocol ID.

Arguments:

  • protocol_id str - The identifier of the protocol to retrieve.

Returns:

  • Optional[Protocol] - The Protocol object if found, else None.

get_implementation​

def get_implementation(protocol_id: str) -> Optional[str]

Gets the implementation associated with a given protocol ID.

Arguments:

  • protocol_id str - The identifier of the protocol.

Returns:

  • Optional[str] - The implementation code if available, else None.

register_implementation​

def register_implementation(protocol_id: str, implementation: str)

Registers an implementation for a specific protocol ID.

Arguments:

  • protocol_id str - The identifier of the protocol.
  • implementation str - The implementation code to associate with the protocol.

Raises:

  • StorageError - If the protocol is not registered.

get_extra_field​

def get_extra_field(protocol_id: str, field: str, default=None)

Retrieves an extra field from a protocol's information.

Arguments:

  • protocol_id str - The identifier of the protocol.
  • field str - The field name to retrieve.
  • default - The default value to return if the field is not present. Defaults to None.

Returns:

  • Any - The value of the specified field, or the default if not found.

set_extra_field​

def set_extra_field(protocol_id: str, field: str, value)

Sets an extra field in a protocol's information.

Arguments:

  • protocol_id str - The identifier of the protocol.
  • field str - The field name to set.
  • value - The value to assign to the field.

Raises:

  • StorageError - If the protocol is not registered.