Semaine 11

2026-07-11 21:38:00 +02:00
parent d3d2416dea
commit 0d071d2843
281 changed files with 54412 additions and 0 deletions
@@ -0,0 +1,157 @@
+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
+