reqivo.utils package

Submodules

reqivo.utils.serialization module

src/reqivo/utils/serialization.py

Serialization utilities for Reqivo (JSON, form-urlencode, etc.).

class reqivo.utils.serialization.Any(*args, **kwargs)

Bases: object

Special type indicating an unconstrained type.

• Any is compatible with every type.

• Any assumed to have all methods.

• All values assumed to be instances of Any.

Note that all the above statements are true from the point of view of static type checkers. At runtime, Any should not be used with instance checks.

reqivo.utils.serialization.to_json(data: Any) → str

Serializes data to a JSON string.

reqivo.utils.timing module

src/reqivo/utils/timing.py

Timeouts configuration.

class reqivo.utils.timing.Timeout(connect: float | None = None, read: float | None = None, total: float | None = None)

Bases: object

Timeout configuration.

connect

Maximum time to wait for connection establishment (socket connect).

Type:

float | None

read

Maximum time to wait for data to be received (socket recv).

Type:

float | None

total

Maximum total time for the operation.

Type:

float | None

classmethod from_float(timeout: float | None) → Timeout

Create a Timeout instance from a single float (total timeout fallback).

reqivo.utils.validators module

src/reqivo/utils/validators.py

Validation utilities for Reqivo.

reqivo.utils.validators.validate_url(url: str) → bool

Simple URL validation.

reqivo.utils.websocket_utils module

src/reqivo/utils/websocket_utils.py

WebSocket utilities for Reqivo.

exception reqivo.utils.websocket_utils.WebSocketError

Bases: Exception

Exception raised for WebSocket-related errors.

reqivo.utils.websocket_utils.apply_mask(data: bytes, mask: bytes) → bytes

Apply XOR mask to data.

reqivo.utils.websocket_utils.create_frame(payload: bytes, opcode: int, mask: bool = True) → bytes

Create a WebSocket frame.

reqivo.utils.websocket_utils.parse_frame_header(data: bytes) → Tuple[int, int, bool, int, int, int, int, bool] | None

Parse WebSocket frame header. Returns: (header_len, payload_len, fin, rsv1, rsv2, rsv3, opcode, masked)