pyphoenix.config

Configuration system for PyPhoenix.

Functions

`get_config`()	Get the global configuration instance.
`set_config`(config)	Set the global configuration instance.

Classes

`ChannelConfig`([default_timeout, ...])	Configuration for channel behavior.
`LoggingConfig`([level, structured, ...])	Configuration for logging.
`PhoenixConfig`([host, port, workers, debug, ...])	Main configuration class for PyPhoenix.
`PresenceConfig`([heartbeat_interval, ...])	Configuration for presence tracking.
`PubSubConfig`([message_ttl, ...])	Configuration for PubSub system.
`SecurityConfig`([require_authentication, ...])	Configuration for security features.
`TransportConfig`([websocket_ping_interval, ...])	Configuration for transport layer.

class pyphoenix.config.ChannelConfig(default_timeout=10.0, max_join_timeout=30.0, rejoin_delay=5.0, max_push_buffer_size=100, enable_middleware=True)[source]

Configuration for channel behavior.

Parameters:

default_timeout (float)
max_join_timeout (float)
rejoin_delay (float)
max_push_buffer_size (int)
enable_middleware (bool)

default_timeout: float = 10.0

max_join_timeout: float = 30.0

rejoin_delay: float = 5.0

max_push_buffer_size: int = 100

enable_middleware: bool = True

__init__(default_timeout=10.0, max_join_timeout=30.0, rejoin_delay=5.0, max_push_buffer_size=100, enable_middleware=True)

Parameters:

default_timeout (float)
max_join_timeout (float)
rejoin_delay (float)
max_push_buffer_size (int)
enable_middleware (bool)

class pyphoenix.config.PubSubConfig(message_ttl=300.0, max_topic_length=255, max_subscribers_per_topic=1000, enable_pattern_matching=True, delivery_guarantee='at_least_once')[source]

Configuration for PubSub system.

Parameters:

message_ttl (float)
max_topic_length (int)
max_subscribers_per_topic (int)
enable_pattern_matching (bool)
delivery_guarantee (str)

message_ttl: float = 300.0

max_topic_length: int = 255

max_subscribers_per_topic: int = 1000

enable_pattern_matching: bool = True

delivery_guarantee: str = 'at_least_once'

__init__(message_ttl=300.0, max_topic_length=255, max_subscribers_per_topic=1000, enable_pattern_matching=True, delivery_guarantee='at_least_once')

Parameters:

message_ttl (float)
max_topic_length (int)
max_subscribers_per_topic (int)
enable_pattern_matching (bool)
delivery_guarantee (str)

class pyphoenix.config.PresenceConfig(heartbeat_interval=30.0, presence_timeout=120.0, cleanup_interval=300.0, max_presences_per_topic=10000)[source]

Configuration for presence tracking.

Parameters:

heartbeat_interval (float)
presence_timeout (float)
cleanup_interval (float)
max_presences_per_topic (int)

heartbeat_interval: float = 30.0

presence_timeout: float = 120.0

cleanup_interval: float = 300.0

max_presences_per_topic: int = 10000

__init__(heartbeat_interval=30.0, presence_timeout=120.0, cleanup_interval=300.0, max_presences_per_topic=10000)

Parameters:

heartbeat_interval (float)
presence_timeout (float)
cleanup_interval (float)
max_presences_per_topic (int)

class pyphoenix.config.TransportConfig(websocket_ping_interval=20.0, websocket_ping_timeout=60.0, max_connections=10000, max_message_size=65536, compression_enabled=True)[source]

Configuration for transport layer.

Parameters:

websocket_ping_interval (float)
websocket_ping_timeout (float)
max_connections (int)
max_message_size (int)
compression_enabled (bool)

websocket_ping_interval: float = 20.0

websocket_ping_timeout: float = 60.0

max_connections: int = 10000

max_message_size: int = 65536

compression_enabled: bool = True

__init__(websocket_ping_interval=20.0, websocket_ping_timeout=60.0, max_connections=10000, max_message_size=65536, compression_enabled=True)

Parameters:

websocket_ping_interval (float)
websocket_ping_timeout (float)
max_connections (int)
max_message_size (int)
compression_enabled (bool)

class pyphoenix.config.SecurityConfig(require_authentication=False, token_expiry=3600.0, max_auth_attempts=3, rate_limit_enabled=True, max_messages_per_second=10)[source]

Configuration for security features.

Parameters:

require_authentication (bool)
token_expiry (float)
max_auth_attempts (int)
rate_limit_enabled (bool)
max_messages_per_second (int)

require_authentication: bool = False

token_expiry: float = 3600.0

max_auth_attempts: int = 3

rate_limit_enabled: bool = True

max_messages_per_second: int = 10

__init__(require_authentication=False, token_expiry=3600.0, max_auth_attempts=3, rate_limit_enabled=True, max_messages_per_second=10)

Parameters:

require_authentication (bool)
token_expiry (float)
max_auth_attempts (int)
rate_limit_enabled (bool)
max_messages_per_second (int)

class pyphoenix.config.LoggingConfig(level='INFO', structured=True, include_request_id=True, log_channel_events=True, log_presence_events=False)[source]

Configuration for logging.

Parameters:

level (str)
structured (bool)
include_request_id (bool)
log_channel_events (bool)
log_presence_events (bool)

level: str = 'INFO'

structured: bool = True

include_request_id: bool = True

log_channel_events: bool = True

log_presence_events: bool = False

__init__(level='INFO', structured=True, include_request_id=True, log_channel_events=True, log_presence_events=False)

Parameters:

level (str)
structured (bool)
include_request_id (bool)
log_channel_events (bool)
log_presence_events (bool)

class pyphoenix.config.PhoenixConfig(host='localhost', port=4000, workers=1, debug=False, channel=<factory>, pubsub=<factory>, presence=<factory>, transport=<factory>, security=<factory>, logging=<factory>, custom=<factory>)[source]

Main configuration class for PyPhoenix.

Parameters:

host (str)
port (int)
workers (int)
debug (bool)
channel (ChannelConfig)
pubsub (PubSubConfig)
presence (PresenceConfig)
transport (TransportConfig)
security (SecurityConfig)
logging (LoggingConfig)
custom (dict[str, Any])

host: str = 'localhost'

port: int = 4000

workers: int = 1

debug: bool = False

channel: ChannelConfig

pubsub: PubSubConfig

presence: PresenceConfig

transport: TransportConfig

security: SecurityConfig

logging: LoggingConfig

custom: dict[str, Any]

classmethod from_env()[source]

Create configuration from environment variables.

Return type:: PhoenixConfig

classmethod from_dict(data)[source]

Create configuration from a dictionary.

Parameters:: data (dict[str, Any])
Return type:: PhoenixConfig

to_dict()[source]

Convert configuration to a dictionary.

Return type:: dict[str, Any]

validate()[source]

Validate configuration and return list of errors.

Return type:: list[str]

__init__(host='localhost', port=4000, workers=1, debug=False, channel=<factory>, pubsub=<factory>, presence=<factory>, transport=<factory>, security=<factory>, logging=<factory>, custom=<factory>)

Parameters:

host (str)
port (int)
workers (int)
debug (bool)
channel (ChannelConfig)
pubsub (PubSubConfig)
presence (PresenceConfig)
transport (TransportConfig)
security (SecurityConfig)
logging (LoggingConfig)
custom (dict[str, Any])

pyphoenix.config.get_config()[source]

Get the global configuration instance.

Return type:: PhoenixConfig

pyphoenix.config.set_config(config)[source]

Set the global configuration instance.

Parameters:: config (PhoenixConfig)
Return type:: None