enrobie.plugins.logger package

Submodules

enrobie.plugins.logger.history module

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

orm_declarative_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.

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.

plugin

client

person

kind

author

anchor

message

create

class enrobie.plugins.logger.history.LoggerHistoryRecord

Bases: BaseModel

Contain the information regarding the chatting history.

Parameters:

• record – Record from the SQLAlchemy query routine.

• kwargs – Keyword arguments passed for downstream.

Fields

Field	Type	Required	Default	Constraints
anchor	str	No	None	min_length=1
author	str	Yes		min_length=1
client	str	Yes		min_length=1
create	str	Yes		min_length=20, max_length=32
kind	Literal[‘privmsg’, ‘chanmsg’]	Yes		min_length=1
message	str	Yes
person	str | None	No	None	min_length=1
plugin	str	Yes		min_length=1

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

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

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

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

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

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

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

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

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.

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

Insert the record into the historical chat interactions.

process(mitem: RobieMessage) → None

Insert the record into the historical chat interactions.

Parameters:

mitem – Item containing information for operation.

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.

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.

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.

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.

enrobie.plugins.logger.params module

class enrobie.plugins.logger.params.LoggerPluginParams

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

Fields

Field	Type	Required	Default	Constraints
clients	list[str]	Yes		min_length=1
database	str	No	'sqlite:///:memory:'	min_length=1
enable	bool	No	False
histories	int	No	100	ge=1, le=10000
locate	str | None	No	None	min_length=1
output	str | None	No	None	min_length=4
status	StatusPluginIconParams	No	factory
trusted	list[str] | None	No	None	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)])]

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

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

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.

validate() → None

Perform advanced validation on the parameters provided.

classmethod schema() → Type[LoggerPluginParams]

Return the configuration parameters relevant for class.

Returns:

Configuration parameters relevant for class.

property params: LoggerPluginParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

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.

Module contents

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.

validate() → None

Perform advanced validation on the parameters provided.

classmethod schema() → Type[LoggerPluginParams]

Return the configuration parameters relevant for class.

Returns:

Configuration parameters relevant for class.

property params: LoggerPluginParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

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.

class enrobie.plugins.logger.LoggerPluginParams

Bases: RobiePluginParams

Process and validate the Robie configuration parameters.

Fields

Field	Type	Required	Default	Constraints
clients	list[str]	Yes		min_length=1
database	str	No	'sqlite:///:memory:'	min_length=1
enable	bool	No	False
histories	int	No	100	ge=1, le=10000
locate	str | None	No	None	min_length=1
output	str | None	No	None	min_length=4
status	StatusPluginIconParams	No	factory
trusted	list[str] | None	No	None	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)])]

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

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')]