Skip to content

aea.registries.base

This module contains registries.

Registry Objects

class Registry(Generic[ItemId, Item], WithLogger, ABC)

This class implements an abstract registry.

__init__

def __init__(agent_name: str = "standalone") -> None

Initialize the registry.

Arguments:

  • agent_name: the name of the agent

register

@abstractmethod
def register(item_id: ItemId,
             item: Item,
             is_dynamically_added: bool = False) -> None

Register an item.

Arguments:

  • item_id: the public id of the item.
  • item: the item.
  • is_dynamically_added: whether or not the item is dynamically added.

Raises:

  • ValueError: if an item is already registered with that item id.

Returns:

None

unregister

@abstractmethod
def unregister(item_id: ItemId) -> Optional[Item]

Unregister an item.

Arguments:

  • item_id: the public id of the item.

Raises:

  • ValueError: if no item registered with that item id.

Returns:

the item

fetch

@abstractmethod
def fetch(item_id: ItemId) -> Optional[Item]

Fetch an item.

Arguments:

  • item_id: the public id of the item.

Returns:

the Item

fetch_all

@abstractmethod
def fetch_all() -> List[Item]

Fetch all the items.

Returns:

the list of items.

ids

@abstractmethod
def ids() -> Set[ItemId]

Return the set of all the used item ids.

Returns:

the set of item ids.

setup

@abstractmethod
def setup() -> None

Set up registry.

Returns:

None

teardown

@abstractmethod
def teardown() -> None

Teardown the registry.

Returns:

None

PublicIdRegistry Objects

class PublicIdRegistry(Generic[Item], Registry[PublicId, Item])

This class implement a registry whose keys are public ids.

In particular, it is able to handle the case when the public id points to the 'latest' version of a package.

__init__

def __init__() -> None

Initialize the registry.

register

def register(public_id: PublicId,
             item: Item,
             is_dynamically_added: bool = False) -> None

Register an item.

unregister

def unregister(public_id: PublicId) -> Item

Unregister an item.

fetch

def fetch(public_id: PublicId) -> Optional[Item]

Fetch an item associated with a public id.

Arguments:

  • public_id: the public id.

Returns:

an item, or None if the key is not present.

fetch_all

def fetch_all() -> List[Item]

Fetch all the items.

ids

def ids() -> Set[PublicId]

Get all the item ids.

setup

def setup() -> None

Set up the items.

teardown

def teardown() -> None

Tear down the items.

AgentComponentRegistry Objects

class AgentComponentRegistry(Registry[ComponentId, Component])

This class implements a simple dictionary-based registry for agent components.

__init__

def __init__(**kwargs: Any) -> None

Instantiate the registry.

Arguments:

  • kwargs: kwargs

register

def register(component_id: ComponentId,
             component: Component,
             is_dynamically_added: bool = False) -> None

Register a component.

Arguments:

  • component_id: the id of the component.
  • component: the component object.
  • is_dynamically_added: whether or not the item is dynamically added.

unregister

def unregister(component_id: ComponentId) -> Optional[Component]

Unregister a component.

Arguments:

  • component_id: the ComponentId

Returns:

the item

fetch

def fetch(component_id: ComponentId) -> Optional[Component]

Fetch the component by id.

Arguments:

  • component_id: the contract id

Returns:

the component or None if the component is not registered

fetch_all

def fetch_all() -> List[Component]

Fetch all the components.

Returns:

the list of registered components.

fetch_by_type

def fetch_by_type(component_type: ComponentType) -> List[Component]

Fetch all the components by a given type..

Arguments:

  • component_type: a component type

Returns:

the list of registered components of a given type.

ids

def ids() -> Set[ComponentId]

Get the item ids.

setup

def setup() -> None

Set up the registry.

teardown

def teardown() -> None

Teardown the registry.

ComponentRegistry Objects

class ComponentRegistry(Registry[Tuple[PublicId, str], SkillComponentType],
                        Generic[SkillComponentType])

This class implements a generic registry for skill components.

__init__

def __init__(**kwargs: Any) -> None

Instantiate the registry.

Arguments:

  • kwargs: kwargs

register

def register(item_id: Tuple[PublicId, str],
             item: SkillComponentType,
             is_dynamically_added: bool = False) -> None

Register a item.

Arguments:

  • item_id: a pair (skill id, item name).
  • item: the item to register.
  • is_dynamically_added: whether or not the item is dynamically added.

Raises:

  • ValueError: if an item is already registered with that item id.

unregister

def unregister(item_id: Tuple[PublicId, str]) -> Optional[SkillComponentType]

Unregister a item.

Arguments:

  • item_id: a pair (skill id, item name).

Returns:

skill component

fetch

def fetch(item_id: Tuple[PublicId, str]) -> Optional[SkillComponentType]

Fetch an item.

Arguments:

  • item_id: the public id of the item.

Returns:

the Item

fetch_by_skill

def fetch_by_skill(skill_id: PublicId) -> List[SkillComponentType]

Fetch all the items of a given skill.

fetch_all

def fetch_all() -> List[SkillComponentType]

Fetch all the items.

unregister_by_skill

def unregister_by_skill(skill_id: PublicId) -> None

Unregister all the components by skill.

ids

def ids() -> Set[Tuple[PublicId, str]]

Get the item ids.

setup

def setup() -> None

Set up the items in the registry.

teardown

def teardown() -> None

Teardown the registry.

HandlerRegistry Objects

class HandlerRegistry(ComponentRegistry[Handler])

This class implements the handlers registry.

__init__

def __init__(**kwargs: Any) -> None

Instantiate the registry.

Arguments:

  • kwargs: kwargs

register

def register(item_id: Tuple[PublicId, str],
             item: Handler,
             is_dynamically_added: bool = False) -> None

Register a handler.

Arguments:

  • item_id: the item id.
  • item: the handler.
  • is_dynamically_added: whether or not the item is dynamically added.

Raises:

  • ValueError: if the protocol is None, or an item with pair (skill_id, protocol_id_ already exists.

unregister

def unregister(item_id: Tuple[PublicId, str]) -> Handler

Unregister a item.

Arguments:

  • item_id: a pair (skill id, item name).

Returns:

the unregistered handler

unregister_by_skill

def unregister_by_skill(skill_id: PublicId) -> None

Unregister all the components by skill.

fetch_by_protocol

def fetch_by_protocol(protocol_id: PublicId) -> List[Handler]

Fetch the handler by the pair protocol id and skill id.

Arguments:

  • protocol_id: the protocol id

Returns:

the handlers registered for the protocol_id and skill_id

fetch_by_protocol_and_skill

def fetch_by_protocol_and_skill(protocol_id: PublicId,
                                skill_id: PublicId) -> Optional[Handler]

Fetch the handler by the pair protocol id and skill id.

Arguments:

  • protocol_id: the protocol id
  • skill_id: the skill id.

Returns:

the handlers registered for the protocol_id and skill_id