enconnect.irc package#

Subpackages#

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)[source]#

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[source]#

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][source]#

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[source]#

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

stop() None[source]#

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.

pydantic model enconnect.irc.models.ClientEvent[source]#

Bases: BaseModel

Contains information returned from the upstream server.

Show JSON schema
{
   "title": "ClientEvent",
   "description": "Contains information returned from the upstream server.",
   "type": "object",
   "properties": {
      "prefix": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Prefix or origin information",
         "title": "Prefix"
      },
      "command": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Code or command for the event",
         "title": "Command"
      },
      "params": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Event or command parameters",
         "title": "Params"
      },
      "original": {
         "description": "Original received from server",
         "minLength": 1,
         "title": "Original",
         "type": "string"
      },
      "kind": {
         "default": "event",
         "description": "Dynamic field parsed from event",
         "enum": [
            "event",
            "chanmsg",
            "privmsg"
         ],
         "title": "Kind",
         "type": "string"
      },
      "isme": {
         "default": false,
         "description": "Indicates message is from client",
         "title": "Isme",
         "type": "boolean"
      },
      "hasme": {
         "default": false,
         "description": "Indicates message mentions client",
         "title": "Hasme",
         "type": "boolean"
      },
      "whome": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Current nickname of when received",
         "title": "Whome"
      },
      "author": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Author"
      },
      "recipient": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Recipient"
      },
      "message": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Message"
      }
   },
   "required": [
      "original"
   ]
}

Config:
  • extra: str = ignore

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Code or command for the event

Constraints:
  • min_length = 1

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

Indicates message mentions client

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

Indicates message is from client

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

Dynamic field parsed from event

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Original received from server

Constraints:
  • min_length = 1

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

Event or command parameters

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

Prefix or origin information

Constraints:
  • min_length = 1

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Current nickname of when received

Constraints:
  • 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.

pydantic model enconnect.irc.params.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema
{
   "title": "ClientParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "port": {
         "default": 6697,
         "description": "Server address for connection",
         "maximum": 65535,
         "minimum": 1,
         "title": "Port",
         "type": "integer"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "operate": {
         "default": "normal",
         "description": "Method for server connection",
         "enum": [
            "normal",
            "service"
         ],
         "title": "Operate",
         "type": "string"
      },
      "nickname": {
         "default": "ircbot",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Nickname",
         "type": "string"
      },
      "username": {
         "default": "ircbot",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Username",
         "type": "string"
      },
      "realname": {
         "default": "Chatting Robie",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Realname",
         "type": "string"
      },
      "password": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Parameter for the integration",
         "title": "Password"
      },
      "servername": {
         "default": "services.invalid",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Servername",
         "type": "string"
      },
      "serverid": {
         "default": "42X",
         "description": "Unique identifier for services",
         "minLength": 1,
         "title": "Serverid",
         "type": "string"
      },
      "ssl_enable": {
         "default": true,
         "description": "Enable connection encryption",
         "title": "Ssl Enable",
         "type": "boolean"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "queue_size": {
         "default": 10000,
         "description": "Maximum size for queued events",
         "maximum": 1000000,
         "minimum": 1000,
         "title": "Queue Size",
         "type": "integer"
      }
   },
   "additionalProperties": false,
   "required": [
      "server"
   ]
}

Config:
  • extra: str = forbid

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Method for server connection

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Server address for connection

Constraints:
  • ge = 1

  • le = 65535

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

Maximum size for queued events

Constraints:
  • ge = 1000

  • le = 1000000

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Server address for connection

Constraints:
  • min_length = 1

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

Unique identifier for services

Constraints:
  • min_length = 1

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Enable connection encryption

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

Verify the ceritifcate valid

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

Timeout connecting to server

Constraints:
  • ge = 1

  • le = 300

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

Parameter for the integration

Constraints:
  • 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)[source]#

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[source]#

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][source]#

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[source]#

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

stop() None[source]#

Gracefully close the connection with the server socket.

pydantic model enconnect.irc.ClientEvent[source]#

Bases: BaseModel

Contains information returned from the upstream server.

Show JSON schema
{
   "title": "ClientEvent",
   "description": "Contains information returned from the upstream server.",
   "type": "object",
   "properties": {
      "prefix": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Prefix or origin information",
         "title": "Prefix"
      },
      "command": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Code or command for the event",
         "title": "Command"
      },
      "params": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Event or command parameters",
         "title": "Params"
      },
      "original": {
         "description": "Original received from server",
         "minLength": 1,
         "title": "Original",
         "type": "string"
      },
      "kind": {
         "default": "event",
         "description": "Dynamic field parsed from event",
         "enum": [
            "event",
            "chanmsg",
            "privmsg"
         ],
         "title": "Kind",
         "type": "string"
      },
      "isme": {
         "default": false,
         "description": "Indicates message is from client",
         "title": "Isme",
         "type": "boolean"
      },
      "hasme": {
         "default": false,
         "description": "Indicates message mentions client",
         "title": "Hasme",
         "type": "boolean"
      },
      "whome": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Current nickname of when received",
         "title": "Whome"
      },
      "author": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Author"
      },
      "recipient": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Recipient"
      },
      "message": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Message"
      }
   },
   "required": [
      "original"
   ]
}

Config:
  • extra: str = ignore

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Code or command for the event

Constraints:
  • min_length = 1

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

Indicates message mentions client

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

Indicates message is from client

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

Dynamic field parsed from event

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Original received from server

Constraints:
  • min_length = 1

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

Event or command parameters

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

Prefix or origin information

Constraints:
  • min_length = 1

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

Dynamic field parsed from event

Constraints:
  • min_length = 1

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

Current nickname of when received

Constraints:
  • min_length = 1

pydantic model enconnect.irc.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema
{
   "title": "ClientParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "port": {
         "default": 6697,
         "description": "Server address for connection",
         "maximum": 65535,
         "minimum": 1,
         "title": "Port",
         "type": "integer"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "operate": {
         "default": "normal",
         "description": "Method for server connection",
         "enum": [
            "normal",
            "service"
         ],
         "title": "Operate",
         "type": "string"
      },
      "nickname": {
         "default": "ircbot",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Nickname",
         "type": "string"
      },
      "username": {
         "default": "ircbot",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Username",
         "type": "string"
      },
      "realname": {
         "default": "Chatting Robie",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Realname",
         "type": "string"
      },
      "password": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Parameter for the integration",
         "title": "Password"
      },
      "servername": {
         "default": "services.invalid",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Servername",
         "type": "string"
      },
      "serverid": {
         "default": "42X",
         "description": "Unique identifier for services",
         "minLength": 1,
         "title": "Serverid",
         "type": "string"
      },
      "ssl_enable": {
         "default": true,
         "description": "Enable connection encryption",
         "title": "Ssl Enable",
         "type": "boolean"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "queue_size": {
         "default": 10000,
         "description": "Maximum size for queued events",
         "maximum": 1000000,
         "minimum": 1000,
         "title": "Queue Size",
         "type": "integer"
      }
   },
   "additionalProperties": false,
   "required": [
      "server"
   ]
}

Config:
  • extra: str = forbid

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Method for server connection

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Server address for connection

Constraints:
  • ge = 1

  • le = 65535

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

Maximum size for queued events

Constraints:
  • ge = 1000

  • le = 1000000

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Server address for connection

Constraints:
  • min_length = 1

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

Unique identifier for services

Constraints:
  • min_length = 1

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

Parameter for the integration

Constraints:
  • min_length = 1

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

Enable connection encryption

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

Verify the ceritifcate valid

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

Timeout connecting to server

Constraints:
  • ge = 1

  • le = 300

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

Parameter for the integration

Constraints:
  • min_length = 1