enrobie.robie package#

Subpackages#

Submodules#

enrobie.robie.common module#

enrobie.robie.config module#

class enrobie.robie.config.RobieConfig(sargs: dict[str, Any] | None = None, files: str | Path | list[str | Path] | list[str] | list[Path] | tuple[str | Path, ...] | set[str] | None = None, cargs: dict[str, Any] | None = None)[source]#

Bases: Config

Contain the configurations from the arguments and files.

Parameters:

  • sargs – Additional arguments on the command line.

  • files – Complete or relative path to config files.

  • cargs – Configuration arguments in dictionary form, which will override contents from the config files.

property inserts: dict[Literal['clients', 'plugins'], dict[str, Any]]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: RobieParams#

Return the Pydantic model containing the configuration.

Warning

This method completely overrides the parent but is based on that code, would be unfortunate if upstream changes meant this breaks or breaks something else.

Returns:

Pydantic model containing the configuration.

merge_params() → None[source]#

Update the Pydantic model containing the configuration.

property config: dict[str, Any]#

Return the configuration dumped from the Pydantic model.

Warning

This method completely overrides the parent but is based on that code, would be unfortunate if upstream changes meant this breaks or breaks something else.

Returns:

Configuration dumped from the Pydantic model.

register(name: str, *, client: Type[RobieClientParams] | None = None, plugin: Type[RobiePluginParams] | None = None, source: dict[str, Any] | None = None, merge: bool = True) → None[source]#

Register the plugin parameters for parameter processing.

Parameters:

  • name – Name of the object within the Robie config.

  • client – Class definition for the instantiation.

  • plugin – Class definition for the instantiation.

  • source – Source for the parameters instantiation.

  • merge – Reprocess all parameters including added.

register_locate() → None[source]#

Register the plugin parameters for parameter processing.

enrobie.robie.robie module#

class enrobie.robie.robie.Robie(config: RobieConfig)[source]#

Bases: object

Interact with chat networks and integrate using plugins.

Parameters:

config – Primary class instance for configuration.

property config: RobieConfig#

Return the Config instance containing the configuration.

Returns:

Config instance containing the configuration.

property logger: RobieLogger#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property jinja2: RobieJinja2#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property childs: RobieChilds#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: RobieParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property console: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property debug: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property dumped: dict[str, Any]#

Return the facts about the attributes from the instance.

Returns:

Facts about the attributes from the instance.

printer(source: RobiePrint, color: int = 6) → None[source]#

Print the contents for the object within Robie instance.

Parameters:

  • source – Content which will be shown after header.

  • color – Override the color used for box character.

register(name: str, *, client: Type[RobieClient] | None = None, plugin: Type[RobiePlugin] | None = None) → None[source]#

Register the plugin with the internal operation routine.

Parameters:

  • name – Name of the object within the Robie config.

  • client – Class definition for the instantiation.

  • plugin – Class definition for the instantiation.

register_locate() → None[source]#

Register the plugin with the internal operation routine.

person(client: RobieClient, check: str) → RobiePerson | None[source]#

Return the heaviest weighted match using provided check.

Parameters:

  • client – Client class instance for Chatting Robie.

  • check – Value to be searched within the haystack.

Returns:

Heaviest weighted match using provided check.

j2parse(value: Any, statics: dict[str, Any] | None = None, literal: bool = True) → Any[source]#

Return the provided input using the Jinja2 environment.

Parameters:

  • value – Input that will be processed and returned.

  • statics – Additional values available for parsing.

  • literal – Determine if Python objects are evaled.

Returns:

Provided input using the Jinja2 environment.

enrobie.robie.service module#

class enrobie.robie.service.RobieService(robie: Robie)[source]#

Bases: object

Multi-threaded service for processing the desired state.

Parameters:

robie – Primary class instance for Chatting Robie.

property robie: Robie#

Return the Robie instance to which the instance belongs.

Returns:

Robie instance to which the instance belongs.

property params: RobieServiceParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property clients: RobieClients#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property plugins: RobiePlugins#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property running: list[str]#

Return the list of threads which are determined running.

Returns:

List of threads which are determined running.

property zombies: list[str]#

Return the list of threads which are determined zombies.

Returns:

List of threads which are determined zombies.

property congest: list[tuple[str, int]]#

Return the list of congested threads and members queues.

Returns:

List of congested threads and members queues.

property enqueue: list[tuple[str, int]]#

Return the list of congested threads and members queues.

Returns:

List of congested threads and members queues.

start() → None[source]#

Start the various threads within the Robie class object.

operate() → None[source]#

Perform the operation related to Robie service members.

operate_clients() → None[source]#

Perform the operation related to Robie service members.

operate_plugins() → None[source]#

Perform the operation related to Robie service members.

operate_healths() → None[source]#

Perform the operation related to Robie service members.

check_zombies() → bool[source]#

Return the boolean indicating while threads are zombies.

Returns:

Boolean indicating while threads are zombies.

check_congest() → bool[source]#

Return the boolean indicating when queues are congested.

Returns:

Boolean indicating when queues are congested.

soft(*args: Any, **kwargs: Any) → None[source]#

Stop the various threads within the Robie class object.

Parameters:

  • kwargs – Keyword arguments ignored by the method.

  • args – Positional arguments ignored by the method.

stop(*args: Any, **kwargs: Any) → None[source]#

Stop the various threads within the Robie class object.

Parameters:

  • kwargs – Keyword arguments ignored by the method.

  • args – Positional arguments ignored by the method.

limit(clients: list[str] | None = None, plugins: list[str] | None = None) → None[source]#

Remove the threads from members if not already started.

Parameters:

  • clients – Names of the clients that are permitted.

  • plugins – Names of the plugins that are permitted.

Module contents#

class enrobie.robie.Robie(config: RobieConfig)[source]#

Bases: object

Interact with chat networks and integrate using plugins.

Parameters:

config – Primary class instance for configuration.

property config: RobieConfig#

Return the Config instance containing the configuration.

Returns:

Config instance containing the configuration.

property logger: RobieLogger#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property jinja2: RobieJinja2#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property childs: RobieChilds#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: RobieParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property console: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property debug: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property dumped: dict[str, Any]#

Return the facts about the attributes from the instance.

Returns:

Facts about the attributes from the instance.

printer(source: RobiePrint, color: int = 6) → None[source]#

Print the contents for the object within Robie instance.

Parameters:

  • source – Content which will be shown after header.

  • color – Override the color used for box character.

register(name: str, *, client: Type[RobieClient] | None = None, plugin: Type[RobiePlugin] | None = None) → None[source]#

Register the plugin with the internal operation routine.

Parameters:

  • name – Name of the object within the Robie config.

  • client – Class definition for the instantiation.

  • plugin – Class definition for the instantiation.

register_locate() → None[source]#

Register the plugin with the internal operation routine.

person(client: RobieClient, check: str) → RobiePerson | None[source]#

Return the heaviest weighted match using provided check.

Parameters:

  • client – Client class instance for Chatting Robie.

  • check – Value to be searched within the haystack.

Returns:

Heaviest weighted match using provided check.

j2parse(value: Any, statics: dict[str, Any] | None = None, literal: bool = True) → Any[source]#

Return the provided input using the Jinja2 environment.

Parameters:

  • value – Input that will be processed and returned.

  • statics – Additional values available for parsing.

  • literal – Determine if Python objects are evaled.

Returns:

Provided input using the Jinja2 environment.

class enrobie.robie.RobieConfig(sargs: dict[str, Any] | None = None, files: str | Path | list[str | Path] | list[str] | list[Path] | tuple[str | Path, ...] | set[str] | None = None, cargs: dict[str, Any] | None = None)[source]#

Bases: Config

Contain the configurations from the arguments and files.

Parameters:

  • sargs – Additional arguments on the command line.

  • files – Complete or relative path to config files.

  • cargs – Configuration arguments in dictionary form, which will override contents from the config files.

property inserts: dict[Literal['clients', 'plugins'], dict[str, Any]]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: RobieParams#

Return the Pydantic model containing the configuration.

Warning

This method completely overrides the parent but is based on that code, would be unfortunate if upstream changes meant this breaks or breaks something else.

Returns:

Pydantic model containing the configuration.

merge_params() → None[source]#

Update the Pydantic model containing the configuration.

property config: dict[str, Any]#

Return the configuration dumped from the Pydantic model.

Warning

This method completely overrides the parent but is based on that code, would be unfortunate if upstream changes meant this breaks or breaks something else.

Returns:

Configuration dumped from the Pydantic model.

register(name: str, *, client: Type[RobieClientParams] | None = None, plugin: Type[RobiePluginParams] | None = None, source: dict[str, Any] | None = None, merge: bool = True) → None[source]#

Register the plugin parameters for parameter processing.

Parameters:

  • name – Name of the object within the Robie config.

  • client – Class definition for the instantiation.

  • plugin – Class definition for the instantiation.

  • source – Source for the parameters instantiation.

  • merge – Reprocess all parameters including added.

register_locate() → None[source]#

Register the plugin parameters for parameter processing.

class enrobie.robie.RobieModels[source]#

Bases: object

Return the class object that was imported within method.

classmethod robie() → Type[RobieParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod printer() → Type[RobiePrinterParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod service() → Type[RobieServiceParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod child() → Type[RobieChildParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod client() → Type[RobieClientParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod plugin() → Type[RobiePluginParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod person() → Type[RobiePersonParams][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod queue() → Type[RobieQueueItem][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod message() → Type[RobieMessage][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

classmethod command() → Type[RobieCommand][source]#

Return the class object that was imported within method.

Returns:

Class object that was imported within method.

class enrobie.robie.RobieService(robie: Robie)[source]#

Bases: object

Multi-threaded service for processing the desired state.

Parameters:

robie – Primary class instance for Chatting Robie.

property robie: Robie#

Return the Robie instance to which the instance belongs.

Returns:

Robie instance to which the instance belongs.

property params: RobieServiceParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property clients: RobieClients#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property plugins: RobiePlugins#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property running: list[str]#

Return the list of threads which are determined running.

Returns:

List of threads which are determined running.

property zombies: list[str]#

Return the list of threads which are determined zombies.

Returns:

List of threads which are determined zombies.

property congest: list[tuple[str, int]]#

Return the list of congested threads and members queues.

Returns:

List of congested threads and members queues.

property enqueue: list[tuple[str, int]]#

Return the list of congested threads and members queues.

Returns:

List of congested threads and members queues.

start() → None[source]#

Start the various threads within the Robie class object.

operate() → None[source]#

Perform the operation related to Robie service members.

operate_clients() → None[source]#

Perform the operation related to Robie service members.

operate_plugins() → None[source]#

Perform the operation related to Robie service members.

operate_healths() → None[source]#

Perform the operation related to Robie service members.

check_zombies() → bool[source]#

Return the boolean indicating while threads are zombies.

Returns:

Boolean indicating while threads are zombies.

check_congest() → bool[source]#

Return the boolean indicating when queues are congested.

Returns:

Boolean indicating when queues are congested.

soft(*args: Any, **kwargs: Any) → None[source]#

Stop the various threads within the Robie class object.

Parameters:

  • kwargs – Keyword arguments ignored by the method.

  • args – Positional arguments ignored by the method.

stop(*args: Any, **kwargs: Any) → None[source]#

Stop the various threads within the Robie class object.

Parameters:

  • kwargs – Keyword arguments ignored by the method.

  • args – Positional arguments ignored by the method.

limit(clients: list[str] | None = None, plugins: list[str] | None = None) → None[source]#

Remove the threads from members if not already started.

Parameters:

  • clients – Names of the clients that are permitted.

  • plugins – Names of the plugins that are permitted.

On this page
Show Source