### 🔄 DB management at startup
- [X] **If the database is empty**
- - [ ] Import initial JSON
- [X] Populate the database
+ - [X] Import initial JSON
- [X] **If the database exists**
- - [ ] Ignore JSON **unless the repository file has changed**
- - [ ] If JSON has changed → update the DB
+ - [X] Ignore JSON
---
# ⭐ Final checklist
-- [ ] Complete DB ↔ YAML management
+- [ ] Complete DB ↔ JSON management
- [ ] BIND/Kea generation with validation & rollback
- [ ] Git versioning
- [ ] BIND/Kea health check
--- /dev/null
+# backend/backup.py
+
+# import standard modules
+import json
+import os
+import time
+from typing import Iterable, List, Tuple, Dict, Any
+
+# Import local modules
+from backend.db.hosts import get_hosts, add_host, reset_hosts_db
+from backend.db.aliases import get_aliases, add_alias, reset_aliases_db
+
+# Import Settings & Logging
+from backend.settings.settings import settings
+from backend.log.log import get_logger
+
+# Logger initialization
+logger = get_logger(__name__)
+
+# ---------------------------------------------------------
+# Internal: load NDJSON utility
+# ---------------------------------------------------------
+def _load_ndjson(path: str) -> Tuple[List[Dict[str, Any]], List[str]]:
+ records: List[Dict[str, Any]] = []
+ errors: List[str] = []
+
+ if not os.path.exists(path):
+ errors.append(f"File not found: {path}")
+ return records, errors
+
+ with open(path, "r", encoding="utf-8") as f:
+ for lineno, line in enumerate(f, start=1):
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ obj = json.loads(line)
+ if isinstance(obj, dict):
+ records.append(obj)
+ else:
+ errors.append(f"{os.path.basename(path)}:{lineno} -> JSON is not an object")
+ except json.JSONDecodeError as e:
+ errors.append(f"{os.path.basename(path)}:{lineno} -> JSON decode error: {str(e)}")
+
+ return records, errors
+
+# ---------------------------------------------------------
+# Save Hosts DB
+# ---------------------------------------------------------
+def store_hosts() -> Dict[str, Any]:
+
+ # Initialization
+ start_ns = time.monotonic_ns()
+ path = os.path.join(settings.DATA_PATH, "hosts.json")
+ stored = 0
+ count_loaded = 0
+ errors: List[str] = []
+
+ try:
+ # Get Hosts List
+ hosts = get_hosts()
+ count_loaded = len(hosts)
+
+ with open(path, "w", encoding="utf-8") as f:
+ for h in hosts:
+ f.write(json.dumps(h, ensure_ascii=False) + "\n")
+ stored += 1
+
+ except Exception as e:
+ logger.exception("store_hosts failed saving records: %s", str(e).strip())
+ errors.append(str(e))
+
+ took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+
+ result: Dict[str, Any] = {
+ "file": path,
+ "count_loaded": count_loaded,
+ "count_stored": stored,
+ "took_ms": took_ms,
+ }
+
+ if errors:
+ result["errors"] = errors
+
+ return result
+
+# ---------------------------------------------------------
+# Restore Hosts DB
+# ---------------------------------------------------------
+def restore_hosts() -> Dict[str, Any]:
+
+ # Initialization
+ start_ns = time.monotonic_ns()
+ path = os.path.join(settings.DATA_PATH, "hosts.json")
+ restored = 0
+
+ # load records from NDJSON file
+ records, errors = _load_ndjson(path)
+
+ try:
+ for r in records:
+ add_host(r)
+ restored += 1
+
+ except Exception as e:
+ logger.exception("restore_hosts failed applying records: %s", str(e).strip())
+ errors.append(str(e));
+
+ took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+ return {
+ "file": path,
+ "count_loaded": len(records),
+ "count_restored": restored,
+ "errors": errors,
+ "took_ms": took_ms,
+ }
+
+# ---------------------------------------------------------
+# Save Aliases DB
+# ---------------------------------------------------------
+def store_aliases() -> Dict[str, Any]:
+
+ # Initialization
+ start_ns = time.monotonic_ns()
+ path = os.path.join(settings.DATA_PATH, "hosts.json")
+ stored = 0
+ count_loaded = 0
+ errors: List[str] = []
+
+ try:
+ # Get Aliases List
+ aliases = get_aliases()
+ count_loaded = len(aliases)
+
+ # Backup Aliases DB
+ path = os.path.join(settings.DATA_PATH, "aliases.json")
+ with open(path, "w", encoding="utf-8") as f:
+ for a in aliases:
+ f.write(json.dumps(a, ensure_ascii=False) + "\n")
+ stored += 1
+
+ except Exception as e:
+ logger.exception("store_aliases failed saving records: %s", str(e).strip())
+ errors.append(str(e))
+
+ took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+
+ result: Dict[str, Any] = {
+ "file": path,
+ "count_loaded": count_loaded,
+ "count_stored": stored,
+ "took_ms": took_ms,
+ }
+
+ if errors:
+ result["errors"] = errors
+
+ return result
+
+# ---------------------------------------------------------
+# Restore Aliases DB
+# ---------------------------------------------------------
+def restore_aliases() -> Dict[str, Any]:
+
+ # Initialization
+ start_ns = time.monotonic_ns()
+ src_path = os.path.join(settings.DATA_PATH, "aliases.json")
+ restored = 0
+
+ # load records from NDJSON file
+ records, errors = _load_ndjson(src_path)
+
+ try:
+ for r in records:
+ add_alias(r)
+ restored += 1
+
+ except Exception as e:
+ logger.exception("restore_aliases failed applying records: %s", str(e).strip())
+ errors.append(str(e));
+
+ took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+ return {
+ "file": src_path,
+ "count_loaded": len(records),
+ "count_restored": restored,
+ "errors": errors,
+ "took_ms": took_ms,
+ }
+
+# ---------------------------------------------------------
+# Backup DB
+# ---------------------------------------------------------
+def backup() -> Dict[str, Any]:
+
+ hosts_result = store_hosts()
+ aliases_result = store_aliases()
+ errors = (hosts_result.get("errors") or []) + (aliases_result.get("errors") or [])
+
+ result = {
+ "hosts": hosts_result,
+ "aliases": aliases_result,
+ }
+
+ if errors:
+ result["errors"] = errors
+
+ return result
+
+# ---------------------------------------------------------
+# Restore DB
+# ---------------------------------------------------------
+def restore(cleanup: bool = True) -> Dict[str, Any]:
+
+ if cleanup:
+ try:
+ reset_hosts_db()
+ reset_aliases_db()
+
+ except Exception as e:
+ logger.exception("Cleanup failed %s", str(e).strip())
+ raise
+
+ hosts_result = restore_hosts()
+ aliases_result = restore_aliases()
+ errors = (hosts_result.get("errors") or []) + (aliases_result.get("errors") or [])
+
+ result = {
+ "cleanup": cleanup,
+ "hosts": hosts_result,
+ "aliases": aliases_result,
+ }
+
+ if errors:
+ result["errors"] = errors
+ # GRGR -> reset db in caso di errori
+
+ return result
+
import backend.db.users
import backend.db.hosts
import backend.db.aliases
+from backend.backup import restore
# Import Settings & Logging
from backend.settings.settings import settings
# Initialize DB tables
init_db()
+ # Restore from backup if requested
+ result = restore(cleanup=False)
+ errors = result.get("errors") or []
+ if errors:
+ logger.warning("Failed restoring database from backup")
+ else:
+ logger.info("Database succesfully restored from backup")
+
# ------------------------------------------------------------------------------
# Bootstrap: setup logging, print welcome, create DB, etc.
# ------------------------------------------------------------------------------
from fastapi import APIRouter, Request, Response, HTTPException, status
from fastapi.responses import FileResponse
import asyncio
-import json
-import os
-import ipaddress
import time
from typing import Iterable, List, Tuple, Dict, Any
# Import local modules
-from backend.db.hosts import get_hosts, add_host, reset_hosts_db
-from backend.db.aliases import get_aliases, add_alias, reset_aliases_db
+from backend.backup import backup, restore
# Import Settings & Logging
from backend.settings.settings import settings
# Create Router
router = APIRouter()
-# ---------------------------------------------------------
-# Save Hosts DB
-# ---------------------------------------------------------
-def save_host():
- # Get Hosts List
- hosts = get_hosts()
-
- # Backup Hosts DB
- path = os.path.join(settings.DATA_PATH, "hosts.json")
- with open(path, "w", encoding="utf-8") as f:
- for h in hosts:
- f.write(json.dumps(h, ensure_ascii=False) + "\n")
-
-# ---------------------------------------------------------
-# Save Aliases DB
-# ---------------------------------------------------------
-def save_aliases():
- # Get Aliases List
- aliases = get_aliases()
-
- # Backup Aliases DB
- path = os.path.join(settings.DATA_PATH, "aliases.json")
- with open(path, "w", encoding="utf-8") as f:
- for a in aliases:
- f.write(json.dumps(a, ensure_ascii=False) + "\n")
-
-# ---------------------------------------------------------
-# Internal: load NDJSON utility
-# ---------------------------------------------------------
-def _load_ndjson(path: str) -> Tuple[List[Dict[str, Any]], List[str]]:
- records: List[Dict[str, Any]] = []
- errors: List[str] = []
-
- if not os.path.exists(path):
- errors.append(f"File not found: {path}")
- return records, errors
-
- with open(path, "r", encoding="utf-8") as f:
- for lineno, line in enumerate(f, start=1):
- line = line.strip()
- if not line:
- continue
- try:
- obj = json.loads(line)
- if isinstance(obj, dict):
- records.append(obj)
- else:
- errors.append(f"{os.path.basename(path)}:{lineno} -> JSON is not an object")
- except json.JSONDecodeError as e:
- errors.append(f"{os.path.basename(path)}:{lineno} -> JSON decode error: {str(e)}")
-
- return records, errors
-
-# ---------------------------------------------------------
-# Restore Hosts DB
-# ---------------------------------------------------------
-def restore_hosts() -> Dict[str, Any]:
-
- # Initialization
- start_ns = time.monotonic_ns()
- src_path = os.path.join(settings.DATA_PATH, "hosts.json")
- restored = 0
-
- # load records from NDJSON file
- records, load_errors = _load_ndjson(src_path)
-
- try:
- for r in records:
- add_host(r)
- restored += 1
-
- except Exception as e:
- logger.exception("restore_hosts failed applying records: %s", str(e).strip())
- raise
-
- took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
- return {
- "file": src_path,
- "count_loaded": len(records),
- "count_restored": restored,
- "load_errors": load_errors,
- "took_ms": took_ms,
- }
-
-# ---------------------------------------------------------
-# Restore Aliases DB
-# ---------------------------------------------------------
-def restore_aliases() -> Dict[str, Any]:
-
- # Initialization
- start_ns = time.monotonic_ns()
- src_path = os.path.join(settings.DATA_PATH, "aliases.json")
- restored = 0
-
- # load records from NDJSON file
- records, load_errors = _load_ndjson(src_path)
-
- try:
- for r in records:
- add_alias(r)
- restored += 1
-
- except Exception as e:
- logger.exception("restore_aliases failed applying records: %s", str(e).strip())
- raise
-
- took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
- return {
- "file": src_path,
- "count_loaded": len(records),
- "count_restored": restored,
- "load_errors": load_errors,
- "took_ms": took_ms,
- }
-
# ---------------------------------------------------------
# API ENDPOINTS
# ---------------------------------------------------------
-@router.get("/api/backup", status_code=status.HTTP_200_OK, responses={
- 200: {"description": "Backup executed successfully"},
- 500: {"description": "Internal server error"},
-})
+@router.get(
+ "/api/backup",
+ status_code=status.HTTP_200_OK,
+ responses={
+ 200: {"description": "Backup executed with success or failure result"},
+ 500: {"description": "Internal server error"},
+ },
+)
async def api_backup(request: Request):
# Initialization
start_ns = time.monotonic_ns()
try:
- # Backup Hosts DB
- save_host()
-
- # Backup Aliases DB
- save_aliases()
-
+ # Backup DB
+ result = backup()
+ errors = result.get("errors") or []
+
took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+
+ if errors:
+ logger.warning("Backup executed with %d error(s)", len(errors))
+ return {
+ "code": "BACKUP_ERROR",
+ "status": "failure",
+ "message": "Some operations failed",
+ "took_ms": took_ms,
+ "results": result,
+ }
+
return {
"code": "BACKUP_OK",
"status": "success",
"message": "BACKUP executed successfully",
"took_ms": took_ms,
+ "results": result,
}
except HTTPException:
# ---------------------------------------------------------
# API: Restore from backup
# ---------------------------------------------------------
-@router.get("/api/restore", status_code=status.HTTP_200_OK, responses={
- 200: {"description": "Restore executed successfully"},
- 400: {"description": "Invalid backup files"},
- 500: {"description": "Internal server error"},
-})
+@router.get(
+ "/api/restore",
+ status_code=status.HTTP_200_OK,
+ responses={
+ 200: {"description": "Restore executed with success or failure result"},
+ 500: {"description": "Internal server error"},
+ }
+)
async def api_restore(request: Request):
start_ns = time.monotonic_ns()
try:
- # 1) Restore hosts
- reset_hosts_db()
- hosts_result = restore_hosts()
-
- # 2) Restore aliases
- reset_aliases_db()
- aliases_result = restore_aliases()
-
- # Se uno dei file ha errori di parsing, segnaliamolo come 400
- load_errors = (hosts_result.get("load_errors") or []) + (aliases_result.get("load_errors") or [])
- if load_errors:
- # Non blocchiamo l'operazione se comunque abbiamo applicato record;
- # ma comunichiamo che ci sono righe scartate.
- logger.warning("Restore completed with parsing issues: %d errors", len(load_errors))
-
+ # Restore hosts DB
+ result = restore()
+ errors = (result.get("errors") or [])
+
took_ms = (time.monotonic_ns() - start_ns) / 1_000_000
+ if errors:
+ return {
+ "code": "RESTORE_ERROR",
+ "status": "failure",
+ "message": "Some operation failed",
+ "took_ms": took_ms,
+ "results": result,
+ }
+
return {
"code": "RESTORE_OK",
"status": "success",
"message": "RESTORE executed successfully",
"took_ms": took_ms,
- "results": {
- "hosts": hosts_result,
- "aliases": aliases_result,
- },
+ "results": result,
}
except HTTPException:
projects / network-manager.git / commitdiff