enconnect.irc package

Subpackages

• enconnect.irc.test package
  • Submodules
  • enconnect.irc.test.helpers module
    • IRCClientSocket
    • client_ircsock()
  • enconnect.irc.test.test_client module
    • client()
    • test_Client()
    • test_Client_connect()
    • test_Client_cover()
  • enconnect.irc.test.test_models module
    • test_ClientEvent()
    • test_ClientEvent_cover()
    • test_ClientEvent_service()
  • Module contents
    • IRCClientSocket
    • client_ircsock()

Submodules

enconnect.irc.client module

Functions and routines associated with Enasis Network Remote Connect.

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

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 canceled: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property connected: bool

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 nickname: str | None

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.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

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.

socket_send(send: str) → None

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

stop() → None

Gracefully close the connection with the server socket.

enconnect.irc.models module

Functions and routines associated with Enasis Network Remote Connect.

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

class enconnect.irc.models.ClientEvent(client: Client, event: str)

Bases: BaseModel

Contains information returned from the upstream server.

author: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', 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)])]

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

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

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

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

model_config = {'extra': 'ignore'}

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

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

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

prefix: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Prefix or origin information', 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)])]

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

enconnect.irc.params module

Functions and routines associated with Enasis Network Remote Connect.

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

class enconnect.irc.params.ClientParams(*, server: Annotated[str, MinLen(min_length=1)], port: Annotated[int, Ge(ge=1), Le(le=65535)] = 6697, timeout: Annotated[int, Ge(ge=1), Le(le=300)] = 30, operate: Literal['normal', 'service'] = 'normal', nickname: Annotated[str, MinLen(min_length=1)] = 'ircbot', username: Annotated[str, MinLen(min_length=1)] = 'ircbot', realname: Annotated[str, MinLen(min_length=1)] = 'Chatting Robie', password: Annotated[str | None, MinLen(min_length=1)] = None, servername: Annotated[str, MinLen(min_length=1)] = 'services.invalid', serverid: Annotated[str, MinLen(min_length=1)] = '42X', ssl_enable: bool = True, ssl_verify: bool = True, queue_size: Annotated[int, Ge(ge=1000), Le(le=1000000)] = 10000)

Bases: BaseModel

Process and validate the class configuration parameters.

model_config = {'extra': 'forbid'}

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

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

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

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Parameter for the integration', 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)])]

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

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

server: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Server address for connection', 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)])]

servername: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='services.invalid', description='Parameter for the integration', 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')]

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

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

Module contents

Functions and routines associated with Enasis Network Remote Connect.

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

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 canceled: bool

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property connected: bool

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 nickname: str | None

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.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

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.

socket_send(send: str) → None

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

stop() → None

Gracefully close the connection with the server socket.

class enconnect.irc.ClientEvent(client: Client, event: str)

Bases: BaseModel

Contains information returned from the upstream server.

author: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Dynamic field parsed from event', 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)])]

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

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

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

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

model_config = {'extra': 'ignore'}

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

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

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

prefix: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Prefix or origin information', 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)])]

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

class enconnect.irc.ClientParams(*, server: Annotated[str, MinLen(min_length=1)], port: Annotated[int, Ge(ge=1), Le(le=65535)] = 6697, timeout: Annotated[int, Ge(ge=1), Le(le=300)] = 30, operate: Literal['normal', 'service'] = 'normal', nickname: Annotated[str, MinLen(min_length=1)] = 'ircbot', username: Annotated[str, MinLen(min_length=1)] = 'ircbot', realname: Annotated[str, MinLen(min_length=1)] = 'Chatting Robie', password: Annotated[str | None, MinLen(min_length=1)] = None, servername: Annotated[str, MinLen(min_length=1)] = 'services.invalid', serverid: Annotated[str, MinLen(min_length=1)] = '42X', ssl_enable: bool = True, ssl_verify: bool = True, queue_size: Annotated[int, Ge(ge=1000), Le(le=1000000)] = 10000)

Bases: BaseModel

Process and validate the class configuration parameters.

model_config = {'extra': 'forbid'}

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

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

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

password: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Parameter for the integration', 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)])]

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

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

server: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Server address for connection', 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)])]

servername: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='services.invalid', description='Parameter for the integration', 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')]

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

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