aea.registries.base

This module contains registries.

Registry Objects

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

This class implements an abstract registry.

__init__

 | __init__()

Initialize the registry.

register

 | @abstractmethod
 | 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_dynamicall_added: whether or not the item is dynamicall added.

Returns:

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

unregister

 | @abstractmethod
 | unregister(item_id: ItemId) -> None

Unregister an item.

Arguments:

  • item_id: the public id of the item.

Returns:

None :raises: ValueError if no item registered with that item id.

fetch

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

Fetch an item.

Arguments:

  • item_id: the public id of the item.

Returns:

the Item

fetch_all

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

Fetch all the items.

Returns:

the list of items.

setup

 | @abstractmethod
 | setup() -> None

Set up registry.

Returns:

None

teardown

 | @abstractmethod
 | teardown() -> None

Teardown the registry.

Returns:

None

AgentComponentRegistry Objects

class AgentComponentRegistry(Registry[ComponentId, Component])

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

__init__

 | __init__() -> None

Instantiate the registry.

Returns:

None

register

 | 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_dynamicall_added: whether or not the item is dynamicall added.

unregister

 | unregister(component_id: ComponentId) -> None

Unregister a component.

Arguments:

  • component_id: the ComponentId

fetch

 | 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

 | fetch_all() -> List[Component]

Fetch all the components.

:return the list of registered components.

fetch_by_type

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

Fetch all the components by a given type..

Arguments:

  • component_type: a component type :return the list of registered components of a given type.

setup

 | setup() -> None

Set up the registry.

Returns:

None

teardown

 | teardown() -> None

Teardown the registry.

Returns:

None

ComponentRegistry Objects

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

This class implements a generic registry for skill components.

__init__

 | __init__() -> None

Instantiate the registry.

Returns:

None

register

 | register(item_id: Tuple[SkillId, 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_dynamicall_added: whether or not the item is dynamicall added.

Returns:

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

unregister

 | unregister(item_id: Tuple[SkillId, str]) -> None

Unregister a item.

Arguments:

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

Returns:

None :raises: ValueError if no item registered with that item id.

fetch

 | fetch(item_id: Tuple[SkillId, str]) -> Optional[SkillComponentType]

Fetch an item.

Arguments:

  • item_id: the public id of the item.

Returns:

the Item

fetch_by_skill

 | fetch_by_skill(skill_id: SkillId) -> List[Item]

Fetch all the items of a given skill.

fetch_all

 | fetch_all() -> List[SkillComponentType]

Fetch all the items.

unregister_by_skill

 | unregister_by_skill(skill_id: SkillId) -> None

Unregister all the components by skill.

setup

 | setup() -> None

Set up the items in the registry.

Returns:

None

teardown

 | teardown() -> None

Teardown the registry.

Returns:

None

HandlerRegistry Objects

class HandlerRegistry(ComponentRegistry[Handler])

This class implements the handlers registry.

__init__

 | __init__() -> None

Instantiate the registry.

Returns:

None

register

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

Register a handler.

Arguments:

  • item_id: the item id.
  • item: the handler.
  • is_dynamicall_added: whether or not the item is dynamicall added.

Returns:

None

Raises:

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

unregister

 | unregister(item_id: Tuple[SkillId, str]) -> None

Unregister a item.

Arguments:

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

Returns:

None :raises: ValueError if no item is registered with that item id.

unregister_by_skill

 | unregister_by_skill(skill_id: SkillId) -> None

Unregister all the components by skill.

fetch_by_protocol

 | fetch_by_protocol(protocol_id: ProtocolId) -> 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

 | fetch_by_protocol_and_skill(protocol_id: ProtocolId, skill_id: SkillId) -> 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