enrobie.robie package#

Subpackages#

Submodules#

enrobie.robie.common module#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

enrobie.robie.config module#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class enrobie.robie.config.RobieConfig(sargs: dict[str, Any] | None = None, files: str | Path | list[str | 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 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.

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.

merge_params() → None[source]#

Update the Pydantic model containing the configuration.

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.

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#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

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 childs: RobieChilds#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property config: RobieConfig#

Return the Config instance containing the configuration.

Returns:

Config instance containing the configuration.

property dumped: dict[str, Any]#

Return the facts about the attributes from the instance.

Returns:

Facts about the attributes from the instance.

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.

property jinja2: RobieJinja2#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property logger: RobieLogger#

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.

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.

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.

enrobie.robie.service module#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

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.

check_congest() → bool[source]#

Return the boolean indicating when queues are congested.

Returns:

Boolean indicating when queues are congested.

check_zombies() → bool[source]#

Return the boolean indicating while threads are zombies.

Returns:

Boolean indicating while threads are zombies.

property clients: RobieClients#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

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.

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.

operate() → None[source]#

Perform the operation related to Robie service members.

operate_clients() → None[source]#

Perform the operation related to Robie service members.

operate_healths() → None[source]#

Perform the operation related to Robie service members.

operate_plugins() → None[source]#

Perform the operation related to Robie service members.

property params: RobieServiceParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property plugins: RobiePlugins#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property robie: Robie#

Return the Robie instance to which the instance belongs.

Returns:

Robie instance to which the instance belongs.

property running: list[str]#

Return the list of threads which are determined running.

Returns:

List of threads which are determined running.

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.

start() → None[source]#

Start the various threads within the Robie class object.

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.

property zombies: list[str]#

Return the list of threads which are determined zombies.

Returns:

List of threads which are determined zombies.

Module contents#

Functions and routines associated with Enasis Network Chatting Robie.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

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 childs: RobieChilds#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property config: RobieConfig#

Return the Config instance containing the configuration.

Returns:

Config instance containing the configuration.

property dumped: dict[str, Any]#

Return the facts about the attributes from the instance.

Returns:

Facts about the attributes from the instance.

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.

property jinja2: RobieJinja2#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property logger: RobieLogger#

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.

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.

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.

class enrobie.robie.RobieConfig(sargs: dict[str, Any] | None = None, files: str | Path | list[str | 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 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.

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.

merge_params() → None[source]#

Update the Pydantic model containing the configuration.

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.

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.RobieService(robie: Robie)[source]#

Bases: object

Multi-threaded service for processing the desired state.

Parameters:

robie – Primary class instance for Chatting Robie.

check_congest() → bool[source]#

Return the boolean indicating when queues are congested.

Returns:

Boolean indicating when queues are congested.

check_zombies() → bool[source]#

Return the boolean indicating while threads are zombies.

Returns:

Boolean indicating while threads are zombies.

property clients: RobieClients#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

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.

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.

operate() → None[source]#

Perform the operation related to Robie service members.

operate_clients() → None[source]#

Perform the operation related to Robie service members.

operate_healths() → None[source]#

Perform the operation related to Robie service members.

operate_plugins() → None[source]#

Perform the operation related to Robie service members.

property params: RobieServiceParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property plugins: RobiePlugins#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property robie: Robie#

Return the Robie instance to which the instance belongs.

Returns:

Robie instance to which the instance belongs.

property running: list[str]#

Return the list of threads which are determined running.

Returns:

List of threads which are determined running.

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.

start() → None[source]#

Start the various threads within the Robie class object.

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.

property zombies: list[str]#

Return the list of threads which are determined zombies.

Returns:

List of threads which are determined zombies.

On this page

This Page