runtimepy.net.factories package

Module contents

A module aggregating commonly used connection factory classes.

class runtimepy.net.factories.HttpConnection(transport: Transport, protocol: QueueProtocol, **kwargs)

Bases: TcpConnection

A class implementing a basic HTTP interface.

expecting_response: bool

async get_handler(response: ResponseHeader, request: RequestHeader, request_data: bytes | None) → bytes | None

Sample handler.

classmethod get_log_prefix(is_ssl: bool = False) → str

Get a logging prefix for this instance.

handlers: dict[HTTPMethod, Callable[[ResponseHeader, RequestHeader, bytes | None], Awaitable[bytes | None]]] = {}

identity = 'runtimepy/5.11.5'

init() → None

Initialize this instance.

log_alias = 'HTTP'

async post_handler(response: ResponseHeader, request: RequestHeader, request_data: bytes | None) → bytes | None

Sample handler.

async process_binary(data: bytes) → bool

Process a binary frame.

async request(request: RequestHeader, data: bytes | None = None) → Tuple[ResponseHeader, bytes | None]

Make an HTTP request.

async request_json(request: RequestHeader, data: bytes | None = None) → Any

Perform a request and convert the response to a data structure by decoding it as JSON.

class runtimepy.net.factories.RuntimepyWebsocketConnection(protocol: ClientConnection | ServerConnection, **kwargs)

Bases: WebsocketJsonMessageConnection

A class implementing a package-specific WebSocket connection.

async async_init() → bool

A runtime initialization routine (executes during ‘process’).

disable_extra() → None

Additional tasks to perform when disabling.

init() → None

Initialize this instance.

poll_connection_metrics: bool

poll_governor: RateLimiter

send_interfaces: dict[str, Callable[[dict[str, Any]], None]]

tab_sender(name: str) → Callable[[dict[str, Any]], None]

Get a tab message-sending interface.

tabs: dict[str, TabState]

ui_time: float

class runtimepy.net.factories.RuntimepyWebsocketJson

Bases: WebsocketConnectionFactory[RuntimepyWebsocketConnection]

WebSocket JSON-connection factory.

kind

alias of RuntimepyWebsocketConnection

class runtimepy.net.factories.TcpConnection(transport: Transport, protocol: QueueProtocol, **kwargs)

Bases: Connection, TransportMixin

A TCP connection interface.

async classmethod app(stop_sig: Event, callback: Callable[[T], None] = None, serving_callback: Callable[[Any], None] = None, manager: ConnectionManager = None, **kwargs) → None

Run an application that serves new connections.

async close() → None

Close this connection.

async classmethod create_connection(backoff: ExponentialBackoff = None, markdown: str = None, **kwargs) → T

Create a TCP connection.

classmethod create_pair(peer: type[V] = None, serve_kwargs: dict[str, Any] = None, connect_kwargs: dict[str, Any] = None, host: str = '127.0.0.1') → AsyncIterator[tuple[V, T]]

Create a connection pair.

classmethod get_log_prefix(is_ssl: bool = False) → str

Get a logging prefix for this instance.

property is_ssl: bool

Determine if this connection uses SSL.

log_alias = 'TCP'

log_prefix = ''

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message tos end.

send_text(data: str) → None

Enqueue a text message to send.

classmethod serve(callback: Callable[[T], None] = None, **kwargs) → AsyncIterator[Any]

Serve incoming connections.

uses_binary_tx_queue = False

uses_text_tx_queue = False

class runtimepy.net.factories.TcpConnectionFactory

Bases: ConnectionFactory, Generic[T]

A class implementing a basic TCP connection factory.

async client(name: str, *args, **kwargs) → Connection

Create a client connection.

kind: type[T]

async server_task(stop_sig: Event, manager: ConnectionManager, started_sem: Semaphore, *args, **kwargs) → Awaitable[None]

Create a task that will run a connection server.

class runtimepy.net.factories.TcpEcho

Bases: TcpConnectionFactory[EchoTcpConnection]

TCP echo-connection factory.

kind

alias of EchoTcpConnection

class runtimepy.net.factories.TcpMessage

Bases: TcpConnectionFactory[TcpPrefixedMessageConnection]

TCP message-connection factory.

kind

alias of TcpPrefixedMessageConnection

class runtimepy.net.factories.TcpMessageEcho

Bases: TcpConnectionFactory[EchoTcpMessageConnection]

TCP message-connection factory.

kind

alias of EchoTcpMessageConnection

class runtimepy.net.factories.TcpNull

Bases: TcpConnectionFactory[NullTcpConnection]

TCP null-connection factory.

kind

alias of NullTcpConnection

class runtimepy.net.factories.UdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol, **kwargs)

Bases: Connection, TransportMixin

A UDP connection interface.

async close() → None

Close this connection.

async classmethod create_connection(markdown: str = None, **kwargs) → T

Create a UDP connection.

async classmethod create_pair() → tuple[T, T]

Create a connection pair.

latest_rx_address: tuple[str, int] | None

log_alias = 'UDP'

abstractmethod async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message to send.

send_text(data: str) → None

Enqueue a text message to send.

sendto(data: bytes, addr: IPv4Host | IPv6Host | tuple[str, int] | tuple[str, int, int, int] = None) → None

Send to a specific address.

set_remote_address(addr: IPv4Host | IPv6Host) → None

Set a new remote address. Note that this doesn’t interact with or attempt to ‘connect’ the underlying socket. That should be done at creation time.

should_connect: bool = True

uses_binary_tx_queue = False

uses_text_tx_queue = False

class runtimepy.net.factories.UdpConnectionFactory

Bases: ConnectionFactory, Generic[T]

A class implementing a basic UDP connection factory.

async client(name: str, *args, **kwargs) → Connection

Create a client connection.

kind: type[T]

class runtimepy.net.factories.UdpEcho

Bases: UdpConnectionFactory[EchoUdpConnection]

UDP echo-connection factory.

kind

alias of EchoUdpConnection

class runtimepy.net.factories.UdpMessage

Bases: UdpConnectionFactory[UdpPrefixedMessageConnection]

UDP message-connection factory.

kind

alias of UdpPrefixedMessageConnection

class runtimepy.net.factories.UdpMessageEcho

Bases: UdpConnectionFactory[EchoUdpMessageConnection]

UDP echo-connection factory.

kind

alias of EchoUdpMessageConnection

class runtimepy.net.factories.UdpNull

Bases: UdpConnectionFactory[NullUdpConnection]

UDP null-connection factory.

kind

alias of NullUdpConnection

class runtimepy.net.factories.WebsocketConnection(protocol: ClientConnection | ServerConnection, **kwargs)

Bases: Connection

A simple websocket connection interface.

async classmethod app(stop_sig: Event, init: Callable[[T], Awaitable[bool]] = None, manager: ConnectionManager = None, serving_callback: Callable[[Server], None] = None, **kwargs) → None

Run a WebSocket-server application.

classmethod client(uri: str, markdown: str = None, **kwargs) → AsyncIterator[T]

A wrapper for connecting a client.

async close() → None

Close this connection.

async classmethod create_connection(uri: str, markdown: str = None, **kwargs) → T

Connect a client to an endpoint.

classmethod create_pair(serve_kwargs: dict[str, Any] = None) → AsyncIterator[tuple[T, T]]

Obtain a connected pair of WebsocketConnection objects.

classmethod serve(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None, **kwargs) → AsyncIterator[Server]

Serve a WebSocket server.

classmethod server_handler(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None) → Callable[[ServerConnection], Awaitable[None]]

A wrapper for passing in a websocket handler and initializing a connection.

class runtimepy.net.factories.WebsocketConnectionFactory

Bases: ConnectionFactory, Generic[T]

A class implementing a basic WebSocket connection factory.

async client(name: str, *args, **kwargs) → Connection

Create a client connection.

kind: type[T]

async server_task(stop_sig: Event, manager: ConnectionManager, started_sem: Semaphore, *args, **kwargs) → Awaitable[None]

Create a task that will run a connection server.

class runtimepy.net.factories.WebsocketEcho

Bases: WebsocketConnectionFactory[EchoWebsocketConnection]

WebSocket echo-connection factory.

kind

alias of EchoWebsocketConnection

class runtimepy.net.factories.WebsocketNull

Bases: WebsocketConnectionFactory[NullWebsocketConnection]

WebSocket null-connection factory.

kind

alias of NullWebsocketConnection