from utils.serverlogging import log_info, log_error
from utils.schema import get_schema
from utils.config import get_config
from tinydb import TinyDB, Query
import os
from typing import Optional
_initialized = False
_db_path = None
# --- Internal helpers ------------------------------------------------------
def _database_dir() -> str:
"""Return absolute database directory or raise."""
db_dir = get_config("databasePath")
if not db_dir:
raise RuntimeError("databasePath not configured. Check config file.")
return os.path.abspath(db_dir)
def _resolve_db_path(db_name: str) -> Optional[str]:
"""Locate the JSON file for *db_name* (case-insensitive)."""
db_dir = _database_dir()
candidate = os.path.join(db_dir, f"{db_name}.json")
if os.path.isfile(candidate):
return candidate
target = db_name.strip().lower()
for fname in os.listdir(db_dir):
if fname.lower() == f"{target}.json":
return os.path.join(db_dir, fname)
return None
def init_db():
global _initialized, _db_path
if _initialized:
return
log_info("db init complete.")
_initialized = True
def validate_record(db_name, record):
schema = get_schema(db_name)
if not schema:
log_error(f"No schema found for database '{db_name}'.")
return False, "Schema not found."
# Simple validation: check required fields and types
required = schema.get("required", [])
properties = schema.get("properties", {})
for field in required:
if field not in record:
log_error(f"Missing required field: {field}")
return False, f"Missing required field: {field}"
for field, rules in properties.items():
if field in record:
expected_type = rules.get("type")
if expected_type:
if expected_type == "string" and not isinstance(record[field], str):
log_error(f"Field '{field}' should be string.")
return False, f"Field '{field}' should be string."
if expected_type == "number" and not isinstance(record[field], (int, float)):
log_error(f"Field '{field}' should be number.")
return False, f"Field '{field}' should be number."
if expected_type == "boolean" and not isinstance(record[field], bool):
log_error(f"Field '{field}' should be boolean.")
return False, f"Field '{field}' should be boolean."
return True, None
def insert_record(db_name, record):
valid, error = validate_record(db_name, record)
if not valid:
return False, error
database_dir = get_config("databasePath")
if not database_dir:
raise RuntimeError("databasePath not configured. Check config file.")
db_path = os.path.join(database_dir, f"{db_name}.json")
db = TinyDB(db_path)
db.insert(record)
log_info(f"Record inserted into '{db_name}'.")
return True, None
def update_record(db_name, record_id, record):
valid, error = validate_record(db_name, record)
if not valid:
return False, error
database_dir = get_config("databasePath")
if not database_dir:
raise RuntimeError("databasePath not configured. Check config file.")
db_path = os.path.join(database_dir, f"{db_name}.json")
db = TinyDB(db_path)
db.update(record, doc_ids=[record_id])
log_info(f"Record {record_id} updated in '{db_name}'.")
return True, None
# Retrieve all records from a TinyDB database
def get_all_records(db_name):
db_path = _resolve_db_path(db_name)
if not db_path:
log_error(f"Database '{db_name}' not found.")
return []
records = TinyDB(db_path).all()
log_info(f"get_all_records: {len(records)} records from '{db_name}'.")
return records
# Search records in a TinyDB database using a query dict
def search_records(db_name, query_dict):
db_path = _resolve_db_path(db_name)
if not db_path:
log_error(f"Database '{db_name}' not found.")
return []
db = TinyDB(db_path)
# Build a combined TinyDB query for case-insensitive equality on each field
expr = None
for key, value in query_dict.items():
clause = Query()[key].test(lambda v, val=str(value).lower(): str(v).lower() == val)
expr = clause if expr is None else (expr & clause)
results = db.search(expr) if expr is not None else db.all()
log_info(f"search_records: {len(results)} matches in '{db_name}' for {query_dict}.")
return results
# Explicitly create a database file if it doesn't exist
def create_database(db_name):
database_dir = get_config("databasePath")
if not database_dir:
raise RuntimeError("databasePath not configured. Check config file.")
db_path = os.path.join(database_dir, f"{db_name}.json")
if not os.path.exists(db_path):
TinyDB(db_path)
log_info(f"Database file '{db_path}' created for '{db_name}'.")
else:
log_info(f"Database file '{db_path}' already exists for '{db_name}'.")
def list_databases():
"""
Lists all TinyDB databases (JSON files) in the database directory defined by config.
Returns a list of database names without .json extension (normalized to lowercase).
"""
import os
db_dir = get_config("databasePath")
if not db_dir:
raise RuntimeError("databasePath not configured. Check config file.")
if not os.path.exists(db_dir):
return []
result = []
for fname in os.listdir(db_dir):
if fname.endswith(".json") and not fname.startswith('.'):
db_name = os.path.splitext(fname)[0].lower() # Normalize names to lowercase
result.append(db_name)
return sorted(result)
