LibBee / src /config.py
nikeshn's picture
Upload 2 files
72fef8d verified
Raw History Blame Contribute Delete
6.07 kB
"""Environment-backed configuration for LibBee."""
from functools import lru_cache
from pathlib import Path
from typing import Literal
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
LIBBEE_VERSION = "4.1.2"
class Settings(BaseSettings):
model_config = SettingsConfigDict(case_sensitive=False, env_file=".env", env_file_encoding="utf-8", extra="ignore")
openai_api_key: str = ""
anthropic_api_key: str = ""
primo_api_key: str = ""
# Optional HF variable. "auto" preserves the existing EU/NA/AP discovery;
# selecting the institution's gateway avoids that discovery on each search.
primo_api_region: Literal["auto", "eu", "na", "ap", "ca", "aps", "cn"] = "auto"
openalex_api_key: str = ""
contact_email: str = "libse@ku.ac.ae"
# The Worker sends this secret in X-LibBee-Origin-Token. Production
# protected routes fail closed when the value is missing.
origin_token: str = ""
environment: Literal["development", "production", "test"] = "production"
allowed_cors_origins: str = "http://localhost:8080,http://127.0.0.1:5500"
admin_password: str = ""
session_secret: str = "change-me"
maintenance_mode: bool = False
maintenance_message: str = "LibBee is receiving an update. Please try again shortly."
max_results: int = 5
max_question_chars: int = 2000
max_history_messages: int = 12
max_message_chars: int = 4000
max_concurrent_agent_requests: int = 20
# Includes the entire JSON envelope and UTF-8 history. The original 32 KiB
# ceiling rejected otherwise valid conversations under the model limits.
max_request_body_bytes: int = 524_288
# /data survives HF Space restarts when persistent storage is attached.
runtime_dir: str = "/data/libbee_runtime"
feedback_store_path: str = ""
metrics_store_path: str = ""
config_store_path: str = ""
rag_index_dir: str = ""
knowledge_dir: str = ""
# The authenticated admin page receives only a short-lived signed token;
# the permanent shared secret is never serialized into browser HTML.
cloudflare_worker_url: str = ""
cloudflare_admin_token: str = ""
# HF talks to LibCal directly because some hosted environments cannot open
# outbound connections to shared workers.dev hostnames.
libcal_client_id: str = ""
libcal_client_secret: str = ""
libcal_events_client_id: str = ""
libcal_events_client_secret: str = ""
@field_validator("max_results")
@classmethod
def reasonable_max_results(cls, value: int) -> int:
if not 1 <= value <= 20:
raise ValueError("max_results must be between 1 and 20")
return value
@field_validator("max_question_chars", "max_message_chars")
@classmethod
def reasonable_text_limit(cls, value: int, info) -> int:
ceiling = 2000 if info.field_name == "max_question_chars" else 4000
if not 256 <= value <= ceiling:
raise ValueError(f"{info.field_name} must be between 256 and {ceiling}")
return value
@field_validator("max_history_messages")
@classmethod
def reasonable_history_limit(cls, value: int) -> int:
if not 0 <= value <= 12:
raise ValueError("max_history_messages must be between 0 and 12")
return value
@field_validator("max_concurrent_agent_requests")
@classmethod
def reasonable_concurrency(cls, value: int) -> int:
if not 1 <= value <= 100:
raise ValueError("max_concurrent_agent_requests must be between 1 and 100")
return value
@field_validator("max_request_body_bytes")
@classmethod
def reasonable_body_limit(cls, value: int) -> int:
if not 16_384 <= value <= 2_097_152:
raise ValueError("max_request_body_bytes must be between 16 KiB and 2 MiB")
return value
@property
def runtime_defaults(self) -> dict:
"""One source of defaults for admin, public config, and agent replies."""
return {
"max_results": self.max_results,
"maintenance_mode": self.maintenance_mode,
"welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.",
"custom_instructions": "",
"announcement": "",
"maintenance_message": self.maintenance_message,
}
@property
def cors_origins(self) -> list[str]:
return [value.strip().rstrip("/") for value in self.allowed_cors_origins.split(",") if value.strip()]
@property
def runtime_path(self) -> Path:
requested = Path(self.runtime_dir)
# Honour explicit paths, including nested directories not created yet.
# Only the absent default HF mount falls back to ephemeral local disk.
if self.runtime_dir == "/data/libbee_runtime" and not Path("/data").is_dir():
return Path("/tmp/libbee_runtime")
return requested
@property
def feedback_path(self) -> Path:
return Path(self.feedback_store_path) if self.feedback_store_path else self.runtime_path / "feedback.jsonl"
@property
def metrics_path(self) -> Path:
return Path(self.metrics_store_path) if self.metrics_store_path else self.runtime_path / "metrics.json"
@property
def config_path(self) -> Path:
return Path(self.config_store_path) if self.config_store_path else self.runtime_path / "runtime_config.json"
@property
def rag_cache_dir(self) -> Path:
return Path(self.rag_index_dir) if self.rag_index_dir else self.runtime_path / "rag_cache"
@property
def kb_dir(self) -> Path:
if self.knowledge_dir:
return Path(self.knowledge_dir)
if Path("/app/data/knowledge").exists():
return Path("/app/data/knowledge")
return Path("data/knowledge")
@lru_cache
def get_settings() -> Settings:
settings = Settings()
settings.runtime_path.mkdir(parents=True, exist_ok=True)
settings.rag_cache_dir.mkdir(parents=True, exist_ok=True)
return settings