Enasis Network Remote Connect#

Documentation Welcome#

Welcome to the project documentation. Below you will find the most common functions and classes within the library, but you may click your way further into the project from there. You can also view the source code.

Discord#

class enconnect.discord.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(*, intents: int = 4609) → None[source]#

Operate the client and populate queue with the messages.

Parameters:: intents – Determine what content will be received.

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, *, data: dict[str, Any] | None = None, files: 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.
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[source]#

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

Remove specified slash command using upstream endpoint.

Parameters:

name – Match of the name of the returned commands.
guild – Discord guild where command is registred.

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

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_channel(unique: str) → dict[str, Any][source]#

Return the information about the object within Discord.

Parameters:: unique – Unique identifier to locate with Discord.
Returns:: Response from upstream request to the server.

message_create(channel: str, **kwargs: Any) → Response[source]#

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_update(channel: str, unique: str, **kwargs: Any) → Response[source]#

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.

message_delete(channel: str, unique: str) → Response[source]#

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

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.

interact_create(event: ClientEvent, **kwargs: Any) → None[source]#

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

Update the interaction using the raw event from server.

Parameters:

event – Raw event received from the network peer.
kwargs – Keyword arguments passed for downstream.

class enconnect.discord.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#
Field	Type	Required	Default	Constraints
`appid`	`str` \| `None`	No	`None`	min_length=1
`queue_size`	`int`	No	`10000`	ge=1000, le=1000000
`ssl_capem`	`str` \| `None`	No	`None`	min_length=1
`ssl_verify`	`bool`	No	`True`
`timeout`	`int`	No	`30`	ge=1, le=300
`token`	`str`	Yes		min_length=1

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

token: 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.discord.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
`data`	`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
`opcode`	`int`	Yes		ge=0
`original`	`dict`[`str`, `Any`]	Yes		min_length=1
`recipient`	`tuple`[`str` \| `None`, `str`] \| `None`	No	`None`	min_length=1
`seqno`	`int` \| `None`	No	`None`	ge=0
`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)])]#

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

data: 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)])]#

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[tuple[str | None, 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)])]#

IRC#

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 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: 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: str) → None[source]#

Transmit provided content through the socket connection.

Parameters:: send – Content which will be sent through socket.

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.

class enconnect.irc.ClientParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#
Field	Type	Required	Default	Constraints
`nickname`	`str`	No	`'ircbot'`	min_length=1
`operate`	`Literal`[‘normal’, ‘service’]	No	`'normal'`
`password`	`str` \| `None`	No	`None`	min_length=1
`port`	`int`	No	`6697`	ge=1, le=65535
`queue_size`	`int`	No	`10000`	ge=1000, le=1000000
`realname`	`str`	No	`'Chatting Robie'`	min_length=1
`server`	`str`	Yes		min_length=1
`serverid`	`str`	No	`'42X'`	min_length=1
`servername`	`str`	No	`'services.invalid'`	min_length=1
`ssl_enable`	`bool`	No	`True`
`ssl_verify`	`bool`	No	`True`
`timeout`	`int`	No	`30`	ge=1, le=300
`username`	`str`	No	`'ircbot'`	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=6697, description='Server address for connection', metadata=[Ge(ge=1), Le(le=65535)])]#

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

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

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

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

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

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

serverid: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='42X', description='Unique identifier for services', 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')]#

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

Bases: BaseModel

Contains information returned from the upstream server.

Fields#
Field	Type	Required	Default	Constraints
`author`	`str` \| `None`	No	`None`	min_length=1
`command`	`str` \| `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`	`str`	Yes		min_length=1
`params`	`str` \| `None`	No	`None`
`prefix`	`str` \| `None`	No	`None`	min_length=1
`recipient`	`str` \| `None`	No	`None`	min_length=1
`whome`	`str` \| `None`	No	`None`	min_length=1

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

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

original: Annotated[str, 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[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Current nickname of when received', metadata=[MinLen(min_length=1)])]#

author: Annotated[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)])]#

Mattermost#

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

Philips#

Hubitat#

Ubiquiti#

Instagram#

Reddit#

class enconnect.reddit.Reddit(params: RedditParams)[source]#

Bases: object

Interact with the cloud service API with various methods.

Parameters:: params – Parameters used to instantiate the class.

property params: RedditParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

property client: HTTPClient#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property token: str | None#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

request_token_block() → str[source]#

Establish new session obtaining token for authorization.

Returns:: Access token used with authenticated requests.

async request_token_async() → str[source]#

Establish new session obtaining token for authorization.

Returns:: Access token used with authenticated requests.

request_block(method: Literal['get', 'post'], path: str, params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, *, httpauth: tuple[str, str] | 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.
data – Optional dict payload included in request.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

async request_async(method: Literal['get', 'post'], path: str, params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, *, httpauth: tuple[str, str] | 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.
data – Optional dict payload included in request.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

listing(unique: str) → RedditListing[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

listing_block(unique: str) → RedditListing[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

async listing_async(unique: str) → RedditListing[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

latest(subred: str, params: dict[str, Any] | None = None) → list[RedditListing][source]#

Return the new items within the provided subreddit path.

Parameters:

subred – Path to subreddit containing the content.
params – Optional parameters included in request.

Returns:

New items within the provided subreddit path.

latest_block(subred: str, params: dict[str, Any] | None = None) → list[RedditListing][source]#

Return the new items within the provided subreddit path.

Parameters:

subred – Path to subreddit containing the content.
params – Optional parameters included in request.

Returns:

New items within the provided subreddit path.

async latest_async(subred: str, params: dict[str, Any] | None = None) → list[RedditListing][source]#

Return the new items within the provided subreddit path.

Parameters:

subred – Path to subreddit containing the content.
params – Optional parameters included in request.

Returns:

New items within the provided subreddit path.

class enconnect.reddit.RedditParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#
Field	Type	Required	Default	Constraints
`client`	`str`	Yes		min_length=1
`password`	`str`	Yes		min_length=1
`secret`	`str`	Yes		min_length=1
`ssl_capem`	`str` \| `None`	No	`None`	min_length=1
`ssl_verify`	`bool`	No	`True`
`timeout`	`int`	No	`30`	ge=1, le=300
`useragent`	`str`	Yes		min_length=1
`username`	`str`	Yes		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)])]#

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

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

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

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

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

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

class enconnect.reddit.RedditListing[source]#

Bases: BaseModel

Contains information returned from the upstream response.

Fields#
Field	Type	Required	Default
`archived`	`bool`	Yes
`author`	`str`	Yes
`created`	`int`	Yes
`domain`	`str`	Yes
`edited`	`bool` \| `float`	Yes
`id`	`str`	Yes
`medias`	`list`[`str`] \| `None`	No	`None`
`name`	`str`	Yes
`permalink`	`str`	Yes
`pinned`	`bool`	Yes
`score`	`int`	Yes
`selftext`	`str` \| `None`	No	`None`
`stickied`	`bool`	Yes
`thumbnail`	`str`	Yes
`title`	`str`	Yes
`url`	`str`	Yes
`url_dest`	`str` \| `None`	No	`None`
`vote_downs`	`int`	Yes
`vote_ups`	`int`	Yes

name: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

id: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

created: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

title: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

selftext: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

author: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

url: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

permalink: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

thumbnail: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

url_dest: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

domain: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

medias: Annotated[list[str] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

pinned: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

edited: Annotated[bool | float, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

stickied: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

archived: Annotated[bool, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

vote_downs: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

vote_ups: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

score: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

YouTube#

class enconnect.youtube.YouTube(params: YouTubeParams)[source]#

Bases: object

Interact with the cloud service API with various methods.

Parameters:: params – Parameters used to instantiate the class.

property params: YouTubeParams#

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

property client: HTTPClient#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

request_block(method: Literal['get'], path: str, params: dict[str, Any] | 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.

Returns:

Response from upstream request to the server.

async request_async(method: Literal['get'], path: str, params: dict[str, Any] | 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.

Returns:

Response from upstream request to the server.

search(params: dict[str, Any] | None = None) → list[YouTubeResult][source]#

Return the results from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

search_block(params: dict[str, Any] | None = None) → list[YouTubeResult][source]#

Return the results from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

async search_async(params: dict[str, Any] | None = None) → list[YouTubeResult][source]#

Return the results from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

videos(params: dict[str, Any] | None = None) → list[YouTubeVideo][source]#

Return the videos from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

videos_block(params: dict[str, Any] | None = None) → list[YouTubeVideo][source]#

Return the videos from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

async videos_async(params: dict[str, Any] | None = None) → list[YouTubeVideo][source]#

Return the videos from the provided search parameters.

Parameters:: params – Optional parameters included in request.
Returns:: Results from the provided search parameters.

video(unique: str) → YouTubeVideo[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

video_block(unique: str) → YouTubeVideo[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

async video_async(unique: str) → YouTubeVideo[source]#

Return the specific content within the social platform.

Parameters:: unique – Unique identifier within social platform.
Returns:: Specific content within the social platform.

class enconnect.youtube.YouTubeParams[source]#

Bases: BaseModel

Process and validate the class configuration parameters.

Fields#
Field	Type	Required	Default	Constraints
`ssl_capem`	`str` \| `None`	No	`None`	min_length=1
`ssl_verify`	`bool`	No	`True`
`timeout`	`int`	No	`30`	ge=1, le=300
`token`	`str`	Yes		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)])]#

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

class enconnect.youtube.YouTubeResult[source]#

Bases: BaseModel

Contains information returned from the upstream response.

Fields#
Field	Type	Required	Default
`about`	`str` \| `None`	No	`None`
`channel`	`str` \| `None`	No	`None`
`channel_title`	`str` \| `None`	No	`None`
`kind`	`Literal`[‘channel’, ‘playlist’, ‘video’]	Yes
`playlist`	`str` \| `None`	No	`None`
`published`	`str`	Yes
`thumbnail`	`str`	Yes
`title`	`str`	Yes
`video`	`str` \| `None`	No	`None`

kind: Annotated[Literal['channel', 'playlist', 'video'], FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

channel: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

playlist: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

video: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

title: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

about: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

channel_title: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

thumbnail: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

published: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

class enconnect.youtube.YouTubeVideo[source]#

Bases: BaseModel

Contains information returned from the upstream response.

Fields#
Field	Type	Required	Default
`about`	`str` \| `None`	No	`None`
`channel`	`str`	Yes
`kind`	`Literal`[‘channel’, ‘playlist’, ‘video’]	Yes
`published`	`str`	Yes
`thumbnail`	`str`	Yes
`title`	`str`	Yes
`video`	`str`	Yes

kind: Annotated[Literal['channel', 'playlist', 'video'], FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

channel: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

video: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

title: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

about: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')]#

thumbnail: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

published: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Value from the server response')]#

General Utilities#

class enconnect.utils.HTTPClient(timeout: int = 30, headers: dict[str, str] | None = None, verify: SSLContext | str | bool = True, capem: str | None = None, httpauth: tuple[str, str] | None = None, retry: int = 3, backoff: float = 3.0, states: set[int] = {429})[source]#

Bases: object

Interact with the upstream server in blocking or async.

Parameters:

timeout – Timeout waiting for the server response.
headers – Optional headers to include in requests.
verify – Require valid certificate from the server.
capem – Optional path to the certificate authority.
httpauth – Optional information for authentication.
retry – How many attempts are made with the server.
backoff – Backoff backoff if encountered retries.
states – Which states will be retried with backoff.

property timeout: int#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property httpauth: tuple[str, str] | None#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property headers: dict[str, str] | None#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property verify: SSLContext | str | bool#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property capem: str | None#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property retry: int#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property backoff: float#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property states: set[int]#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property client_block: Client#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property client_async: AsyncClient#

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

request_block(method: Literal['delete', 'get', 'post', 'patch', 'put'], location: 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, headers: dict[str, str] | None = None, httpauth: tuple[str, str] | None = None) → Response[source]#

Return the response for upstream request to the server.

Parameters:

method – Method for operation with the API server.
location – Location with path for server request.
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.
headers – Optional headers to include in requests.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

async request_async(method: Literal['delete', 'get', 'post', 'patch', 'put'], location: 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, headers: dict[str, str] | None = None, httpauth: tuple[str, str] | None = None) → Response[source]#

Return the response for upstream request to the server.

Parameters:

method – Method for operation with the API server.
location – Location with path for server request.
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.
headers – Optional headers to include in requests.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

stream_block(method: Literal['delete', 'get', 'post', 'patch', 'put'], location: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, *, data: dict[str, Any] | None = None, timeout: int | None = None, headers: dict[str, str] | None = None, httpauth: tuple[str, str] | None = None) → Iterator[str][source]#

Return the response for upstream request to the server.

Parameters:

method – Method for operation with the API server.
location – Location with path for server request.
params – Optional parameters included in request.
json – Optional JSON payload included in request.
data – Optional dict payload included in request.
timeout – Timeout waiting for the server response.
headers – Optional headers to include in requests.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

async stream_async(method: Literal['delete', 'get', 'post', 'patch', 'put'], location: str, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, *, data: dict[str, Any] | None = None, timeout: int | None = None, headers: dict[str, str] | None = None, httpauth: tuple[str, str] | None = None) → AsyncIterator[str][source]#

Return the response for upstream request to the server.

Parameters:

method – Method for operation with the API server.
location – Location with path for server request.
params – Optional parameters included in request.
json – Optional JSON payload included in request.
data – Optional dict payload included in request.
timeout – Timeout waiting for the server response.
headers – Optional headers to include in requests.
httpauth – Optional information for authentication.

Returns:

Response from upstream request to the server.

enconnect.utils.dumlog(**kwargs: Any) → None[source]#

Capture and drop messages for would be provided logger.

Parameters:: kwargs – Keyword arguments which would be passed.