fabricatio_core.models.role

Module that contains the Role class for managing workflows and their event registrations.

Attributes

`RoleName`
`EventPattern`
`ROLE_REGISTRY`
`EXCLUDED_FIELDS`	The set of fields that should not be resolved during configuration resolution.

Classes

Role

Class that represents a role with a registry of events and workflows.

Functions

`register_role`(→ None)	Register the role into the global registry.
`unregister_role`(→ None)	Unregister the role from the global registry.
`clear_registry`(→ None)	Clear the global registry of all registered roles.
`get_registered_role`(…)	Get a registered role by name.

Module Contents

type fabricatio_core.models.role.RoleName = str

type fabricatio_core.models.role.EventPattern = str

fabricatio_core.models.role.ROLE_REGISTRY: Dict[RoleName, Role]

class fabricatio_core.models.role.Role(/, **data: Any)[source]

Bases: fabricatio_core.models.generic.WithBriefing

Class that represents a role with a registry of events and workflows.

A Role serves as a container for workflows, managing their registration to events and providing them with shared configuration like tools and personality.

model_config: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: RoleName = None: The name of the role.

description: str = '': A brief description of the role’s responsibilities and capabilities.

subscriptions: Dict[EventPattern, fabricatio_core.models.action.WorkFlow] = None: A dictionary of event-workflow pairs.

classmethod new(subscriptions: Dict[EventPattern, fabricatio_core.models.action.WorkFlow], /, name: RoleName | None = None, description: str = '', dispatch_on_init: bool = False, **kwargs: Unpack[TypedDict]) → Self[source]: Create a new Role.

classmethod with_bio(name: RoleName | None = None, description: str = '') → Self[source]: Create a new Role with a bio.

classmethod with_subscriptions(subscriptions: Dict[EventPattern, fabricatio_core.models.action.WorkFlow]) → Self[source]: Create a new Role with subscription specified only.

property briefing: str

Get the briefing of the role.

Returns:: The briefing of the role.
Return type:: str

property accept_events: List[str]

Get the set of events that the role accepts.

Returns:: The set of events that the role accepts.
Return type:: Set[Event]

model_post_init(__context: Any) → None[source]: Register the role.

configure(/, **kwargs) → Self[source]
configure(fn: Callable[[Self], None]) → Self: Configure the role.

subscribe(event: fabricatio_core.rust.Event | EventPattern, workflow: fabricatio_core.models.action.WorkFlow) → Self[source]: Register a workflow to the role’s registry.

unsubscribe(event: fabricatio_core.rust.Event | EventPattern) → Self[source]: Unregister a workflow from the role’s registry for the given event.

dispatch(resolve_config: bool = True) → Self[source]

Register each workflow in the registry to its corresponding event in the event bus.

Returns:: The role instance for method chaining
Return type:: Self

undo_dispatch() → Self[source]

Unregister each workflow in the registry from its corresponding event in the event bus.

Returns:: The role instance for method chaining
Return type:: Self

resolve_configuration() → Self[source]

Resolve and bind shared configuration to workflows and their components.

This method ensures that any shared configuration from the role or workflows is properly propagated to the workflow steps and nested components. If the role is a ScopedConfig, it holds configuration for all workflows. Similarly, if a workflow itself is a ScopedConfig, it holds configuration for its own steps.

Returns:: The role instance with resolved configurations.
Return type:: Self

fabricatio_core.models.role.register_role(role: Role, override: bool = True) → None[source]: Register the role into the global registry.

fabricatio_core.models.role.unregister_role(role: Role | RoleName) → None[source]: Unregister the role from the global registry.

fabricatio_core.models.role.clear_registry() → None[source]: Clear the global registry of all registered roles.

fabricatio_core.models.role.get_registered_role(role_name: RoleName) → Role[source]
fabricatio_core.models.role.get_registered_role(role_name: Set[RoleName]) → List[Role]: Get a registered role by name.

fabricatio_core.models.role.EXCLUDED_FIELDS: The set of fields that should not be resolved during configuration resolution.