Enasis Network Remote Connect#

Discord Parameters#

pydantic model enconnect.discord.ClientParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "ClientParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "appid": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Optional application identifier",
         "title": "Appid"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      },
      "queue_size": {
         "default": 10000,
         "description": "Maximum size for queued events",
         "maximum": 1000000,
         "minimum": 1000,
         "title": "Queue Size",
         "type": "integer"
      }
   },
   "additionalProperties": false,
   "required": [
      "token"
   ]
}

Config:

extra: str = forbid

Fields:

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

Optional application identifier

Constraints:

min_length = 1

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

Maximum size for queued events

Constraints:

ge = 1000
le = 1000000

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Discord Client#

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 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][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_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.

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.

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_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.

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.

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

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[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.

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

Return the content received from the socket connection.

Returns:: Content received from the socket connection.

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

Transmit provided content through the socket connection.

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

stop() → None[source]: Gracefully close the connection with the server socket.

pydantic model enconnect.discord.ClientEvent[source]

Bases: BaseModel

Contains information returned from the upstream server.

Show JSON schema

{
   "title": "ClientEvent",
   "description": "Contains information returned from the upstream server.",
   "type": "object",
   "properties": {
      "type": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Type of event that occurred",
         "title": "Type"
      },
      "opcode": {
         "description": "Type of operation performed",
         "minimum": 0,
         "title": "Opcode",
         "type": "integer"
      },
      "data": {
         "anyOf": [
            {
               "additionalProperties": true,
               "minProperties": 1,
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Payload with the event data",
         "title": "Data"
      },
      "seqno": {
         "anyOf": [
            {
               "minimum": 0,
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Event number within squence",
         "title": "Seqno"
      },
      "original": {
         "additionalProperties": true,
         "description": "Original received from server",
         "minProperties": 1,
         "title": "Original",
         "type": "object"
      },
      "kind": {
         "default": "event",
         "description": "Dynamic field parsed from event",
         "enum": [
            "event",
            "chanmsg",
            "privmsg"
         ],
         "title": "Kind",
         "type": "string"
      },
      "isme": {
         "default": false,
         "description": "Indicates message is from client",
         "title": "Isme",
         "type": "boolean"
      },
      "hasme": {
         "default": false,
         "description": "Indicates message mentions client",
         "title": "Hasme",
         "type": "boolean"
      },
      "whome": {
         "anyOf": [
            {
               "maxItems": 2,
               "minItems": 1,
               "prefixItems": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "string"
                  }
               ],
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Current nickname of when received",
         "title": "Whome"
      },
      "author": {
         "anyOf": [
            {
               "maxItems": 2,
               "minItems": 1,
               "prefixItems": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "string"
                  }
               ],
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Author"
      },
      "recipient": {
         "anyOf": [
            {
               "maxItems": 2,
               "minItems": 1,
               "prefixItems": [
                  {
                     "anyOf": [
                        {
                           "type": "string"
                        },
                        {
                           "type": "null"
                        }
                     ]
                  },
                  {
                     "type": "string"
                  }
               ],
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Recipient"
      },
      "message": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Message"
      }
   },
   "required": [
      "opcode",
      "original"
   ]
}

Config:

extra: str = ignore

Fields:

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

Dynamic field parsed from event

Constraints:

min_length = 1

field data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])] = None

Payload with the event data

Constraints:

min_length = 1

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

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

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

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Type of operation performed

Constraints:

ge = 0

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

Original received from server

Constraints:

min_length = 1

field 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)])] = None

Dynamic field parsed from event

Constraints:

min_length = 1

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

Event number within squence

Constraints:

ge = 0

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

Type of event that occurred

Constraints:

min_length = 1

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

Current nickname of when received

Constraints:

min_length = 1

Instagram Parameters#

pydantic model enconnect.instagram.InstagramParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "InstagramParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "token"
   ]
}

Config:

extra: str = forbid

Fields:

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Instagram Client#

class enconnect.instagram.Instagram(params: InstagramParams)[source]

Bases: object

Interact with the cloud service API with various methods.

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

property client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

latest() → list[InstagramMedia][source]

Return the posts from the account associated with user.

Returns:: Posts from the account associated with user.

async latest_async() → list[InstagramMedia][source]

Return the posts from the account associated with user.

Returns:: Posts from the account associated with user.

latest_block() → list[InstagramMedia][source]

Return the posts from the account associated with user.

Returns:: Posts from the account associated with user.

async media_async(unique: int | str) → InstagramMedia[source]

Return the specific content within the social platform.

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

media_block(unique: int | str) → InstagramMedia[source]

Return the specific content within the social platform.

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

property params: InstagramParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

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.

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.

pydantic model enconnect.instagram.InstagramMedia[source]

Bases: BaseModel

Contains information returned from the upstream response.

Show JSON schema

{
   "title": "InstagramMedia",
   "description": "Contains information returned from the upstream response.",
   "type": "object",
   "properties": {
      "caption": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Caption"
      },
      "id": {
         "description": "Value from the server response",
         "title": "Id",
         "type": "string"
      },
      "shared": {
         "anyOf": [
            {
               "type": "boolean"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Shared"
      },
      "type": {
         "description": "Value from the server response",
         "title": "Type",
         "type": "string"
      },
      "location": {
         "description": "Value from the server response",
         "title": "Location",
         "type": "string"
      },
      "permalink": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Permalink"
      },
      "thumbnail": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Thumbnail"
      },
      "timestamp": {
         "description": "Value from the server response",
         "title": "Timestamp",
         "type": "string"
      },
      "username": {
         "description": "Value from the server response",
         "title": "Username",
         "type": "string"
      }
   },
   "additionalProperties": true,
   "required": [
      "id",
      "type",
      "location",
      "timestamp",
      "username"
   ]
}

Config:

extra: str = allow

Fields:

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

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

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

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

field shared: Annotated[bool | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Value from the server response')] = None: Value from the server response

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

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

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

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

IRC Parameters#

pydantic model enconnect.irc.ClientParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

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

Config:

extra: str = forbid

Fields:

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

Parameter for the integration

Constraints:

min_length = 1

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

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

Parameter for the integration

Constraints:

min_length = 1

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

Server address for connection

Constraints:

ge = 1
le = 65535

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

Maximum size for queued events

Constraints:

ge = 1000
le = 1000000

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

Parameter for the integration

Constraints:

min_length = 1

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

Server address for connection

Constraints:

min_length = 1

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

Unique identifier for services

Constraints:

min_length = 1

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

Parameter for the integration

Constraints:

min_length = 1

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

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

IRC Client#

class enconnect.irc.Client(params: ClientParams, logger: Callable[[...], None] | None = None)[source]

Bases: object

Establish and maintain connection with the chat service.

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

property canceled: bool

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property connected: bool

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property mqueue: Queue[ClientEvent]

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property nickname: str | None

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

operate() → None[source]: Operate the client and populate queue with the messages.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

socket_recv() → Iterator[str][source]

Return the content received from the socket connection.

Note

Method could be further optimized using an internal buffer within the class after receiving more bytes.

Returns:: Content received from the socket connection.

socket_send(send: str) → None[source]

Transmit provided content through the socket connection.

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

stop() → None[source]: Gracefully close the connection with the server socket.

pydantic model enconnect.irc.ClientEvent[source]

Bases: BaseModel

Contains information returned from the upstream server.

Show JSON schema

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

Config:

extra: str = ignore

Fields:

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Code or command for the event

Constraints:

min_length = 1

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

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

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

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Original received from server

Constraints:

min_length = 1

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

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

Prefix or origin information

Constraints:

min_length = 1

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Current nickname of when received

Constraints:

min_length = 1

Mattermost Parameters#

pydantic model enconnect.mattermost.ClientParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "ClientParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "port": {
         "default": 443,
         "description": "Server address for connection",
         "maximum": 65535,
         "minimum": 1,
         "title": "Port",
         "type": "integer"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "teamid": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Teamid",
         "type": "string"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      },
      "queue_size": {
         "default": 10000,
         "description": "Maximum size for queued events",
         "maximum": 1000000,
         "minimum": 1000,
         "title": "Queue Size",
         "type": "integer"
      }
   },
   "additionalProperties": false,
   "required": [
      "server",
      "token",
      "teamid"
   ]
}

Config:

extra: str = forbid

Fields:

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

Server address for connection

Constraints:

ge = 1
le = 65535

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

Maximum size for queued events

Constraints:

ge = 1000
le = 1000000

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

Server address for connection

Constraints:

min_length = 1

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Parameter for the integration

Constraints:

min_length = 1

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Mattermost Client#

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 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[source]: Operate the client and populate queue with the messages.

property params: ClientParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

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.

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

Return the content received from the socket connection.

Returns:: Content received from the socket connection.

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

Transmit provided content through the socket connection.

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

stop() → None[source]: Gracefully close the connection with the server socket.

pydantic model enconnect.mattermost.ClientEvent[source]

Bases: BaseModel

Contains information returned from the upstream server.

Show JSON schema

{
   "title": "ClientEvent",
   "description": "Contains information returned from the upstream server.",
   "type": "object",
   "properties": {
      "type": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Type of event that occurred",
         "title": "Type"
      },
      "data": {
         "anyOf": [
            {
               "additionalProperties": true,
               "minProperties": 1,
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Payload with the event data",
         "title": "Data"
      },
      "broadcast": {
         "anyOf": [
            {
               "additionalProperties": true,
               "minProperties": 1,
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Payload with the event data",
         "title": "Broadcast"
      },
      "seqno": {
         "anyOf": [
            {
               "minimum": 0,
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Event number within squence",
         "title": "Seqno"
      },
      "status": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Type of event that occurred",
         "title": "Status"
      },
      "error": {
         "anyOf": [
            {
               "additionalProperties": true,
               "minProperties": 1,
               "type": "object"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Payload with the event data",
         "title": "Error"
      },
      "seqre": {
         "anyOf": [
            {
               "minimum": 0,
               "type": "integer"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Reply number within squence",
         "title": "Seqre"
      },
      "original": {
         "additionalProperties": true,
         "description": "Original received from server",
         "minProperties": 1,
         "title": "Original",
         "type": "object"
      },
      "kind": {
         "default": "event",
         "description": "Dynamic field parsed from event",
         "enum": [
            "event",
            "chanmsg",
            "privmsg"
         ],
         "title": "Kind",
         "type": "string"
      },
      "isme": {
         "default": false,
         "description": "Indicates message is from client",
         "title": "Isme",
         "type": "boolean"
      },
      "hasme": {
         "default": false,
         "description": "Indicates message mentions client",
         "title": "Hasme",
         "type": "boolean"
      },
      "whome": {
         "anyOf": [
            {
               "maxItems": 2,
               "minItems": 1,
               "prefixItems": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "string"
                  }
               ],
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Current nickname of when received",
         "title": "Whome"
      },
      "author": {
         "anyOf": [
            {
               "maxItems": 2,
               "minItems": 1,
               "prefixItems": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "string"
                  }
               ],
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Author"
      },
      "recipient": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Recipient"
      },
      "message": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Dynamic field parsed from event",
         "title": "Message"
      }
   },
   "required": [
      "original"
   ]
}

Config:

extra: str = ignore

Fields:

Dynamic field parsed from event

Constraints:

min_length = 1

field broadcast: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])] = None

Payload with the event data

Constraints:

min_length = 1

field data: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])] = None

Payload with the event data

Constraints:

min_length = 1

field error: Annotated[dict[str, Any] | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Payload with the event data', metadata=[MinLen(min_length=1)])] = None

Payload with the event data

Constraints:

min_length = 1

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

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

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

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Original received from server

Constraints:

min_length = 1

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

Dynamic field parsed from event

Constraints:

min_length = 1

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

Event number within squence

Constraints:

ge = 0

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

Reply number within squence

Constraints:

ge = 0

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

Type of event that occurred

Constraints:

min_length = 1

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

Type of event that occurred

Constraints:

min_length = 1

Current nickname of when received

Constraints:

min_length = 1

Philips Hue Parameters#

pydantic model enconnect.philips.BridgeParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "BridgeParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "server",
      "token"
   ]
}

Config:

extra: str = forbid

Fields:

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

Server address for connection

Constraints:

min_length = 1

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Philips Hue Client#

class enconnect.philips.Bridge(params: BridgeParams)[source]

Bases: object

Interact with the local service API with various methods.

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

property client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

async events_async(timeout: int | None = None) → AsyncIterator[dict[str, Any]][source]

Return the response for upstream request to the server.

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

events_block(timeout: int | None = None) → Iterator[dict[str, Any]][source]

Return the response for upstream request to the server.

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

property params: BridgeParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

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.

Hubitat Elevation Parameters#

pydantic model enconnect.hubitat.BridgeParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "BridgeParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "appid": {
         "description": "Parameter for the integration",
         "minimum": 0,
         "title": "Appid",
         "type": "integer"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "ssl_verify": {
         "default": false,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "server",
      "appid",
      "token"
   ]
}

Config:

extra: str = forbid

Fields:

field appid: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Parameter for the integration', metadata=[Ge(ge=0)])] [Required]

Parameter for the integration

Constraints:

ge = 0

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

Server address for connection

Constraints:

min_length = 1

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Hubitat Elevation Client#

class enconnect.hubitat.Bridge(params: BridgeParams)[source]

Bases: object

Interact with the local service API with various methods.

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

property client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property params: BridgeParams

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, *, 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.
timeout – Timeout waiting for the server response. This will override the default client instantiated.

Returns:

Response from upstream request to the server.

Reddit Parameters#

pydantic model enconnect.reddit.RedditParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "RedditParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "username": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Username",
         "type": "string"
      },
      "password": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Password",
         "type": "string"
      },
      "client": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Client",
         "type": "string"
      },
      "secret": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Secret",
         "type": "string"
      },
      "useragent": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Useragent",
         "type": "string"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "username",
      "password",
      "client",
      "secret",
      "useragent"
   ]
}

Config:

extra: str = forbid

Fields:

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

Parameter for the integration

Constraints:

min_length = 1

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

Parameter for the integration

Constraints:

min_length = 1

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

Parameter for the integration

Constraints:

min_length = 1

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

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

Parameter for the integration

Constraints:

min_length = 1

Reddit Client#

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 client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

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.

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.

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.

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.

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.

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.

property params: RedditParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

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.

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_token_async() → str[source]

Establish new session obtaining token for authorization.

Returns:: Access token used with authenticated requests.

request_token_block() → str[source]

Establish new session obtaining token for authorization.

Returns:: Access token used with authenticated requests.

property token: str | None

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

pydantic model enconnect.reddit.RedditListing[source]

Bases: BaseModel

Contains information returned from the upstream response.

Show JSON schema

{
   "title": "RedditListing",
   "description": "Contains information returned from the upstream response.",
   "type": "object",
   "properties": {
      "name": {
         "description": "Value from the server response",
         "title": "Name",
         "type": "string"
      },
      "id": {
         "description": "Value from the server response",
         "title": "Id",
         "type": "string"
      },
      "created": {
         "description": "Value from the server response",
         "title": "Created",
         "type": "integer"
      },
      "title": {
         "description": "Value from the server response",
         "title": "Title",
         "type": "string"
      },
      "selftext": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Selftext"
      },
      "author": {
         "description": "Value from the server response",
         "title": "Author",
         "type": "string"
      },
      "url": {
         "description": "Value from the server response",
         "title": "Url",
         "type": "string"
      },
      "permalink": {
         "description": "Value from the server response",
         "title": "Permalink",
         "type": "string"
      },
      "thumbnail": {
         "description": "Value from the server response",
         "title": "Thumbnail",
         "type": "string"
      },
      "url_dest": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Url Dest"
      },
      "domain": {
         "description": "Value from the server response",
         "title": "Domain",
         "type": "string"
      },
      "medias": {
         "anyOf": [
            {
               "items": {
                  "type": "string"
               },
               "type": "array"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Medias"
      },
      "pinned": {
         "description": "Value from the server response",
         "title": "Pinned",
         "type": "boolean"
      },
      "edited": {
         "anyOf": [
            {
               "type": "boolean"
            },
            {
               "type": "number"
            }
         ],
         "description": "Value from the server response",
         "title": "Edited"
      },
      "stickied": {
         "description": "Value from the server response",
         "title": "Stickied",
         "type": "boolean"
      },
      "archived": {
         "description": "Value from the server response",
         "title": "Archived",
         "type": "boolean"
      },
      "vote_downs": {
         "description": "Value from the server response",
         "title": "Vote Downs",
         "type": "integer"
      },
      "vote_ups": {
         "description": "Value from the server response",
         "title": "Vote Ups",
         "type": "integer"
      },
      "score": {
         "description": "Value from the server response",
         "title": "Score",
         "type": "integer"
      }
   },
   "required": [
      "name",
      "id",
      "created",
      "title",
      "author",
      "url",
      "permalink",
      "thumbnail",
      "domain",
      "pinned",
      "edited",
      "stickied",
      "archived",
      "vote_downs",
      "vote_ups",
      "score"
   ]
}

Config:

extra: str = ignore

Fields:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Ubiquiti Router Parameters#

pydantic model enconnect.ubiquiti.RouterParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "RouterParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "server": {
         "description": "Server address for connection",
         "minLength": 1,
         "title": "Server",
         "type": "string"
      },
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "username": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Username",
         "type": "string"
      },
      "password": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Password",
         "type": "string"
      },
      "site": {
         "default": "default",
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Site",
         "type": "string"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "server",
      "username",
      "password"
   ]
}

Config:

extra: str = forbid

Fields:

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

Parameter for the integration

Constraints:

min_length = 1

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

Server address for connection

Constraints:

min_length = 1

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

Parameter for the integration

Constraints:

min_length = 1

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

Ubiquiti Router Client#

class enconnect.ubiquiti.Router(params: RouterParams)[source]

Bases: object

Interact with the local service API with various methods.

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

property client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property params: RouterParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

reqroxy(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.

Note

Most endpoints on Ubiquiti routers are prefixed with not only the site but also /proxy/network/api. The method will handle concatenating the values for you. It will also handle authenticating if not already.

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.

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.

request_cookie() → Response[source]

Establish new session obtaining cookie for authorization.

Returns:: Response from upstream request to the server.

YouTube Parameters#

pydantic model enconnect.youtube.YouTubeParams[source]

Bases: BaseModel

Process and validate the class configuration parameters.

Show JSON schema

{
   "title": "YouTubeParams",
   "description": "Process and validate the class configuration parameters.",
   "type": "object",
   "properties": {
      "timeout": {
         "default": 30,
         "description": "Timeout connecting to server",
         "maximum": 300,
         "minimum": 1,
         "title": "Timeout",
         "type": "integer"
      },
      "token": {
         "description": "Parameter for the integration",
         "minLength": 1,
         "title": "Token",
         "type": "string"
      },
      "ssl_verify": {
         "default": true,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Verify",
         "type": "boolean"
      },
      "ssl_capem": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Verify the ceritifcate valid",
         "title": "Ssl Capem"
      }
   },
   "additionalProperties": false,
   "required": [
      "token"
   ]
}

Config:

extra: str = forbid

Fields:

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

Verify the ceritifcate valid

Constraints:

min_length = 1

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

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

Timeout connecting to server

Constraints:

ge = 1
le = 300

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

Parameter for the integration

Constraints:

min_length = 1

YouTube Client#

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 client: HTTPClient

Return the value for the attribute from class instance.

Returns:: Value for the attribute from class instance.

property params: YouTubeParams

Return the Pydantic model containing the configuration.

Returns:: Pydantic model containing the configuration.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

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.

pydantic model enconnect.youtube.YouTubeResult[source]

Bases: BaseModel

Contains information returned from the upstream response.

Show JSON schema

{
   "title": "YouTubeResult",
   "description": "Contains information returned from the upstream response.",
   "type": "object",
   "properties": {
      "kind": {
         "description": "Value from the server response",
         "enum": [
            "channel",
            "playlist",
            "video"
         ],
         "title": "Kind",
         "type": "string"
      },
      "channel": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Channel"
      },
      "playlist": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Playlist"
      },
      "video": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Video"
      },
      "title": {
         "description": "Value from the server response",
         "title": "Title",
         "type": "string"
      },
      "about": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "About"
      },
      "channel_title": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Value from the server response",
         "title": "Channel Title"
      },
      "thumbnail": {
         "description": "Value from the server response",
         "title": "Thumbnail",
         "type": "string"
      },
      "published": {
         "description": "Value from the server response",
         "title": "Published",
         "type": "string"
      }
   },
   "required": [
      "kind",
      "title",
      "thumbnail",
      "published"
   ]
}

Config:

extra: str = ignore

Fields:

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

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

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

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

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

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

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

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

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

Low Level 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 backoff: float

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 client_async: AsyncClient

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 headers: dict[str, str] | None

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.

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.

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.

property retry: int

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.

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.

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.

property timeout: int

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.

Enasis Network Remote Connect#

Discord Parameters#

Discord Client#

Instagram Parameters#

Instagram Client#

IRC Parameters#

IRC Client#

Mattermost Parameters#

Mattermost Client#

Philips Hue Parameters#

Philips Hue Client#

Hubitat Elevation Parameters#

Hubitat Elevation Client#

Reddit Parameters#

Reddit Client#

Ubiquiti Router Parameters#

Ubiquiti Router Client#

YouTube Parameters#

YouTube Client#

Low Level Utilities#

This Page