live-campus-mcs-p-2027.2/Semaine_11/Jour_03/database.py

import sqlite3
from pathlib import Path
from contextlib import contextmanager
from typing import Any, Iterable, Optional

from classes import Equipment, Interface

BASE_DIR = Path(__file__).resolve().parent
DB_DIR = BASE_DIR / "databases"
DB_PATH = DB_DIR / "inventory.db"

SCHEMA = """
CREATE TABLE IF NOT EXISTS equipments (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    hostname    TEXT    NOT NULL DEFAULT '[HOSTNAME]',
    type        TEXT    NOT NULL DEFAULT 'Server',
    ip_address  TEXT    NOT NULL DEFAULT '255.255.255.0',
    is_active   INTEGER NOT NULL DEFAULT 1,
    latency     REAL
);

CREATE TABLE IF NOT EXISTS interfaces (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    port         INTEGER NOT NULL,
    service      TEXT    NOT NULL DEFAULT '[SERVICE]',
    equipment_id INTEGER NOT NULL,
    FOREIGN KEY (equipment_id) REFERENCES equipments(id) ON DELETE CASCADE
);
"""
def _ensure_db_dir() -> None:
    DB_DIR.mkdir(parents=True, exist_ok=True)

@contextmanager
def get_connection():
    _ensure_db_dir()
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    try:
        yield conn
    finally:
        conn.close()

def init_database() -> None:
    with get_connection() as conn:
        conn.executescript(SCHEMA)
        conn.commit()

def execute_query(query: str, params: Iterable[Any] = ()) -> list[sqlite3.Row]:
    with get_connection() as conn:
        cursor = conn.execute(query, tuple(params))
        return cursor.fetchall()

def execute_modification(query: str, params: Iterable[Any] = ()) -> int:
    with get_connection() as conn:
        cursor = conn.execute(query, tuple(params))
        conn.commit()
        return cursor.rowcount

def _row_to_equipment(row: sqlite3.Row) -> Equipment:
    return Equipment(
        id=row["id"],
        hostname=row["hostname"],
        type=row["type"],
        ip_address=row["ip_address"],
        is_active=bool(row["is_active"]),
        latency=row["latency"],
    )

def _row_to_interface(row: sqlite3.Row) -> Interface:
    return Interface(
        id=row["id"],
        port=row["port"],
        service=row["service"],
        equipment_id=row["equipment_id"],
    )

def fetch_all_equipments() -> list[Equipment]:
    rows = execute_query(
        "SELECT id, hostname, type, ip_address, is_active, latency "
        "FROM equipments ORDER BY id"
    )
    equipments = [_row_to_equipment(r) for r in rows]
    for eq in equipments:
        eq.interfaces = fetch_interfaces_for_equipment(eq.id)  #type: ignore
    return equipments

def fetch_equipment_by_id(equipment_id: int) -> Optional[Equipment]:
    rows = execute_query(
        "SELECT id, hostname, type, ip_address, is_active, latency "
        "FROM equipments WHERE id = ?",
        (equipment_id,),
    )
    if not rows:
        return None
    eq = _row_to_equipment(rows[0])
    eq.interfaces = fetch_interfaces_for_equipment(eq.id)  #type: ignore
    return eq

def fetch_interfaces_for_equipment(equipment_id: int) -> list[Interface]:
    rows = execute_query(
        "SELECT id, port, service, equipment_id FROM interfaces "
        "WHERE equipment_id = ? ORDER BY id",
        (equipment_id,),
    )
    return [_row_to_interface(r) for r in rows]

def insert_equipment(eq: Equipment) -> int:
    with get_connection() as conn:
        cursor = conn.execute(
            "INSERT INTO equipments (hostname, type, ip_address, is_active, latency) "
            "VALUES (?, ?, ?, ?, ?)",
            (eq.hostname, eq.type, eq.ip_address, int(eq.is_active), eq.latency),
        )
        eq_id = cursor.lastrowid
        eq.id = eq_id  # <-- ajout
        for iface in eq.interfaces:
            conn.execute(
                "INSERT INTO interfaces (port, service, equipment_id) "
                "VALUES (?, ?, ?)",
                (iface.port, iface.service, eq_id),
            )
        conn.commit()
        return eq_id    #type:ignore

def update_equipment(eq: Equipment) -> bool:
    return execute_modification(
        "UPDATE equipments SET hostname=?, type=?, ip_address=?, "
        "is_active=?, latency=? WHERE id=?",
        (eq.hostname, eq.type, eq.ip_address, int(eq.is_active),
         eq.latency, eq.id),
    ) > 0

def delete_equipment(equipment_id: int) -> bool:
    return execute_modification(
        "DELETE FROM equipments WHERE id = ?", (equipment_id,)
    ) > 0

def reset_database() -> None:
    from init_db import main as seed

    if DB_PATH.exists():
        DB_PATH.unlink()
    init_database()
    seed()

def prompt_reset() -> bool:
    while True:
        choice = input("Réinitialiser la base de données ? [o/N] : ").strip().lower()
        if choice in {"o", "oui", "y", "yes"}:
            return True
        if choice in {"", "n", "non", "no"}:
            return False
        print("Réponse invalide. Saisissez 'o' (oui) ou 'n' (non).")

# Gestion API