feat: open source multi-user AI notebook with observability
This commit is contained in:
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def _read_secret(file_var: str, value_var: str) -> str | None:
|
||||
secret_file = os.getenv(file_var)
|
||||
if secret_file:
|
||||
path = Path(secret_file)
|
||||
if path.is_file():
|
||||
return path.read_text(encoding="utf-8").strip()
|
||||
value = os.getenv(value_var)
|
||||
return value.strip() if value else None
|
||||
|
||||
|
||||
def _as_bool(name: str, default: bool) -> bool:
|
||||
value = os.getenv(name)
|
||||
if value is None:
|
||||
return default
|
||||
return value.lower() in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
data_dir: Path
|
||||
database_path: Path
|
||||
users: tuple["UserSeed", ...]
|
||||
session_secret: str | None
|
||||
deepseek_api_key: str | None
|
||||
deepseek_base_url: str
|
||||
deepseek_model: str
|
||||
cookie_secure: bool
|
||||
auth_disabled: bool
|
||||
|
||||
@classmethod
|
||||
def load(cls) -> "Settings":
|
||||
data_dir = Path(os.getenv("DATA_DIR", "./data")).resolve()
|
||||
data_dir.mkdir(parents=True, exist_ok=True)
|
||||
return cls(
|
||||
data_dir=data_dir,
|
||||
database_path=data_dir / "notes.sqlite3",
|
||||
users=_load_users(),
|
||||
session_secret=_read_secret("SESSION_SECRET_FILE", "SESSION_SECRET"),
|
||||
deepseek_api_key=_read_secret(
|
||||
"DEEPSEEK_API_KEY_FILE", "DEEPSEEK_API_KEY"
|
||||
),
|
||||
deepseek_base_url=os.getenv(
|
||||
"DEEPSEEK_BASE_URL", "https://api.deepseek.com"
|
||||
).rstrip("/"),
|
||||
deepseek_model=os.getenv("DEEPSEEK_MODEL", "deepseek-v4-pro"),
|
||||
cookie_secure=_as_bool("COOKIE_SECURE", True),
|
||||
auth_disabled=_as_bool("AUTH_DISABLED", False),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class UserSeed:
|
||||
id: str
|
||||
label: str
|
||||
role: str
|
||||
access_key_hash: str
|
||||
debug_sharing: bool = False
|
||||
|
||||
|
||||
def _load_users() -> tuple[UserSeed, ...]:
|
||||
raw = _read_secret("USERS_FILE", "USERS_JSON")
|
||||
if not raw:
|
||||
return ()
|
||||
payload = json.loads(raw)
|
||||
if not isinstance(payload, list):
|
||||
raise ValueError("users configuration must be a JSON array")
|
||||
users: list[UserSeed] = []
|
||||
seen: set[str] = set()
|
||||
for item in payload:
|
||||
if not isinstance(item, dict):
|
||||
raise ValueError("each configured user must be an object")
|
||||
user = UserSeed(
|
||||
id=str(item["id"]),
|
||||
label=str(item["label"]).strip()[:40],
|
||||
role=str(item["role"]),
|
||||
access_key_hash=str(item["access_key_hash"]),
|
||||
debug_sharing=bool(item.get("debug_sharing", False)),
|
||||
)
|
||||
if not user.id or user.id in seen:
|
||||
raise ValueError("configured user ids must be unique")
|
||||
if user.role not in {"admin", "member"}:
|
||||
raise ValueError("configured user role must be admin or member")
|
||||
if not user.label:
|
||||
raise ValueError("configured user label cannot be empty")
|
||||
seen.add(user.id)
|
||||
users.append(user)
|
||||
return tuple(users)
|
||||
Reference in New Issue
Block a user