Coverage for nexios\config\base.py: 95%
78 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-05-21 20:31 +0100
1import json
2import os
3import multiprocessing
4from typing import (
5 Any,
6 Callable,
7 Dict,
8 List,
9 Optional,
10 Union,
11 Literal,
12 TypedDict,
13 TypeVar,
14 cast,
15 get_type_hints,
16)
18# Type definitions for server configuration
19InterfaceType = Literal["asgi", "wsgi", "asgi-http"]
20HttpProtocolType = Literal["h11", "h2", "auto"]
21LogLevelType = Literal["critical", "error", "warning", "info", "debug", "trace"]
22ServerType = Literal["auto", "uvicorn", "granian"]
25class ServerConfigDict(TypedDict, total=False):
26 """TypedDict for server configuration options."""
28 host: str
29 port: int
30 workers: int
31 interface: InterfaceType
32 http_protocol: HttpProtocolType
33 log_level: LogLevelType
34 reload: bool
35 threading: bool
36 access_log: bool
37 server: ServerType
40# Type for configuration validation functions
41T = TypeVar("T")
42ValidationFunc = Callable[[T], bool]
45class MakeConfig:
46 """
47 A dynamic configuration class that allows nested dictionary access as attributes,
48 with optional validation and immutability.
50 Attributes:
51 _config (dict): Stores configuration data.
52 _immutable (bool): If True, prevents modifications.
53 _validate (dict): Stores validation rules for keys.
55 Example Usage:
56 config = MakeConfig({"db": {"host": "localhost"}}, immutable=True)
57 print(config.db.host) # "localhost"
58 """
60 def __init__(
61 self,
62 config: Dict[str, Any],
63 defaults: Optional[Dict[str, Any]] = None,
64 validate: Optional[Dict[str, Callable[[Any], bool]]] = None,
65 immutable: bool = False,
66 ):
67 """
68 Initialize the configuration object.
70 Args:
71 config (dict): Initial configuration.
72 defaults (dict, optional): Default values for missing keys.
73 validate (dict, optional): Validation rules (e.g., {"port": lambda x: x > 0}).
74 immutable (bool, optional): If True, prevents modifications.
75 """
76 self._config: Dict[str, Any] = {}
77 self._immutable: bool = immutable
78 self._validate: Dict[str, Callable[[Any], bool]] = validate or {}
80 # Apply defaults before setting config
81 merged_config = {**(defaults or {}), **config}
83 for key, value in merged_config.items():
84 self._set_config(key, value)
86 def _set_config(self, key: str, value: Optional[Any]):
87 """Validates and sets a configuration key."""
88 if key in self._validate:
89 if not self._validate[key](value):
90 raise ValueError(f"Invalid value for '{key}': {value}")
91 if isinstance(value, dict):
92 value = MakeConfig(value, immutable=self._immutable) # type: ignore
93 self._config[key] = value
95 def __setattr__(self, name: str, value: Any):
96 """Handles attribute assignment while respecting immutability."""
97 if name in {"_config", "_immutable", "_validate"}:
98 super().__setattr__(name, value)
99 elif self._immutable:
100 raise AttributeError(f"Cannot modify immutable config: '{name}'")
101 else:
102 self._set_config(name, value)
104 def __getattr__(self, name: str) -> Any:
105 """Handles attribute access, returning None if key is missing."""
106 return self._config.get(name, None)
108 def _get_nested(self, path: str) -> Any:
109 """
110 Retrieve a value from nested keys, returning None if any part is missing.
112 Args:
113 path (str): Dot-separated path, e.g., "db.host".
115 Returns:
116 Any: The value found or None.
117 """
118 keys = path.split(".")
119 current: Any = self
120 for key in keys:
121 if not isinstance(current, MakeConfig):
122 return None
123 current = current._config.get(key, None)
124 return current
126 def __getitem__(self, path: str) -> Any:
127 """Allow dictionary-like access via dot-separated keys."""
128 return self._get_nested(path)
130 def to_dict(self) -> Dict[str, Any]:
131 """Convert configuration to a standard dictionary."""
133 def recurse(config: "MakeConfig") -> Dict[str, Any]:
134 if isinstance(config, MakeConfig): # type: ignore
135 return {k: recurse(v) for k, v in config._config.items()}
136 return config
138 return recurse(self)
140 def to_json(self) -> str:
141 """Convert configuration to a JSON string."""
142 return json.dumps(self.to_dict(), indent=4)
144 def __repr__(self) -> str:
145 return f"MakeConfig({self.to_dict()})"
148# Server configuration validation
149SERVER_VALIDATION: Dict[str, ValidationFunc[Any]] = {
150 # Host validation: must be a string
151 "host": lambda x: isinstance(x, str),
152 # Port validation: must be an integer between 1 and 65535
153 "port": lambda x: isinstance(x, int) and 1 <= x <= 65535,
154 # Workers validation: must be a positive integer
155 "workers": lambda x: isinstance(x, int) and x > 0,
156 # Interface validation: must be one of the supported interface types
157 "interface": lambda x: isinstance(x, str) and x in ["asgi", "wsgi", "asgi-http"],
158 # HTTP protocol validation: must be one of the supported protocols
159 "http_protocol": lambda x: isinstance(x, str) and x in ["h11", "h2", "auto"],
160 # Log level validation: must be one of the supported log levels
161 "log_level": lambda x: isinstance(x, str)
162 and x in ["critical", "error", "warning", "info", "debug", "trace"],
163 # Reload validation: must be a boolean
164 "reload": lambda x: isinstance(x, bool),
165 # Threading validation: must be a boolean
166 "threading": lambda x: isinstance(x, bool),
167 # Access log validation: must be a boolean
168 "access_log": lambda x: isinstance(x, bool),
169 # Server validation: must be one of the supported server types
170 "server": lambda x: isinstance(x, str) and x in ["auto", "uvicorn", "granian"],
171}
174# Function to safely get environment variables with type conversion
175def _get_env_int(key: str, default: int) -> int:
176 """Get an integer environment variable with a default value."""
177 try:
178 value = os.environ.get(key)
179 return int(value) if value is not None else default
180 except (ValueError, TypeError):
181 return default
184def _get_env_bool(key: str, default: bool) -> bool:
185 """Get a boolean environment variable with a default value."""
186 value = os.environ.get(key)
187 if value is None:
188 return default
189 return value.lower() in ("true", "1", "t", "yes", "y")
192"""
193Default server configuration for Nexios applications.
194This configuration is used when running the application with 'nexios run'.
195Environment variables can override these defaults:
197HOST: The host to bind the server to (default: 127.0.0.1)
198PORT: The port to bind the server to (default: 4000)
199WORKERS: Number of worker processes (default: min(CPU count + 1, 8))
200INTERFACE: Server interface type: asgi, wsgi, or asgi-http (default: asgi)
201HTTP_PROTOCOL: HTTP protocol: h11, h2, or auto (default: auto)
202LOG_LEVEL: Logging level (default: info)
203RELOAD: Enable auto-reload on code changes (default: true)
204THREADING: Enable threading (default: false)
205ACCESS_LOG: Enable access logging (default: true)
206SERVER: Server to use: auto, uvicorn, or granian (default: auto)
208Example usage in code:
210```python
211from nexios.config import get_config
213config = get_config()
214server_config = config.server
216# Access server settings
217host = server_config.host # "127.0.0.1"
218port = server_config.port # 4000
219```
221Or override in your application:
223```python
224from nexios.config import MakeConfig, DEFAULT_SERVER_CONFIG
226# Create custom server config
227my_server_config = {
228 **DEFAULT_SERVER_CONFIG,
229 "port": 8000,
230 "workers": 4
231}
233# Create application config
234app_config = MakeConfig({
235 "debug": True,
236 "server": my_server_config
237})
238```
239"""
240DEFAULT_SERVER_CONFIG: ServerConfigDict = {
241 # The host to bind the server to
242 "host": os.environ.get("HOST", "127.0.0.1"),
243 # The port to bind the server to
244 "port": _get_env_int("PORT", 4000),
245 # Number of worker processes to use
246 "workers": _get_env_int("WORKERS", min(multiprocessing.cpu_count() + 1, 8)),
247 # The interface to use (asgi, wsgi, or asgi-http)
248 "interface": cast(InterfaceType, os.environ.get("INTERFACE", "asgi")),
249 # The HTTP protocol to use (h11, h2, or auto)
250 "http_protocol": cast(HttpProtocolType, os.environ.get("HTTP_PROTOCOL", "auto")),
251 # The log level to use
252 "log_level": cast(LogLevelType, os.environ.get("LOG_LEVEL", "info")),
253 # Whether to enable auto-reloading when code changes
254 "reload": _get_env_bool("RELOAD", True),
255 # Whether to enable threading
256 "threading": _get_env_bool("THREADING", False),
257 # Whether to enable access logging
258 "access_log": _get_env_bool("ACCESS_LOG", True),
259 # The server to use (auto, uvicorn, or granian)
260 "server": cast(ServerType, os.environ.get("SERVER", "auto")),
261}