## ✨ Features
-- Static frontend served by the application (`FRONTEND_DIR`)
+- Static frontend served by the application (`FRONTEND_PATH`)
- Persistent SQLite database (`/data/database.db`)
- Configurable logging to console and/or file
- Login protection with configurable rate-limit
- Admin credentials configurable via env or Docker secrets
-- Support for `SESSION_SECRET`: custom key for cookies (if missing, it is generated automatically)
+- Support for `SESSION_SECRET`: custom key for cookie signing (required in production; auto-generated in development if not provided)
---
# --- Session secret (optional but recommended in production) ---
# SESSION_SECRET=****ReplaceWithYourSecret*****
```
-If SESSION_SECRET is not set, the app generates a random key on each restart -> existing sessions become invalid.
+If SESSION_SECRET is not set, the application generates a new random key at each restart, which invalidates all existing sessions.
### 3) 🐳 Example `docker-compose.yml`
```yaml
- "${HTTP_PORT:-8000}:8000"
environment:
# Frontend
- FRONTEND_DIR: "/app/frontend"
+ FRONTEND_PATH: "/app/frontend"
# Database
DB_FILE: "/data/database.db"
DB_RESET: "${DB_RESET:-false}"
## 🔧 Supported environment variables
| Variable | Default | Description |
|----------|---------|-------------|
-| `FRONTEND_DIR` | /app/frontend | Frontend directory |
+| `FRONTEND_PATH` | /app/frontend | Frontend directory |
| `DATA_PATH`| /data | Data Path for DB and Backups |
| `DB_FILE` | database.db | SQLite file |
| `DB_RESET` | false | Reset DB on every startup |
from fastapi.responses import FileResponse, RedirectResponse, JSONResponse, Response
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable
-import os
# Import Routers
from backend.routes.about import router as about_router
# ------------------------------------------------------------------------------
# Homepage
def home(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "index.html"))
+ return FileResponse(settings.FRONTEND_PATH / "index.html")
# Homepage JS
def js_home(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/index.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/index.js")
# Modals
def modals(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "modals.html"))
+ return FileResponse(settings.FRONTEND_PATH / "modals.html")
# CSS variables
def css_variables(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/variables.css"))
+ return FileResponse(settings.FRONTEND_PATH / "css/variables.css")
# CSS Layout
def css_layout(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "css/layout.css"))
+ return FileResponse(settings.FRONTEND_PATH, "css/layout.css")
# JS Common
def js_common(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/common.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/common.js")
# JS API
def js_api(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/api.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/api.js")
# JS Services
def js_services(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/services.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/services.js")
# favicon
def favicon(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "favicon.ico"))
+ return FileResponse(settings.FRONTEND_PATH / "favicon.ico")
# ------------------------------------------------------------------------------
# Creates and configures the FastAPI app
# Import standard modules
import logging
-import os
# Import backend modules
from backend.db.db import init_db
import backend.db.users
import backend.db.hosts
import backend.db.aliases
-from backend.backup import backup_restore
# Import Settings & Logging
from backend.settings.settings import settings
)
logger.info(
"App settings: frontend=%s | host=%s | port=%d | secret=%s",
- settings.FRONTEND_DIR, settings.HTTP_HOST, settings.HTTP_PORT, masked_secret
+ str(settings.FRONTEND_PATH), settings.HTTP_HOST, settings.HTTP_PORT, masked_secret
)
logger.info(
"Database: file=%s | reset=%s",
- settings.DB_FILE, settings.DB_RESET
+ str(settings.DB_FILE), settings.DB_RESET
)
logger.info(
"Log: level=%s, to_file=%s, file=%s",
- settings.LOG_LEVEL, settings.LOG_TO_FILE, settings.LOG_FILE
+ settings.LOG_LEVEL, settings.LOG_TO_FILE, str(settings.LOG_FILE)
)
logger.info(
"Users: admin=%s | password=%s | hash=%s | hash_file=%s",
- settings.ADMIN_USER, masked_admin_pwd, masked_admin_hash, settings.ADMIN_PASSWORD_HASH_FILE
+ settings.ADMIN_USER, masked_admin_pwd, masked_admin_hash, str(settings.ADMIN_PASSWORD_HASH_FILE)
)
logger.info(
"DNS: host file=%s | alias file=%s | reverse file=%s",
)
logger.info(
"Backup: path=%s",
- settings.BACKUP_PATH
+ str(settings.BACKUP_PATH)
)
logger.info(
"App features: ping_workers=%d",
# Create DB if needed
# ================================
def create_db(logger):
+ db_path = settings.DB_FILE
+
# Reset database if requested
- if settings.DB_RESET and os.path.exists(settings.DB_FILE):
- logger.info("Removing existing database: %s", settings.DB_FILE)
- os.remove(settings.DB_FILE)
+ if settings.DB_RESET and db_path.exists():
+ logger.info("Removing existing database: %s", db_path)
+ db_path.unlink()
# Skip creation if DB already exists
- if os.path.exists(settings.DB_FILE):
+ if db_path.exists():
logger.info("Database already exists. Nothing to do.")
return
- logger.info("Creating database: %s", settings.DB_FILE)
+ logger.info("Creating database: %s", db_path)
# Ensure directory exists
- os.makedirs(os.path.dirname(settings.DB_FILE) or ".", exist_ok=True)
+ db_path.parent.mkdir(parents=True, exist_ok=True)
# Initialize DB tables
init_db()
- # Restore from backup if requested
- result = backup_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.
# ------------------------------------------------------------------------------
# Create or update database
create_db(logger)
-
- #os.makedirs(DATA_DIR, exist_ok=True)
- #os.makedirs(BIND_DIR, exist_ok=True)
- #os.makedirs(KEA_DIR, exist_ok=True)
-
- #if not os.path.exists(DB_PATH):
- # conn = sqlite3.connect(DB_PATH)
- # # eventuale executescript(schema) qui
- # conn.close()
-
- #named_conf = os.path.join(BIND_DIR, "named.conf")
- #if not os.path.exists(named_conf):
- # with open(named_conf, "w") as f:
- # f.write("// generated by network-manager\n")
-
- #kea_conf = os.path.join(KEA_DIR, "kea-dhcp4.conf")
- #if not os.path.exists(kea_conf):
- # with open(kea_conf, "w") as f:
- # f.write("{\n // generated by network-manager\n}\n")
# backend/db/db.py
# Import standard modules
-import os
import sqlite3
# Import Settings & Logging
def get_db():
global _connection
if _connection is None:
- os.makedirs(os.path.dirname(settings.DB_FILE) or ".", exist_ok=True)
+ settings.DB_FILE.parent.mkdir(parents=True, exist_ok=True)
_connection = sqlite3.connect(settings.DB_FILE, check_same_thread=False)
_connection.row_factory = sqlite3.Row
_connection.execute("PRAGMA foreign_keys = ON;")
# import standard modules
import csv
-import os
from pathlib import Path
from typing import Any, Dict, List, Optional
leases = []
index = 1 # 1-based id for frontend
- path = Path(settings.DHCP4_LEASES_FILE)
+ path = settings.DHCP4_LEASES_FILE
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
# SELECT SINGLE LEASE
# -----------------------------
def get_lease(lease_id: int) -> Optional[Dict[str, Any]]:
- path = Path(settings.DHCP4_LEASES_FILE)
+ path = settings.DHCP4_LEASES_FILE
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
# -----------------------------
def delete_lease(lease_id: int):
- path = Path(settings.DHCP4_LEASES_FILE)
+ path = settings.DHCP4_LEASES_FILE
if not path.exists():
raise FileNotFoundError(f"File not found: {path}")
import logging
import logging.config
-import os
from typing import Optional
# Module-level flag to prevent re-initialization
if to_file:
if log_file is not None:
# Ensure the log directory exists and add a rotating file handler
- log_dir = os.path.dirname(log_file) or "."
- os.makedirs(log_dir, exist_ok=True)
+ log_file.parent.mkdir(parents=True, exist_ok=True)
# handler for generic log
handlers["file"] = {
"class": "logging.handlers.RotatingFileHandler",
active_handlers.append("file")
if log_access_file is not None:
# Ensure the log directory exists and add a rotating file handler
- log_dir = os.path.dirname(log_access_file) or "."
- os.makedirs(log_dir, exist_ok=True)
+ log_access_file.parent.mkdir(parents=True, exist_ok=True)
# handler for access log
handlers["access_file"] = {
"class": "logging.handlers.RotatingFileHandler",
from fastapi.responses import FileResponse
import ipaddress
import time
-import os
# Import local modules
from backend.db.aliases import (
# Aliass page
@router.get("/aliases")
def aliases(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "aliases.html"))
+ return FileResponse(settings.FRONTEND_PATH / "aliases.html")
# Serve aliases.js
@router.get("/js/aliases.js")
def js_aliases():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/aliases.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/aliases.js")
# ---------------------------------------------------------
# Get Aliass
from fastapi.responses import FileResponse
import ipaddress
import time
-import os
# Import local modules
from backend.db.hosts import get_hosts
# Devices page
@router.get("/devices")
def devices(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "devices.html"))
+ return FileResponse(settings.FRONTEND_PATH / "devices.html")
# Serve devices.js
@router.get("/js/devices.js")
def js_devices():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/devices.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/devices.js")
# ---------------------------------------------------------
# Get Devices
from fastapi import APIRouter, Request, Response, HTTPException, status
from fastapi.responses import FileResponse
import json
-import os
import time
# Import local modules
# Leases page
@router.get("/leases")
def leases(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "leases.html"))
+ return FileResponse(settings.FRONTEND_PATH / "leases.html")
# Serve leases.js
@router.get("/js/leases.js")
def js_leases():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/leases.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/leases.js")
# ---------------------------------------------------------
# Reload
from fastapi.responses import FileResponse
import asyncio
import json
-import os
import ipaddress
import time
ext_cname = get_config("external_name")
# Save DNS Host and Aliases for the EXT DNS
- path = settings.DNS_HOST_FILE + "_ext"
+ path = settings.DNS_HOST_FILE.with_name(
+ settings.DNS_HOST_FILE.name + "_ext"
+ )
with open(path, "w", encoding="utf-8") as f:
for h in hosts:
name = h.get("name").ljust(20)
from fastapi.responses import FileResponse
import ipaddress
import time
-import os
# Import local modules
from backend.db.hosts import (
# Hosts page
@router.get("/hosts")
def hosts(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "hosts.html"))
+ return FileResponse(settings.FRONTEND_PATH / "hosts.html")
# Serve hosts.js
@router.get("/js/hosts.js")
def js_hosts():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/hosts.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/hosts.js")
# ---------------------------------------------------------
# Get Hosts
# import standard modules
from fastapi import APIRouter, Request, Response, HTTPException, status
from fastapi.responses import FileResponse
-import os
import time
# Import local modules
# Login page
@router.get("/login")
def login_page(request: Request):
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "login.html"))
+ return FileResponse(settings.FRONTEND_PATH / "login.html")
# Serve login.js
@router.get("/js/login.js")
def css_login():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/login.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/login.js")
# Serve session.js
@router.get("/js/session.js")
def css_login():
- return FileResponse(os.path.join(settings.FRONTEND_DIR, "js/session.js"))
+ return FileResponse(settings.FRONTEND_PATH / "js/session.js")
# ---------------------------------------------------------
# Login API
# Uvicorn config da settings with fallback
host=(settings.HTTP_HOST or "0.0.0.0")
- port=int(settings.HTTP_PORT or 8000)
+ port=(settings.HTTP_PORT or 8000)
log_level=(settings.LOG_LEVEL or "info").lower()
workers = 1 # GRGR in prod valuta gunicorn+uvicorn workers
#reload = os.getenv("UVICORN_RELOAD", "false").lower() == "true"
# ---------------------------------------------------------
# Frontend
# ---------------------------------------------------------
-FRONTEND_DIR = "/app/frontend"
+FRONTEND_PATH = "/app/frontend"
# ---------------------------------------------------------
# Data Path (DB + Backup)
# Web
# ---------------------------------------------------------
HTTP_HOST = "0.0.0.0"
-HTTP_PORT = "8000"
-LOGIN_MAX_ATTEMPTS = "5"
-LOGIN_WINDOW_SECONDS = "600"
+HTTP_PORT = 8000
+LOGIN_MAX_ATTEMPTS = 5
+LOGIN_WINDOW_SECONDS = 600
# ---------------------------------------------------------
# Admin
# ---------------------------------------------------------
# DNS
# ---------------------------------------------------------
-DNS_HOST_FILE=f"/dns/etc/{DOMAIN}/hosts.inc"
-DNS_ALIAS_FILE=f"/dns/etc/{DOMAIN}/aliases.inc"
+DNS_HOST_FILE = "/dns/etc/{domain}/hosts.inc"
+DNS_ALIAS_FILE = "/dns/etc/{domain}/aliases.inc"
DNS_REVERSE_FILE="/dns/etc/reverse/hosts.inc"
# ---------------------------------------------------------
import datetime
from pathlib import Path
from typing import Optional
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field
# Import Parameters
from . import config, default
+from backend.utils import to_int, to_bool
# ---------------------------------------------------------
-# Convert value to boolean
+# Internal: load secret
# ---------------------------------------------------------
-def _to_bool(val) -> bool:
- if isinstance(val, bool):
- return val
- if val is None:
- return False
- return str(val).strip().lower() in {"1", "true", "yes", "on"}
+def _load_secret_key() -> str:
+ key = (
+ os.getenv("SESSION_SECRET")
+ or _read_text_if_exists(os.getenv("SECRET_KEY_FILE"))
+ )
+
+ if not key:
+ if not to_bool(os.getenv("DEV"), False):
+ print("WARNING: SECRET_KEY auto-generated (not safe for production)")
+ key = secrets.token_urlsafe(64)
+
+ return key.strip()
+
+# ---------------------------------------------------------
+# Internal: load admin hash
+# ---------------------------------------------------------
+def _load_admin_hash() -> Optional[str]:
+ env = os.getenv("ADMIN_PASSWORD_HASH")
+ if env:
+ return env
+
+ file_env = os.getenv("ADMIN_PASSWORD_HASH_FILE")
+ if file_env:
+ return _read_text_if_exists(file_env)
+
+ return _read_text_if_exists(default.ADMIN_PASSWORD_HASH_FILE)
# ---------------------------------------------------------
# Read text from file if it exists
# Versioning
APP_VERSION: str = Field(default_factory=lambda: config.APP_VERSION)
- DEVEL: bool = Field(default_factory=lambda: _to_bool(os.getenv("DEV", False)))
+ DEVEL: bool = Field(default_factory=lambda: to_bool(os.getenv("DEV"), False))
# DATA_PATH
- DATA_PATH: str = Field(default_factory=lambda: os.getenv("DATA_PATH", default.DATA_PATH))
+ DATA_PATH: Path = Field(default_factory=lambda: Path(os.getenv("DATA_PATH", default.DATA_PATH)))
# Frontend
- FRONTEND_DIR: str = Field(default_factory=lambda: os.getenv("FRONTEND_DIR", default.FRONTEND_DIR))
+ FRONTEND_PATH: Path = Field(default_factory=lambda: Path(os.getenv("FRONTEND_PATH", default.FRONTEND_PATH)))
# Database
- DB_FILE: str = Field(default_factory=lambda: os.getenv("DB_FILE", default.DB_FILE))
- DB_RESET: bool = Field(default_factory=lambda: _to_bool(os.getenv("DB_RESET", default.DB_RESET)))
+ DB_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DB_FILE", default.DB_FILE)))
+ DB_RESET: bool = Field(default_factory=lambda: to_bool(os.getenv("DB_RESET"), default.DB_RESET))
# Log
LOG_LEVEL: str = Field(default_factory=lambda: os.getenv("LOG_LEVEL", default.LOG_LEVEL))
- LOG_TO_FILE: bool = Field(default_factory=lambda: _to_bool(os.getenv("LOG_TO_FILE", default.LOG_TO_FILE)))
- LOG_FILE: str = Field(default_factory=lambda: os.getenv("LOG_FILE", default.LOG_FILE))
- LOG_ACCESS_FILE: str = Field(default_factory=lambda: os.getenv("LOG_ACCESS_FILE", default.LOG_ACCESS_FILE))
+ LOG_TO_FILE: bool = Field(default_factory=lambda: to_bool(os.getenv("LOG_TO_FILE"), default.LOG_TO_FILE))
+ LOG_FILE: Path = Field(default_factory=lambda: Path(os.getenv("LOG_FILE", default.LOG_FILE)))
+ LOG_ACCESS_FILE: Path = Field(default_factory=lambda: Path(os.getenv("LOG_ACCESS_FILE", default.LOG_ACCESS_FILE)))
# Hosts
DOMAIN: str = Field(default_factory=lambda: os.getenv("DOMAIN", default.DOMAIN))
- EXTERNAL_NAME: str = Field(default_factory=lambda: os.getenv("EXTERNAL_NAME", default.DOMAIN))
+ EXTERNAL_NAME: str = Field(default_factory=lambda: os.getenv("EXTERNAL_NAME", default.EXTERNAL_NAME))
# Web
HTTP_HOST: str = Field(default_factory=lambda: os.getenv("HTTP_HOST", default.HTTP_HOST))
- HTTP_PORT: int = Field(default_factory=lambda: int(os.getenv("HTTP_PORT", default.HTTP_PORT)))
- SECRET_KEY: str = Field(default_factory=lambda: (
- (os.getenv("SESSION_SECRET") or _read_text_if_exists(os.getenv("SECRET_KEY_FILE")) or secrets.token_urlsafe(64)).strip()
- ))
- LOGIN_MAX_ATTEMPTS: int = Field(default_factory=lambda: int(os.getenv("LOGIN_MAX_ATTEMPTS", default.LOGIN_MAX_ATTEMPTS)))
- LOGIN_WINDOW_SECONDS: int = Field(default_factory=lambda: int(os.getenv("LOGIN_WINDOW_SECONDS", default.LOGIN_WINDOW_SECONDS)))
+ HTTP_PORT: int = Field(default_factory=lambda: to_int(os.getenv("HTTP_PORT"), default.HTTP_PORT))
+ SECRET_KEY: str = Field(default_factory=_load_secret_key)
+ LOGIN_MAX_ATTEMPTS: int = Field(default_factory=lambda: to_int(os.getenv("LOGIN_MAX_ATTEMPTS"), default.LOGIN_MAX_ATTEMPTS))
+ LOGIN_WINDOW_SECONDS: int = Field(default_factory=lambda: to_int(os.getenv("LOGIN_WINDOW_SECONDS"), default.LOGIN_WINDOW_SECONDS))
# Admin
ADMIN_USER: str = Field(default_factory=lambda: os.getenv("ADMIN_USER", default.ADMIN_USER))
ADMIN_PASSWORD: str = Field(default_factory=lambda: os.getenv("ADMIN_PASSWORD", default.ADMIN_PASSWORD))
- ADMIN_PASSWORD_HASH_FILE: str = Field(default_factory=lambda: os.getenv("ADMIN_PASSWORD_HASH_FILE", default.ADMIN_PASSWORD_HASH_FILE))
- ADMIN_PASSWORD_HASH: Optional[str] = Field(default_factory=lambda: (
- (os.getenv("ADMIN_PASSWORD_HASH") or _read_text_if_exists(os.getenv("ADMIN_PASSWORD_HASH_FILE", default.ADMIN_PASSWORD_HASH_FILE)) or None)
- ))
+ ADMIN_PASSWORD_HASH_FILE: Path = Field(default_factory=lambda: Path(os.getenv("ADMIN_PASSWORD_HASH_FILE", default.ADMIN_PASSWORD_HASH_FILE)))
+ ADMIN_PASSWORD_HASH: Optional[str] = Field(default_factory=_load_admin_hash)
# DNS
- DNS_HOST_FILE: str = Field(default_factory=lambda: os.getenv("DNS_HOST_FILE", default.DNS_HOST_FILE))
- DNS_ALIAS_FILE: str = Field(default_factory=lambda: os.getenv("DNS_ALIAS_FILE", default.DNS_ALIAS_FILE))
- DNS_REVERSE_FILE: str = Field(default_factory=lambda: os.getenv("DNS_REVERSE_FILE", default.DNS_REVERSE_FILE))
+ DNS_HOST_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DNS_HOST_FILE", default.DNS_HOST_FILE)))
+ DNS_ALIAS_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DNS_ALIAS_FILE", default.DNS_ALIAS_FILE)))
+ DNS_REVERSE_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DNS_REVERSE_FILE", default.DNS_REVERSE_FILE)))
# DHCP
- DHCP4_HOST_FILE: str = Field(default_factory=lambda: os.getenv("DHCP4_HOST_FILE", default.DHCP4_HOST_FILE))
- DHCP4_LEASES_FILE: str = Field(default_factory=lambda: os.getenv("DHCP4_LEASES_FILE", default.DHCP4_LEASES_FILE))
- DHCP6_HOST_FILE: str = Field(default_factory=lambda: os.getenv("DHCP6_HOST_FILE", default.DHCP6_HOST_FILE))
- DHCP6_LEASES_FILE: str = Field(default_factory=lambda: os.getenv("DHCP6_LEASES_FILE", default.DHCP6_LEASES_FILE))
+ DHCP4_HOST_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DHCP4_HOST_FILE", default.DHCP4_HOST_FILE)))
+ DHCP4_LEASES_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DHCP4_LEASES_FILE", default.DHCP4_LEASES_FILE)))
+ DHCP6_HOST_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DHCP6_HOST_FILE", default.DHCP6_HOST_FILE)))
+ DHCP6_LEASES_FILE: Path = Field(default_factory=lambda: Path(os.getenv("DHCP6_LEASES_FILE", default.DHCP6_LEASES_FILE)))
# Backup
- BACKUP_PATH: str= Field(default_factory=lambda: os.getenv("BACKUP_PATH", default.BACKUP_PATH))
+ BACKUP_PATH: Path = Field(default_factory=lambda: Path(os.getenv("BACKUP_PATH", default.BACKUP_PATH)))
# APP Features
- PING_WORKERS: int = Field(default_factory=lambda: int(os.getenv("PING_WORKERS", default.PING_WORKERS)))
+ PING_WORKERS: int = Field(default_factory=lambda: to_int(os.getenv("PING_WORKERS"), default.PING_WORKERS))
+ # ---------------------------------------------------------
+ # Post init process
+ # ---------------------------------------------------------
def model_post_init(self, __context) -> None:
if self.DEVEL:
ts = datetime.datetime.now().strftime("%Y%m%d-%H%M")
object.__setattr__(self, "APP_VERSION", f"{self.APP_VERSION}-dev-{ts}")
- else:
- object.__setattr__(self, "APP_VERSION", self.APP_VERSION)
-
- # Database
- self.DB_FILE = self.DATA_PATH + "/" + self.DB_FILE
- self.LOG_FILE = self.DATA_PATH + "/" + self.LOG_FILE
- self.LOG_ACCESS_FILE = self.DATA_PATH + "/" + self.LOG_ACCESS_FILE
-
- # Backup
- self.BACKUP_PATH = self.BACKUP_PATH if os.path.isabs(self.BACKUP_PATH) else os.path.join(self.DATA_PATH, self.BACKUP_PATH)
- # Update DNS Files
- if self.DOMAIN.lower() != default.DOMAIN.lower():
- self.DNS_HOST_FILE = self.DNS_HOST_FILE.replace(default.DOMAIN, self.DOMAIN)
- self.DNS_ALIAS_FILE = self.DNS_ALIAS_FILE.replace(default.DOMAIN, self.DOMAIN)
+ # Folder Data Creation
+ self.DATA_PATH.mkdir(parents=True, exist_ok=True)
+
+ # Update DB file path including DATA_PATH
+ if not self.DB_FILE.is_absolute():
+ object.__setattr__(self, "DB_FILE", self.DATA_PATH / self.DB_FILE)
+
+ # Update Log files path including DATA_PATH
+ if not self.LOG_FILE.is_absolute():
+ object.__setattr__(self, "LOG_FILE", self.DATA_PATH / self.LOG_FILE)
+ if not self.LOG_ACCESS_FILE.is_absolute():
+ object.__setattr__(self, "LOG_ACCESS_FILE", self.DATA_PATH / self.LOG_ACCESS_FILE)
+
+ # Updated Backup Path
+ if not self.BACKUP_PATH.is_absolute():
+ object.__setattr__(self, "BACKUP_PATH", self.DATA_PATH / self.BACKUP_PATH)
+ self.BACKUP_PATH.parent.mkdir(parents=True, exist_ok=True)
+
+ # Update DNS references based on domain name
+ if "{domain}" in str(self.DNS_HOST_FILE):
+ object.__setattr__(
+ self,
+ "DNS_HOST_FILE",
+ Path(str(self.DNS_HOST_FILE).format(domain=self.DOMAIN)),
+ )
+ if "{domain}" in str(self.DNS_ALIAS_FILE):
+ object.__setattr__(
+ self,
+ "DNS_ALIAS_FILE",
+ Path(str(self.DNS_ALIAS_FILE).format(domain=self.DOMAIN)),
+ )
# ---------------------------------------------------------
# Singleton
# backend/db/utils.py
# Import standard modules
-import os
-import platform
import subprocess
-# -----------------------------
-# Load hash from file
-# -----------------------------
-def load_hash(path: str):
- if path and os.path.exists(path):
- with open(path, "r") as f:
- return f.read().strip()
- return None
-
# -----------------------------
# Normalize string (strip and convert empty to None)
# -----------------------------
# -----------------------------
# convert string to int (returns None if conversion fails)
# -----------------------------
-def to_int(v: str):
+def to_int(v: str, default: int | None = None) -> int | None:
v = (v or "").strip()
if not v or v.lower() == "null":
- return None
+ return default
try:
return int(v)
except ValueError:
- return None
+ return default
# -----------------------------
# convert string to bool (returns None if conversion fails)
# -----------------------------
-def to_bool(v: str):
+def to_bool(v: str, default: bool | None = None) -> bool | None:
v = (v or "").strip().lower()
if v in ("true", "1", "yes", "y"):
return True
if v in ("false", "0", "no", "n"):
return False
- return None
+ return default
# -----------------------------
# check if host is active (ping)
--bg-selected: rgba(var(--accent-rgb), 0.12);
--border-selected: rgba(var(--accent-rgb), 0.4);
-}
\ No newline at end of file
+}
projects / network-manager.git / commitdiff