import os
from tinydb import TinyDB, Query
from utils.config import get_config
from utils.serverlogging import log_info, log_error
# Internal state
_initialized = False
_schemas_db = None
def init_schema():
"""
Initialize schema management: config, paths.
"""
global _initialized, _schemas_db
if _initialized:
return
_schemas_db = get_config("schemaPath")
if not _schemas_db:
raise RuntimeError("schemaPath not configured. Check config file.")
_initialized = True
log_info("schema init complete.")
def get_schema(db_name):
"""Return schema for *db_name* (case-insensitive)."""
if not _initialized:
raise RuntimeError("Schema module not initialized. Call init_schema() first.")
db_name_norm = db_name.strip().lower()
db = TinyDB(_schemas_db)
SchemaQuery = Query()
# First try exact lowercase match (recommended path going forward)
result = db.search(SchemaQuery.db_name == db_name_norm)
# Fallback: any case-insensitive match for legacy records
if not result:
result = db.search(SchemaQuery.db_name.test(lambda v: str(v).lower() == db_name_norm))
return result[0].get("schema") if result else None
def set_schema(db_name, schema_dict):
if not _initialized:
raise RuntimeError("Schema module not initialized. Call init_schema() first.")
db = TinyDB(_schemas_db)
SchemaQuery = Query()
db_name_norm = db_name.strip().lower()
# Fetch existing schema and merge with provided one
existing = db.search(SchemaQuery.db_name == db_name_norm)
if existing:
current_schema = existing[0].get("schema", {})
updated_schema = {**current_schema, **schema_dict}
db.update({"schema": updated_schema}, SchemaQuery.db_name == db_name_norm)
else:
db.insert({"db_name": db_name_norm, "schema": schema_dict})
log_info(f"Schema for '{db_name}' updated.")