enrobie.plugins.logger package

Subpackages

• enrobie.plugins.logger.test package
  • Submodules
  • enrobie.plugins.logger.test.test_history module
    • test_LoggerHistory()
    • test_LoggerHistory_cover()
  • enrobie.plugins.logger.test.test_plugin module
    • test_LoggerPlugin()
    • test_LoggerPlugin_cover()
  • Module contents

Submodules

enrobie.plugins.logger.history 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.plugins.logger.history.LoggerHistory(plugin: LoggerPlugin)

Bases: object

Store the historical information for chat interactions.

Parameters:

plugin – Plugin class instance for Chatting Robie.

expunge(client: str, kind: str, anchor: str) → None

Remove the expired historical chat interaction records.

Parameters:

• client – Unique identifier for the client object.

• kind – What kind of Robie message we dealing with.

• anchor – Unique identifier for the chat context.

insert(*, client: str, person: str | None, kind: str, author: str, anchor: str, message: str) → None

Insert the record into the historical chat interactions.

plaintext(limit: int | None = None, mitem: RobieMessage | None = None, *, client: str | None = None, person: str | None = None, kind: str | None = None, author: str | None = None, anchor: str | None = None) → list[str]

Return the historical records for the chat interactions.

Note

Additional keyword arguments ignored when mitem.

Parameters:

• limit – Optionally restrict the records returned.

• mitem – Item containing information for operation.

Returns:

Historical records for the chat interactions.

process(mitem: RobieMessage) → None

Insert the record into the historical chat interactions.

Parameters:

mitem – Item containing information for operation.

records(mitem: RobieMessage, limit: int | None = None) → list[LoggerHistoryRecord]

Return the historical records for the chat interactions.

Parameters:

• mitem – Item containing information for operation.

• limit – Optionally restrict the records returned.

Returns:

Historical records for the chat interactions.

search(limit: int | None = None, *, client: str | None = None, person: str | None = None, kind: str | None = None, author: str | None = None, anchor: str | None = None) → list[LoggerHistoryRecord]

Return the historical records for the chat interactions.

Parameters:

limit – Optionally restrict the records returned.

Returns:

Historical records for the chat interactions.

class enrobie.plugins.logger.history.LoggerHistoryRecord(record: LoggerHistoryTable | None = None, *, plugin: Annotated[str, MinLen(min_length=1)], client: Annotated[str, MinLen(min_length=1)], person: Annotated[str | None, MinLen(min_length=1)] = None, kind: Annotated[Literal['privmsg', 'chanmsg'], MinLen(min_length=1)], author: Annotated[str, MinLen(min_length=1)], anchor: Annotated[str, MinLen(min_length=1)] = None, message: str, create: Annotated[str, MinLen(min_length=20), MaxLen(max_length=32)])

Bases: BaseModel

Contain the information regarding the chatting history.

Parameters:

• record – Record from the SQLAlchemy query routine.

• kwargs – Keyword arguments passed for downstream.

anchor: Annotated[str, FieldInfo(annotation=NoneType, required=False, default=None, description='Channel or private context', metadata=[MinLen(min_length=1)])]

author: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Nickname of message author', metadata=[MinLen(min_length=1)])]

client: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Client name where originated', metadata=[MinLen(min_length=1)])]

create: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='When the record was created', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])]

kind: Annotated[Literal['privmsg', 'chanmsg'], FieldInfo(annotation=NoneType, required=True, description='What kind of original message', metadata=[MinLen(min_length=1)])]

message: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Historical interaction content')]

model_config = {'extra': 'forbid'}

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

person: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Person that author is matched', metadata=[MinLen(min_length=1)])]

plugin: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Plugin name where originated', metadata=[MinLen(min_length=1)])]

class enrobie.plugins.logger.history.LoggerHistoryTable(**kwargs)

Bases: SQLBase

Schematic for the database operations using SQLAlchemy.

Note

Fields are not completely documented for this model.

anchor

author

client

create

kind

message

person

plugin

class enrobie.plugins.logger.history.SQLBase(**kwargs: Any)

Bases: DeclarativeBase

Some additional class that SQLAlchemy requires to work.

metadata: ClassVar[MetaData] = MetaData()

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[registry] = <sqlalchemy.orm.decl_api.registry object>

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

enrobie.plugins.logger.params 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.plugins.logger.params.LoggerPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, database: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)] = 'sqlite:///:memory:', histories: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=10000)] = 100, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], output: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=4)] = None, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]

database: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='sqlite:///:memory:', description='Database connection string', metadata=[MinLen(min_length=1)])]

histories: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=100, description='Number of messages per anchor', metadata=[Ge(ge=1), Le(le=10000)])]

model_config = {'extra': 'forbid'}

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

output: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional path where logs append', metadata=[MinLen(min_length=4)])]

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]

enrobie.plugins.logger.plugin 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.plugins.logger.plugin.LoggerPlugin(robie: Robie, name: str, params: RobieChildParams)

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for interacting with an LLM model.

property history: LoggerHistory

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate() → None

Perform the operation related to Robie service threads.

property params: LoggerPluginParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

classmethod schema() → Type[LoggerPluginParams]

Return the configuration parameters relevant for class.

Returns:

Configuration parameters relevant for class.

validate() → None

Perform advanced validation on the parameters provided.

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.plugins.logger.LoggerPlugin(robie: Robie, name: str, params: RobieChildParams)

Bases: RobiePlugin

Integrate with the Robie routine and perform operations.

Note

This plugin allows for interacting with an LLM model.

property history: LoggerHistory

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate() → None

Perform the operation related to Robie service threads.

property params: LoggerPluginParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

classmethod schema() → Type[LoggerPluginParams]

Return the configuration parameters relevant for class.

Returns:

Configuration parameters relevant for class.

validate() → None

Perform advanced validation on the parameters provided.

class enrobie.plugins.logger.LoggerPluginParams(_parse: ~typing.Callable[[...], ~typing.Any] | None = None, *, enable: bool = False, locate: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=1)] = None, trusted: ~typing.Annotated[list[str] | None, ~annotated_types.MinLen(min_length=1)] = None, database: ~typing.Annotated[str, ~annotated_types.MinLen(min_length=1)] = 'sqlite:///:memory:', histories: ~typing.Annotated[int, ~annotated_types.Ge(ge=1), ~annotated_types.Le(le=10000)] = 100, clients: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], output: ~typing.Annotated[str | None, ~annotated_types.MinLen(min_length=4)] = None, status: ~enrobie.plugins.status.params.StatusPluginIconParams = <factory>)

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

clients: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, description='List of clients to enable plugin', metadata=[MinLen(min_length=1)])]

database: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='sqlite:///:memory:', description='Database connection string', metadata=[MinLen(min_length=1)])]

histories: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=100, description='Number of messages per anchor', metadata=[Ge(ge=1), Le(le=10000)])]

model_config = {'extra': 'forbid'}

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

output: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional path where logs append', metadata=[MinLen(min_length=4)])]

status: Annotated[StatusPluginIconParams, FieldInfo(annotation=NoneType, required=False, default_factory=StatusPluginIconParams, description='Icon used per chat platform')]