59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
from typing import TYPE_CHECKING, BinaryIO
|
|
|
|
import redis
|
|
|
|
if TYPE_CHECKING:
|
|
import tomli as tomllib
|
|
else:
|
|
try:
|
|
import tomllib
|
|
except ImportError:
|
|
import tomli as tomllib
|
|
|
|
|
|
class ControllerConfig:
|
|
__slots__ = ('redis_read', 'redis_write', )
|
|
|
|
redis_read: 'redis.Redis[bytes]' # Set decode_responses to False
|
|
redis_write: 'redis.Redis[bytes]' # Set decode_responses to False
|
|
|
|
def __init__(
|
|
self,
|
|
redis_read: 'redis.Redis[bytes]',
|
|
redis_write: 'redis.Redis[bytes]',
|
|
) -> None:
|
|
self.redis_read = redis_read
|
|
self.redis_write = redis_write
|
|
|
|
|
|
class Config:
|
|
__slots__ = ('controller_config', )
|
|
|
|
controller_config: ControllerConfig
|
|
|
|
def __init__(self, controller_config: ControllerConfig) -> None:
|
|
self.controller_config = controller_config
|
|
|
|
|
|
def from_toml(toml: BinaryIO) -> Config:
|
|
"""
|
|
Loads the given toml and builds a config instance out of it
|
|
"""
|
|
toml_dict = tomllib.load(toml)
|
|
|
|
redis_read = redis.Redis(
|
|
host=toml_dict['controller']['redis']['read']['host'],
|
|
port=toml_dict['controller']['redis']['read']['port'],
|
|
decode_responses=False,
|
|
)
|
|
|
|
redis_write = redis.Redis(
|
|
host=toml_dict['controller']['redis']['write']['host'],
|
|
port=toml_dict['controller']['redis']['write']['port'],
|
|
decode_responses=False,
|
|
)
|
|
|
|
controller_config = ControllerConfig(redis_read, redis_write)
|
|
|
|
return Config(controller_config)
|