pyphoenix.middleware

Middleware system for PyPhoenix channels and messages.

Classes

AuthMiddleware([auth_handler])	Middleware that handles authentication and authorization.
LoggingMiddleware([log_level])	Middleware that logs all channel activities.
Middleware()	Base class for middleware components.
MiddlewareStack([middleware])	Manages and executes middleware in order.
RateLimitMiddleware([max_messages_per_second])	Middleware that implements rate limiting.

class pyphoenix.middleware.Middleware

Base class for middleware components.

abstract async process_join(payload, socket, next_handler)

Process channel join requests.

Parameters:

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

dict[str, Any]

abstract async process_message(event, payload, socket, next_handler)

Process incoming messages.

Parameters:

• event (str)

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

Any

abstract async process_leave(reason, socket, next_handler)

Process channel leave events.

Parameters:

• reason (str)

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

None

class pyphoenix.middleware.LoggingMiddleware(log_level='info')

Middleware that logs all channel activities.

Parameters:

log_level (str)

__init__(log_level='info')

Parameters:

log_level (str)

async process_join(payload, socket, next_handler)

Log join attempts and results.

Parameters:

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

dict[str, Any]

async process_message(event, payload, socket, next_handler)

Log message processing.

Parameters:

• event (str)

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

Any

async process_leave(reason, socket, next_handler)

Log leave events.

Parameters:

• reason (str)

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

None

class pyphoenix.middleware.AuthMiddleware(auth_handler=None)

Middleware that handles authentication and authorization.

Parameters:

auth_handler (Optional[Callable[[dict[str, Any]], Any]])

__init__(auth_handler=None)

Parameters:

auth_handler (Optional[Callable[[dict[str, Any]], Any]])

async process_join(payload, socket, next_handler)

Authenticate join requests.

Parameters:

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

dict[str, Any]

async process_message(event, payload, socket, next_handler)

Check authorization for message events.

Parameters:

• event (str)

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

Any

async process_leave(reason, socket, next_handler)

Handle leave cleanup.

Parameters:

• reason (str)

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

None

class pyphoenix.middleware.RateLimitMiddleware(max_messages_per_second=10)

Middleware that implements rate limiting.

Parameters:

max_messages_per_second (int)

__init__(max_messages_per_second=10)

Parameters:

max_messages_per_second (int)

async process_join(payload, socket, next_handler)

No rate limiting on joins.

Parameters:

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

dict[str, Any]

async process_message(event, payload, socket, next_handler)

Apply rate limiting to messages.

Parameters:

• event (str)

• payload (dict[str, Any])

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

Any

async process_leave(reason, socket, next_handler)

Clean up rate limiting data on leave.

Parameters:

• reason (str)

• socket (Any)

• next_handler (Callable[..., Any])

Return type:

None

class pyphoenix.middleware.MiddlewareStack(middleware=None)

Manages and executes middleware in order.

Parameters:

middleware (Optional[list[Middleware]])

__init__(middleware=None)

Parameters:

middleware (Optional[list[Middleware]])

add(middleware)

Add middleware to the stack.

Parameters:

middleware (Middleware)

Return type:

None

remove(middleware)

Remove middleware from the stack.

Parameters:

middleware (Middleware)

Return type:

None

async process_join(payload, socket, final_handler)

Process join through all middleware.

Parameters:

• payload (dict[str, Any])

• socket (Any)

• final_handler (Callable[..., Any])

Return type:

dict[str, Any]

async process_message(event, payload, socket, final_handler)

Process message through all middleware.

Parameters:

• event (str)

• payload (dict[str, Any])

• socket (Any)

• final_handler (Callable[..., Any])

Return type:

Any

async process_leave(reason, socket, final_handler)

Process leave through all middleware.

Parameters:

• reason (str)

• socket (Any)

• final_handler (Callable[..., Any])

Return type:

None