enconnect.irc package

Submodules

enconnect.irc.client module

class enconnect.irc.client.Client(params: ClientParams, logger: Callable[[...], None] | None = None)

Bases: object

Establish and maintain connection with the chat service.

Parameters:

params – Parameters used to instantiate the class.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property connected: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property nickname: str | None

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property mqueue: Queue[ClientEvent]

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property canceled: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate() → None

Operate the client and populate queue with the messages.

stop() → None

Gracefully close the connection with the server socket.

socket_send(send: str) → None

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

socket_recv() → Iterator[str]

Return the content received from the socket connection.

Note

Method could be further optimized using an internal buffer within the class after receiving more bytes.

Returns:

Content received from the socket connection.

enconnect.irc.models module

class enconnect.irc.models.ClientEvent

Bases: BaseModel

Contains information returned from the upstream server.

Fields

Field	Type	Required	Default	Constraints
author	str | None	No	None	min_length=1
command	str | None	No	None	min_length=1
hasme	bool	No	False
isme	bool	No	False
kind	Literal[‘event’, ‘chanmsg’, ‘privmsg’]	No	'event'
message	str | None	No	None	min_length=1
original	str	Yes		min_length=1
params	str | None	No	None
prefix	str | None	No	None	min_length=1
recipient	str | None	No	None	min_length=1
whome	str | None	No	None	min_length=1

prefix: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Prefix or origin information', metadata=[MinLen(min_length=1)])]

command: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Code or command for the event', metadata=[MinLen(min_length=1)])]

params: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event or command parameters')]

original: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Original received from server', metadata=[MinLen(min_length=1)])]

kind: Annotated[Literal['event', 'chanmsg', 'privmsg'], FieldInfo(annotation=NoneType, required=False, default='event', description='Dynamic field parsed from event')]

isme: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Indicates message is from client')]

hasme: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Indicates message mentions client')]

whome: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Current nickname of when received', metadata=[MinLen(min_length=1)])]

author: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]

recipient: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]

message: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]

enconnect.irc.params module

class enconnect.irc.params.ClientParams

Bases: BaseModel

Process and validate the class configuration parameters.

Fields

Field	Type	Required	Default	Constraints
nickname	str	No	'ircbot'	min_length=1
operate	Literal[‘normal’, ‘service’]	No	'normal'
password	str | None	No	None	min_length=1
port	int	No	6697	ge=1, le=65535
queue_size	int	No	10000	ge=1000, le=1000000
realname	str	No	'Chatting Robie'	min_length=1
server	str	Yes		min_length=1
serverid	str	No	'42X'	min_length=1
servername	str	No	'services.invalid'	min_length=1
ssl_enable	bool	No	True
ssl_verify	bool	No	True
timeout	int	No	30	ge=1, le=300
username	str	No	'ircbot'	min_length=1

server: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Server address for connection', metadata=[MinLen(min_length=1)])]

port: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=6697, description='Server address for connection', metadata=[Ge(ge=1), Le(le=65535)])]

timeout: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=30, description='Timeout connecting to server', metadata=[Ge(ge=1), Le(le=300)])]

operate: Annotated[Literal['normal', 'service'], FieldInfo(annotation=NoneType, required=False, default='normal', description='Method for server connection')]

nickname: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='ircbot', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

username: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='ircbot', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

realname: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='Chatting Robie', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

servername: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='services.invalid', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

serverid: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='42X', description='Unique identifier for services', metadata=[MinLen(min_length=1)])]

ssl_enable: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Enable connection encryption')]

ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]

queue_size: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=10000, description='Maximum size for queued events', metadata=[Ge(ge=1000), Le(le=1000000)])]

Module contents

class enconnect.irc.Client(params: ClientParams, logger: Callable[[...], None] | None = None)

Bases: object

Establish and maintain connection with the chat service.

Parameters:

params – Parameters used to instantiate the class.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

property connected: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property nickname: str | None

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property mqueue: Queue[ClientEvent]

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property canceled: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate() → None

Operate the client and populate queue with the messages.

stop() → None

Gracefully close the connection with the server socket.

socket_send(send: str) → None

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

socket_recv() → Iterator[str]

Return the content received from the socket connection.

Note

Method could be further optimized using an internal buffer within the class after receiving more bytes.

Returns:

Content received from the socket connection.

class enconnect.irc.ClientParams

Bases: BaseModel

Process and validate the class configuration parameters.

Fields

Field	Type	Required	Default	Constraints
nickname	str	No	'ircbot'	min_length=1
operate	Literal[‘normal’, ‘service’]	No	'normal'
password	str | None	No	None	min_length=1
port	int	No	6697	ge=1, le=65535
queue_size	int	No	10000	ge=1000, le=1000000
realname	str	No	'Chatting Robie'	min_length=1
server	str	Yes		min_length=1
serverid	str	No	'42X'	min_length=1
servername	str	No	'services.invalid'	min_length=1
ssl_enable	bool	No	True
ssl_verify	bool	No	True
timeout	int	No	30	ge=1, le=300
username	str	No	'ircbot'	min_length=1

server: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Server address for connection', metadata=[MinLen(min_length=1)])]

port: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=6697, description='Server address for connection', metadata=[Ge(ge=1), Le(le=65535)])]

timeout: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=30, description='Timeout connecting to server', metadata=[Ge(ge=1), Le(le=300)])]

operate: Annotated[Literal['normal', 'service'], FieldInfo(annotation=NoneType, required=False, default='normal', description='Method for server connection')]

nickname: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='ircbot', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

username: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='ircbot', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

realname: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='Chatting Robie', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

servername: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='services.invalid', description='Parameter for the integration', metadata=[MinLen(min_length=1)])]

serverid: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='42X', description='Unique identifier for services', metadata=[MinLen(min_length=1)])]

ssl_enable: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Enable connection encryption')]

ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]

queue_size: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=10000, description='Maximum size for queued events', metadata=[Ge(ge=1000), Le(le=1000000)])]

class enconnect.irc.ClientEvent

Bases: BaseModel

Contains information returned from the upstream server.

Fields

Field	Type	Required	Default	Constraints
author	str | None	No	None	min_length=1
command	str | None	No	None	min_length=1
hasme	bool	No	False
isme	bool	No	False
kind	Literal[‘event’, ‘chanmsg’, ‘privmsg’]	No	'event'
message	str | None	No	None	min_length=1
original	str	Yes		min_length=1
params	str | None	No	None
prefix	str | None	No	None	min_length=1
recipient	str | None	No	None	min_length=1
whome	str | None	No	None	min_length=1

prefix: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Prefix or origin information', metadata=[MinLen(min_length=1)])]

command: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Code or command for the event', metadata=[MinLen(min_length=1)])]

params: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event or command parameters')]

original: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Original received from server', metadata=[MinLen(min_length=1)])]

kind: Annotated[Literal['event', 'chanmsg', 'privmsg'], FieldInfo(annotation=NoneType, required=False, default='event', description='Dynamic field parsed from event')]

isme: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Indicates message is from client')]

hasme: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Indicates message mentions client')]

whome: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Current nickname of when received', metadata=[MinLen(min_length=1)])]

author: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]

recipient: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]

message: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', metadata=[MinLen(min_length=1)])]