enconnect.discord package

Subpackages

• enconnect.discord.test package
  • Submodules
  • enconnect.discord.test.helpers module
    • DSCClientSocket
    • client_dscsock()
  • enconnect.discord.test.test_client module
    • client()
    • test_Client()
    • test_Client_connect()
    • test_Client_cover()
  • enconnect.discord.test.test_models module
    • test_ClientEvent()
    • test_ClientEvent_cover()
  • Module contents
    • DSCClientSocket
    • client_dscsock()

Submodules

enconnect.discord.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.discord.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.

get_channel(unique: str) → dict[str, Any]

Return the information about the object within Discord.

Parameters:

unique – Unique identifier to locate with Discord.

Returns:

Response from upstream request to the server.

get_guild(unique: str) → dict[str, Any]

Return the information about the object within Discord.

Parameters:

unique – Unique identifier to locate with Discord.

Returns:

Response from upstream request to the server.

interact_create(event: ClientEvent, **kwargs: Any) → None

Create the interaction using the raw event from server.

Parameters:

• event – Raw event received from the network peer.

• kwargs – Keyword arguments passed for downstream.

interact_update(event: ClientEvent, **kwargs: Any) → None

Update the interaction using the raw event from server.

Parameters:

• event – Raw event received from the network peer.

• kwargs – Keyword arguments passed for downstream.

message_create(channel: str, **kwargs: Any) → Response

Create the message using the unique Discord identifier.

Parameters:

• channel – Unique identifier in the Discord server.

• kwargs – Keyword arguments passed for downstream.

Returns:

Response from upstream request to the server.

message_delete(channel: str, unique: str) → Response

Delete the message using the unique Discord identifiers.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

Returns:

Response from upstream request to the server.

message_react(channel: str, unique: str, emoji: str) → Response

React to the message with the provided emoji identifier.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

• emoji – Unique identifier in the Discord server.

Returns:

Response from upstream request to the server.

message_update(channel: str, unique: str, **kwargs: Any) → Response

Update the message using the unique Discord identifiers.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

• payload – Message payload with the Discord syntax.

• kwargs – Keyword arguments passed for downstream.

Returns:

Response from upstream request to the server.

property mqueue: Queue[ClientEvent]

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property nickname: tuple[str, str] | None

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate(*, intents: int = 4609) → None

Operate the client and populate queue with the messages.

Parameters:

intents – Determine what content will be received.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

request(method: Literal['delete', 'get', 'post', 'patch', 'put'], path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, *, data: dict[str, Any] | None = None, files: dict[str, Any] | None = None, timeout: int | None = None) → Response

Return the response for upstream request to the server.

Parameters:

• method – Method for operation with the API server.

• path – Path for the location to upstream endpoint.

• params – Optional parameters included in request.

• json – Optional JSON payload included in request.

• data – Optional dict payload included in request.

• files – Optional file payload included in request.

• timeout – Timeout waiting for the server response. This will override the default client instantiated.

Returns:

Response from upstream request to the server.

slash_create(command: dict[str, Any], guild: str | None = None) → Response

Register the specified command using upstream endpoint.

Parameters:

• command – Parameters for registering the command.

• guild – Discord guild for registering the command.

Returns:

Response from upstream request to the server.

slash_delete(name: str, guild: str | None = None) → None

Remove specified slash command using upstream endpoint.

Parameters:

• name – Match of the name of the returned commands.

• guild – Discord guild where command is registred.

socket_recv() → dict[str, Any] | None

Return the content received from the socket connection.

Returns:

Content received from the socket connection.

socket_send(send: dict[str, Any]) → 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.discord.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.discord.models.ClientEvent(client: Client, event: dict[str, Any])

Bases: BaseModel

Contains information returned from the upstream server.

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

data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', 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].

opcode: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Type of operation performed', metadata=[Ge(ge=0)])]

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

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

seqno: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event number within squence', metadata=[Ge(ge=0)])]

type: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Type of event that occurred', metadata=[MinLen(min_length=1)])]

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

enconnect.discord.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.discord.params.ClientParams(*, appid: Annotated[str | None, MinLen(min_length=1)] = None, token: Annotated[str, MinLen(min_length=1)], timeout: Annotated[int, Ge(ge=1), Le(le=300)] = 30, ssl_verify: bool = True, ssl_capem: Annotated[str | None, MinLen(min_length=1)] = None, queue_size: Annotated[int, Ge(ge=1000), Le(le=1000000)] = 10000)

Bases: BaseModel

Process and validate the class configuration parameters.

appid: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional application identifier', metadata=[MinLen(min_length=1)])]

model_config = {'extra': 'forbid'}

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

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

ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]

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

token: Annotated[str, FieldInfo(annotation=NoneType, required=True, 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.discord.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.

get_channel(unique: str) → dict[str, Any]

Return the information about the object within Discord.

Parameters:

unique – Unique identifier to locate with Discord.

Returns:

Response from upstream request to the server.

get_guild(unique: str) → dict[str, Any]

Return the information about the object within Discord.

Parameters:

unique – Unique identifier to locate with Discord.

Returns:

Response from upstream request to the server.

interact_create(event: ClientEvent, **kwargs: Any) → None

Create the interaction using the raw event from server.

Parameters:

• event – Raw event received from the network peer.

• kwargs – Keyword arguments passed for downstream.

interact_update(event: ClientEvent, **kwargs: Any) → None

Update the interaction using the raw event from server.

Parameters:

• event – Raw event received from the network peer.

• kwargs – Keyword arguments passed for downstream.

message_create(channel: str, **kwargs: Any) → Response

Create the message using the unique Discord identifier.

Parameters:

• channel – Unique identifier in the Discord server.

• kwargs – Keyword arguments passed for downstream.

Returns:

Response from upstream request to the server.

message_delete(channel: str, unique: str) → Response

Delete the message using the unique Discord identifiers.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

Returns:

Response from upstream request to the server.

message_react(channel: str, unique: str, emoji: str) → Response

React to the message with the provided emoji identifier.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

• emoji – Unique identifier in the Discord server.

Returns:

Response from upstream request to the server.

message_update(channel: str, unique: str, **kwargs: Any) → Response

Update the message using the unique Discord identifiers.

Parameters:

• channel – Unique identifier in the Discord server.

• unique – Unique identifier in the Discord server.

• payload – Message payload with the Discord syntax.

• kwargs – Keyword arguments passed for downstream.

Returns:

Response from upstream request to the server.

property mqueue: Queue[ClientEvent]

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property nickname: tuple[str, str] | None

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

operate(*, intents: int = 4609) → None

Operate the client and populate queue with the messages.

Parameters:

intents – Determine what content will be received.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

request(method: Literal['delete', 'get', 'post', 'patch', 'put'], path: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, *, data: dict[str, Any] | None = None, files: dict[str, Any] | None = None, timeout: int | None = None) → Response

Return the response for upstream request to the server.

Parameters:

• method – Method for operation with the API server.

• path – Path for the location to upstream endpoint.

• params – Optional parameters included in request.

• json – Optional JSON payload included in request.

• data – Optional dict payload included in request.

• files – Optional file payload included in request.

• timeout – Timeout waiting for the server response. This will override the default client instantiated.

Returns:

Response from upstream request to the server.

slash_create(command: dict[str, Any], guild: str | None = None) → Response

Register the specified command using upstream endpoint.

Parameters:

• command – Parameters for registering the command.

• guild – Discord guild for registering the command.

Returns:

Response from upstream request to the server.

slash_delete(name: str, guild: str | None = None) → None

Remove specified slash command using upstream endpoint.

Parameters:

• name – Match of the name of the returned commands.

• guild – Discord guild where command is registred.

socket_recv() → dict[str, Any] | None

Return the content received from the socket connection.

Returns:

Content received from the socket connection.

socket_send(send: dict[str, Any]) → 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.discord.ClientEvent(client: Client, event: dict[str, Any])

Bases: BaseModel

Contains information returned from the upstream server.

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

data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', 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].

opcode: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Type of operation performed', metadata=[Ge(ge=0)])]

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

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

seqno: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event number within squence', metadata=[Ge(ge=0)])]

type: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Type of event that occurred', metadata=[MinLen(min_length=1)])]

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

class enconnect.discord.ClientParams(*, appid: Annotated[str | None, MinLen(min_length=1)] = None, token: Annotated[str, MinLen(min_length=1)], timeout: Annotated[int, Ge(ge=1), Le(le=300)] = 30, ssl_verify: bool = True, ssl_capem: Annotated[str | None, MinLen(min_length=1)] = None, queue_size: Annotated[int, Ge(ge=1000), Le(le=1000000)] = 10000)

Bases: BaseModel

Process and validate the class configuration parameters.

appid: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional application identifier', metadata=[MinLen(min_length=1)])]

model_config = {'extra': 'forbid'}

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

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

ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]

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

token: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Parameter for the integration', metadata=[MinLen(min_length=1)])]