1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
|
"""
MeshBay Hub configuration.
Priority (highest first):
1. Environment variables (MESHBAY_*)
2. Config file (/etc/meshbay/hub.toml or --config)
3. Built-in defaults
Production config file example: /etc/meshbay/hub.toml
"""
import os
from dataclasses import dataclass, field
from pathlib import Path
try:
import tomllib
except ImportError:
import tomli as tomllib # type: ignore[no-redef]
DEFAULT_CONFIG_PATHS = [
Path("/etc/meshbay/hub.toml"),
Path.home() / ".config" / "meshbay" / "hub.toml",
]
@dataclass
class DatabaseConfig:
url: str = "sqlite+aiosqlite:///:memory:"
@dataclass
class ServerConfig:
host: str = "127.0.0.1"
port: int = 8000
workers: int = 1
@dataclass
class HubIdentityConfig:
id: str = "meshbay.org"
private_key_path: Path = field(default_factory=lambda:
Path.home() / ".config" / "meshbay" / "hub_private.pem")
@dataclass
class JWTConfig:
access_token_ttl: int = 3600 # 1 hour
refresh_token_ttl: int = 86400 * 30 # 30 days
@dataclass
class HubConfig:
db: DatabaseConfig = field(default_factory=DatabaseConfig)
server: ServerConfig = field(default_factory=ServerConfig)
identity: HubIdentityConfig = field(default_factory=HubIdentityConfig)
jwt: JWTConfig = field(default_factory=JWTConfig)
def load_config(path: Path | None = None) -> HubConfig:
cfg = HubConfig()
# Find and load TOML
candidates = [path] if path else DEFAULT_CONFIG_PATHS
for p in candidates:
if p and p.exists():
raw = tomllib.loads(p.read_text())
if db := raw.get("database", {}):
cfg.db.url = db.get("url", cfg.db.url)
if srv := raw.get("server", {}):
cfg.server.host = srv.get("host", cfg.server.host)
cfg.server.port = srv.get("port", cfg.server.port)
cfg.server.workers = srv.get("workers", cfg.server.workers)
if idn := raw.get("hub", {}):
cfg.identity.id = idn.get("id", cfg.identity.id)
if kp := idn.get("private_key_path"):
cfg.identity.private_key_path = Path(kp).expanduser()
if jwt := raw.get("jwt", {}):
cfg.jwt.access_token_ttl = jwt.get("access_token_ttl", cfg.jwt.access_token_ttl)
cfg.jwt.refresh_token_ttl = jwt.get("refresh_token_ttl", cfg.jwt.refresh_token_ttl)
break
# Env var overrides
if url := os.environ.get("MESHBAY_DATABASE_URL"):
cfg.db.url = url
if host := os.environ.get("MESHBAY_HUB_HOST"):
cfg.server.host = host
if port := os.environ.get("MESHBAY_HUB_PORT"):
cfg.server.port = int(port)
if hub_id := os.environ.get("MESHBAY_HUB_ID"):
cfg.identity.id = hub_id
if kp := os.environ.get("MESHBAY_HUB_KEY"):
cfg.identity.private_key_path = Path(kp).expanduser()
return cfg
|