enconnect.mattermost package#

Submodules#

enconnect.mattermost.client module#

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

Operate the client and populate queue with the messages.

stop() → None[source]#

Gracefully close the connection with the server socket.

socket_send(send: dict[str, Any]) → None[source]#

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

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

Return the content received from the socket connection.

Returns:

Content received from the socket connection.

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

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.

enconnect.mattermost.models module#

class enconnect.mattermost.models.ClientEvent[source]#

Bases: BaseModel

Contains information returned from the upstream server.

Fields#

Field

Type

Required

Default

Constraints

author

tuple[str, str] | None

No

None

min_length=1

broadcast

dict[str, Any] | None

No

None

min_length=1

data

dict[str, Any] | None

No

None

min_length=1

error

dict[str, Any] | 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

dict[str, Any]

Yes

min_length=1

recipient

str | None

No

None

min_length=1

seqno

int | None

No

None

ge=0

seqre

int | None

No

None

ge=0

status

str | None

No

None

min_length=1

type

str | None

No

None

min_length=1

whome

tuple[str, str] | None

No

None

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)])]#
data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', 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)])]#
seqno: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event 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)])]#
error: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])]#
seqre: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Reply number within squence', metadata=[Ge(ge=0)])]#
original: Annotated[dict[str, Any], 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[tuple[str, str] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Current nickname of when received', metadata=[MinLen(min_length=1)])]#
author: Annotated[tuple[str, 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.mattermost.params module#

class enconnect.mattermost.params.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#

Field

Type

Required

Default

Constraints

port

int

No

443

ge=1, le=65535

queue_size

int

No

10000

ge=1000, le=1000000

server

str

Yes

min_length=1

ssl_capem

str | None

No

None

min_length=1

ssl_verify

bool

No

True

teamid

str

Yes

min_length=1

timeout

int

No

30

ge=1, le=300

token

str

Yes

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=443, description='Server address for connection', metadata=[Ge(ge=1), Le(le=65535)])]#
token: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Parameter for the integration', metadata=[MinLen(min_length=1)])]#
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)])]#
ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]#
ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]#
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.mattermost.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 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: tuple[str, 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[source]#

Operate the client and populate queue with the messages.

stop() → None[source]#

Gracefully close the connection with the server socket.

socket_send(send: dict[str, Any]) → None[source]#

Transmit provided content through the socket connection.

Parameters:

send – Content which will be sent through socket.

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

Return the content received from the socket connection.

Returns:

Content received from the socket connection.

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

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.

class enconnect.mattermost.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#

Field

Type

Required

Default

Constraints

port

int

No

443

ge=1, le=65535

queue_size

int

No

10000

ge=1000, le=1000000

server

str

Yes

min_length=1

ssl_capem

str | None

No

None

min_length=1

ssl_verify

bool

No

True

teamid

str

Yes

min_length=1

timeout

int

No

30

ge=1, le=300

token

str

Yes

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=443, description='Server address for connection', metadata=[Ge(ge=1), Le(le=65535)])]#
token: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Parameter for the integration', metadata=[MinLen(min_length=1)])]#
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)])]#
ssl_verify: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Verify the ceritifcate valid')]#
ssl_capem: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Verify the ceritifcate valid', metadata=[MinLen(min_length=1)])]#
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.mattermost.ClientEvent[source]#

Bases: BaseModel

Contains information returned from the upstream server.

Fields#

Field

Type

Required

Default

Constraints

author

tuple[str, str] | None

No

None

min_length=1

broadcast

dict[str, Any] | None

No

None

min_length=1

data

dict[str, Any] | None

No

None

min_length=1

error

dict[str, Any] | 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

dict[str, Any]

Yes

min_length=1

recipient

str | None

No

None

min_length=1

seqno

int | None

No

None

ge=0

seqre

int | None

No

None

ge=0

status

str | None

No

None

min_length=1

type

str | None

No

None

min_length=1

whome

tuple[str, str] | None

No

None

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)])]#
data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', 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)])]#
seqno: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Event 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)])]#
error: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])]#
seqre: Annotated[int | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Reply number within squence', metadata=[Ge(ge=0)])]#
original: Annotated[dict[str, Any], 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[tuple[str, str] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Current nickname of when received', metadata=[MinLen(min_length=1)])]#
author: Annotated[tuple[str, 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)])]#