enconnect.mattermost package

Subpackages

• enconnect.mattermost.test package
  • Submodules
  • enconnect.mattermost.test.helpers module
    • MTMClientSocket
    • client_mtmsock()
  • enconnect.mattermost.test.test_client module
    • client()
    • test_Client()
    • test_Client_connect()
    • test_Client_cover()
  • enconnect.mattermost.test.test_models module
    • test_ClientEvent()
    • test_ClientEvent_cover()
  • Module contents
    • MTMClientSocket
    • client_mtmsock()

Submodules

enconnect.mattermost.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.mattermost.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: tuple[str, 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.

request(method: Literal['delete', 'get', 'post', 'patch', 'put'], path: str, params: dict[str, Any] | None = None, json: 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.

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

Returns:

Response from upstream request to the server.

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

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

error: 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].

original: Annotated[dict[str, Any], FieldInfo(annotation=NoneType, required=True, description='Original received from server', 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)])]

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

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

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

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.mattermost.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.mattermost.params.ClientParams(*, server: Annotated[str, MinLen(min_length=1)], port: Annotated[int, Ge(ge=1), Le(le=65535)] = 443, token: Annotated[str, MinLen(min_length=1)], teamid: 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.

model_config = {'extra': 'forbid'}

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

port: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=443, 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)])]

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

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

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

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.mattermost.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: tuple[str, 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.

request(method: Literal['delete', 'get', 'post', 'patch', 'put'], path: str, params: dict[str, Any] | None = None, json: 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.

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

Returns:

Response from upstream request to the server.

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

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

error: 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].

original: Annotated[dict[str, Any], FieldInfo(annotation=NoneType, required=True, description='Original received from server', 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)])]

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

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

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

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.mattermost.ClientParams(*, server: Annotated[str, MinLen(min_length=1)], port: Annotated[int, Ge(ge=1), Le(le=65535)] = 443, token: Annotated[str, MinLen(min_length=1)], teamid: 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.

model_config = {'extra': 'forbid'}

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

port: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=443, 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)])]

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

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

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

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