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__

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

Initialize the registry.

Arguments:

  • agent_name: the name of the agent

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

Returns:

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

unregister

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

Unregister an item.

Arguments:

  • item_id: the public id of the item.

Returns:

the item :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.

ids

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

Return the set of all the used item ids.

Returns:

the set of item ids.

setup

 | @abstractmethod
 | setup() -> None

Set up registry.

Returns:

None

teardown

 | @abstractmethod
 | 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__

 | __init__() -> None

Initialize the registry.

register

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

Register an item.

unregister

 | unregister(public_id: PublicId) -> Item

Unregister an item.

fetch

 | 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

 | fetch_all() -> List[Item]

Fetch all the items.

ids

 | ids() -> Set[PublicId]

Get all the item ids.

setup

 | setup() -> None

Set up the items.

teardown

 | 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__

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

Instantiate the registry.

Arguments:

  • kwargs: kwargs

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

unregister

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

Unregister a component.

Arguments:

  • component_id: the ComponentId

Returns:

the item

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.

Returns:

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

Returns:

the list of registered components of a given type.

ids

 | ids() -> Set[ComponentId]

Get the item ids.

setup

 | setup() -> None

Set up the registry.

teardown

 | 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__

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

Instantiate the registry.

Arguments:

  • kwargs: kwargs

register

 | 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

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

Unregister a item.

Arguments:

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

Returns:

skill component :raises: ValueError if no item registered with that item id.

fetch

 | 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

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

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: PublicId) -> None

Unregister all the components by skill.

ids

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

Get the item ids.

setup

 | setup() -> None

Set up the items in the registry.

teardown

 | teardown() -> None

Teardown the registry.

HandlerRegistry Objects

class HandlerRegistry(ComponentRegistry[Handler])

This class implements the handlers registry.

__init__

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

Instantiate the registry.

Arguments:

  • kwargs: kwargs

register

 | 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

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

Unregister a item.

Arguments:

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

Returns:

the unregistered handler :raises: ValueError if no item is registered with that item id.

unregister_by_skill

 | unregister_by_skill(skill_id: PublicId) -> None

Unregister all the components by skill.

fetch_by_protocol

 | 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

 | 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

Back to top