LotusGuild/gandalf
3
1
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b9412ff6bad1530d8c3c2582080417c328d16e1c
gandalf/db.py
T
jaredandClaude Opus 5.5 b9412ff6ba
Lint / JS (eslint) (push) Successful in 7s
Details
Lint / Python (flake8) (push) Successful in 1m36s
Details
Lint / Notify on failure (push) Skipped
Details
Lint / Deploy (push) Skipped
Details
Security / Python Security (bandit) (push) Successful in 1m40s
Details
Lint / Python (flake8) (pull_request) Successful in 52s
Details
Test / Python Tests (pytest) (push) Successful in 1m4s
Details
Lint / JS (eslint) (pull_request) Successful in 7s
Details
Lint / Notify on failure (pull_request) Skipped
Details
Lint / Deploy (pull_request) Skipped
Details
Security / Python Security (bandit) (pull_request) Successful in 1m13s
Details
Test / Python Tests (pytest) (pull_request) Successful in 1m15s
Details
Add cutsheets: planned cabling checked against discovery 
A new Cutsheet page records where every cable is meant to go and compares
it with what Gandalf discovers.

- Inventory for what discovery can't see: patch panels, wall jacks,
  shelves and rack positions. Panels and jacks pass through front to rear.
- Cables with two endpoints, ID, type, length, color, status (planned,
  installed, dark, retired) and notes; the far end may be unterminated.
- Each cable's full run is traced through panels and jacks, and runs
  ending on a switch port are checked against its discovered neighbor
  (LLDP, client table, or a DAC serial seen at both ends): matches,
  wrong neighbor, no link, missing port, port conflict, unverified, idle.
- Undocumented links: live switch ports no cable mentions, one click to
  add with what Gandalf already knows. Port panels show the planned cable.
- Labels to make: cable IDs and panel-port labels not yet applied, with a
  print view, for a handheld label maker.
- CSV export, and all-or-nothing import that upserts by cable ID.
- Cables, panels and jacks appear in the dashboard search.

Editing needs the admin group and a JSON request body. Every change is
written to cutsheet_history in the same transaction. Tables are created
on first use.

Refs #12

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 16:05:08 -05:00

790 lines
30 KiB
Python
Raw Blame History

"""Database operations for Gandalf network monitor."""
import json
import logging
import threading
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from typing import Optional
import pymysql
import pymysql.cursors
logger = logging.getLogger(__name__)
_config_cache = None
_local = threading.local()
def _config() -> dict:
global _config_cache
if _config_cache is None:
with open('config.json') as f:
_config_cache = json.load(f)['database']
return _config_cache
def _new_conn(cfg: dict):
return pymysql.connect(
host=cfg['host'],
port=cfg.get('port', 3306),
user=cfg['user'],
password=cfg['password'],
database=cfg['name'],
autocommit=True,
cursorclass=pymysql.cursors.DictCursor,
connect_timeout=10,
charset='utf8mb4',
# All app code treats database times as UTC. Without this, a server in another
# time zone shifts displayed ages and timed suppression expiries by its offset.
init_command="SET time_zone='+00:00'",
)
@contextmanager
def get_conn():
"""Yield a per-thread cached database connection, reconnecting as needed."""
cfg = _config()
conn = getattr(_local, 'conn', None)
if conn is None:
conn = _new_conn(cfg)
_local.conn = conn
else:
try:
conn.ping(reconnect=True)
except Exception:
try:
conn.close()
except Exception:
pass
_local.conn = None
conn = _new_conn(cfg)
_local.conn = conn
yield conn
# ---------------------------------------------------------------------------
# Monitor state (key/value store)
# ---------------------------------------------------------------------------
def set_state(key: str, value) -> None:
if not isinstance(value, str):
value = json.dumps(value, default=str)
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO monitor_state (key_name, value)
VALUES (%s, %s)
ON DUPLICATE KEY UPDATE value=VALUES(value), updated_at=NOW()""",
(key, value),
)
def get_state(key: str, default=None):
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute('SELECT value FROM monitor_state WHERE key_name=%s', (key,))
row = cur.fetchone()
return row['value'] if row else default
# ---------------------------------------------------------------------------
# Interface baseline tracking
# ---------------------------------------------------------------------------
def get_baseline() -> dict:
raw = get_state('interface_baseline')
if raw:
try:
return json.loads(raw)
except Exception:
logger.error('Failed to parse interface_baseline JSON; resetting baseline')
return {}
def set_baseline(baseline: dict) -> None:
set_state('interface_baseline', json.dumps(baseline))
# ---------------------------------------------------------------------------
# Network events
# ---------------------------------------------------------------------------
def upsert_event(
event_type: str,
severity: str,
source_type: str,
target_name: str,
target_detail: str,
description: str,
) -> tuple:
"""Insert or update a network event. Returns (id, is_new, consecutive_failures, ticket_id)."""
detail = target_detail or ''
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT id, consecutive_failures, ticket_id FROM network_events
WHERE event_type=%s AND target_name=%s AND target_detail=%s
AND resolved_at IS NULL LIMIT 1""",
(event_type, target_name, detail),
)
existing = cur.fetchone()
if existing:
new_count = existing['consecutive_failures'] + 1
cur.execute(
"""UPDATE network_events
SET last_seen=NOW(), consecutive_failures=%s, description=%s
WHERE id=%s""",
(new_count, description, existing['id']),
)
return existing['id'], False, new_count, existing.get('ticket_id')
else:
cur.execute(
"""INSERT INTO network_events
(event_type, severity, source_type, target_name, target_detail, description)
VALUES (%s, %s, %s, %s, %s, %s)""",
(event_type, severity, source_type, target_name, detail, description),
)
return cur.lastrowid, True, 1, None
def get_events_for_target(target_name: str, limit: int = 25) -> list:
"""Return a device's events, unresolved first, then most recent."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT * FROM network_events WHERE target_name=%s
ORDER BY resolved_at IS NOT NULL, COALESCE(resolved_at, last_seen) DESC
LIMIT %s""",
(target_name, limit),
)
rows = cur.fetchall()
for r in rows:
for k in ('first_seen', 'last_seen', 'resolved_at'):
if r.get(k) and hasattr(r[k], 'isoformat'):
r[k] = r[k].isoformat() + 'Z'
return rows
def resolve_event(event_type: str, target_name: str, target_detail: str = '') -> None:
detail = target_detail or ''
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""UPDATE network_events SET resolved_at=NOW()
WHERE event_type=%s AND target_name=%s AND target_detail=%s
AND resolved_at IS NULL""",
(event_type, target_name, detail),
)
def set_ticket_id(event_id: int, ticket_id: str) -> None:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
'UPDATE network_events SET ticket_id=%s WHERE id=%s',
(ticket_id, event_id),
)
def get_active_events(limit: int = 200, offset: int = 0) -> list:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT * FROM network_events
WHERE resolved_at IS NULL
ORDER BY
FIELD(severity,'critical','warning','info'),
first_seen DESC
LIMIT %s OFFSET %s""",
(limit, offset),
)
rows = cur.fetchall()
for r in rows:
for k in ('first_seen', 'last_seen'):
if r.get(k) and hasattr(r[k], 'isoformat'):
r[k] = r[k].isoformat() + 'Z'
return rows
def count_active_events() -> int:
"""Return count of all unresolved events (for pagination)."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT COUNT(*) AS n FROM network_events WHERE resolved_at IS NULL"
)
return cur.fetchone()['n']
def get_recent_resolved(hours: int = 24, limit: int = 50) -> list:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT * FROM network_events
WHERE resolved_at IS NOT NULL
AND resolved_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
ORDER BY resolved_at DESC LIMIT %s""",
(hours, limit),
)
rows = cur.fetchall()
for r in rows:
for k in ('first_seen', 'last_seen', 'resolved_at'):
if r.get(k) and hasattr(r[k], 'isoformat'):
r[k] = r[k].isoformat() + 'Z'
return rows
def get_status_summary() -> dict:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT severity, COUNT(*) as cnt FROM network_events
WHERE resolved_at IS NULL GROUP BY severity"""
)
counts = {r['severity']: r['cnt'] for r in cur.fetchall()}
cur.execute(
"""SELECT COUNT(*) as cnt FROM network_events
WHERE resolved_at IS NOT NULL
AND resolved_at > DATE_SUB(NOW(), INTERVAL 24 HOUR)"""
)
resolved_24h = cur.fetchone()['cnt']
return {
'critical': counts.get('critical', 0),
'warning': counts.get('warning', 0),
'info': counts.get('info', 0),
'resolved_24h': resolved_24h,
}
# ---------------------------------------------------------------------------
# Suppression rules
# ---------------------------------------------------------------------------
def get_active_suppressions() -> list:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT * FROM suppression_rules
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
ORDER BY created_at DESC"""
)
rows = cur.fetchall()
for r in rows:
for k in ('created_at', 'expires_at'):
if r.get(k) and hasattr(r[k], 'isoformat'):
r[k] = r[k].isoformat() + 'Z'
return rows
def get_suppression_history(limit: int = 50) -> list:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
'SELECT * FROM suppression_rules ORDER BY created_at DESC LIMIT %s',
(limit,),
)
rows = cur.fetchall()
for r in rows:
for k in ('created_at', 'expires_at'):
if r.get(k) and hasattr(r[k], 'isoformat'):
r[k] = r[k].isoformat() + 'Z'
return rows
def create_suppression(
target_type: str,
target_name: str,
target_detail: str,
reason: str,
suppressed_by: str,
expires_minutes: Optional[int] = None,
) -> int:
expires_at = None
if expires_minutes:
expires_at = datetime.now(timezone.utc) + timedelta(minutes=int(expires_minutes))
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO suppression_rules
(target_type, target_name, target_detail, reason, suppressed_by, expires_at, active)
VALUES (%s, %s, %s, %s, %s, %s, TRUE)""",
(target_type, target_name or '', target_detail or '', reason, suppressed_by, expires_at),
)
return cur.lastrowid
def deactivate_suppression(sup_id: int) -> None:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
'UPDATE suppression_rules SET active=FALSE WHERE id=%s', (sup_id,)
)
def cleanup_expired_suppressions() -> int:
"""Mark expired time-limited suppressions as inactive. Returns count deactivated."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""UPDATE suppression_rules
SET active=FALSE
WHERE active=TRUE AND expires_at IS NOT NULL AND expires_at <= NOW()"""
)
n = cur.rowcount
if n:
logger.info(f'Deactivated {n} expired suppression(s)')
return n
def purge_old_resolved_events(days: int = 90) -> int:
"""Delete resolved events older than `days` days. Returns count deleted."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""DELETE FROM network_events
WHERE resolved_at IS NOT NULL
AND resolved_at < DATE_SUB(NOW(), INTERVAL %s DAY)""",
(days,),
)
n = cur.rowcount
if n:
logger.info(f'Purged {n} old resolved event(s) (>{days}d)')
return n
# ---------------------------------------------------------------------------
# WAN modem optical history
# ---------------------------------------------------------------------------
def ensure_modem_schema() -> None:
"""Create the modem history table so deploys need no manual migration."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""CREATE TABLE IF NOT EXISTS modem_samples (
sampled_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP PRIMARY KEY,
rx_power_dbm DECIMAL(5,1) NULL,
tx_power_dbm DECIMAL(5,1) NULL,
temp_c DECIMAL(5,1) NULL
) ENGINE=InnoDB"""
)
def record_modem_sample(rx_power_dbm, tx_power_dbm, temp_c) -> None:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO modem_samples (sampled_at, rx_power_dbm, tx_power_dbm, temp_c)
VALUES (NOW(), %s, %s, %s)
ON DUPLICATE KEY UPDATE rx_power_dbm=VALUES(rx_power_dbm),
tx_power_dbm=VALUES(tx_power_dbm), temp_c=VALUES(temp_c)""",
(rx_power_dbm, tx_power_dbm, temp_c),
)
def get_modem_rx_values(days: int = 7) -> list:
"""Return recent Rx power readings (dBm) for computing a drift baseline."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT rx_power_dbm FROM modem_samples
WHERE rx_power_dbm IS NOT NULL
AND sampled_at > DATE_SUB(NOW(), INTERVAL %s DAY)""",
(days,),
)
return [float(r['rx_power_dbm']) for r in cur.fetchall()]
def get_modem_history(hours: int = 168, bucket_seconds: int = 3600) -> list:
"""Return optical history averaged into time buckets, oldest first."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT FLOOR(UNIX_TIMESTAMP(sampled_at) / %s) * %s AS ts,
ROUND(AVG(rx_power_dbm), 1) AS rx_avg,
MIN(rx_power_dbm) AS rx_min,
ROUND(AVG(tx_power_dbm), 1) AS tx_avg,
ROUND(AVG(temp_c), 1) AS temp_avg
FROM modem_samples
WHERE sampled_at > DATE_SUB(NOW(), INTERVAL %s HOUR)
GROUP BY ts ORDER BY ts""",
(bucket_seconds, bucket_seconds, hours),
)
return [
{k: (float(v) if v is not None else None) for k, v in row.items()}
for row in cur.fetchall()
]
def purge_old_modem_samples(days: int = 90) -> int:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
'DELETE FROM modem_samples WHERE sampled_at < DATE_SUB(NOW(), INTERVAL %s DAY)',
(days,),
)
return cur.rowcount
def check_suppressed(suppressions: list, target_type: str, target_name: str, target_detail: str = '') -> bool:
"""Check suppression against a pre-loaded list (avoids per-call DB queries)."""
for s in suppressions:
if s['target_type'] == 'all':
return True
if s['target_type'] == target_type and s['target_name'] == target_name:
if not (s.get('target_detail') or ''):
return True
if target_detail and s.get('target_detail') == target_detail:
return True
return False
def is_suppressed(target_type: str, target_name: str, target_detail: str = '') -> bool:
with get_conn() as conn:
with conn.cursor() as cur:
# Global suppression (all)
cur.execute(
"""SELECT id FROM suppression_rules
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
AND target_type='all' LIMIT 1"""
)
if cur.fetchone():
return True
if not target_name:
return False
# Host-level suppression (covers all interfaces on that host)
cur.execute(
"""SELECT id FROM suppression_rules
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
AND target_type=%s AND target_name=%s
AND target_detail='' LIMIT 1""",
(target_type, target_name),
)
if cur.fetchone():
return True
# Interface/device-specific suppression
if target_detail:
cur.execute(
"""SELECT id FROM suppression_rules
WHERE active=TRUE AND (expires_at IS NULL OR expires_at > NOW())
AND target_type=%s AND target_name=%s AND target_detail=%s LIMIT 1""",
(target_type, target_name, target_detail),
)
if cur.fetchone():
return True
return False
# ---------------------------------------------------------------------------
# Cutsheets (see cutsheet.py). Every change writes a cutsheet_history row in
# the same transaction, so the log can't miss an edit.
# ---------------------------------------------------------------------------
_CUTSHEET_SCHEMA = (
"""CREATE TABLE IF NOT EXISTS cutsheet_items (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
kind VARCHAR(20) NOT NULL,
port_count SMALLINT NULL,
location VARCHAR(255) NOT NULL DEFAULT '',
rack_u SMALLINT NULL,
rack_height SMALLINT NULL,
notes TEXT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uq_item_name (name)
) ENGINE=InnoDB""",
"""CREATE TABLE IF NOT EXISTS cutsheet_cables (
id INT AUTO_INCREMENT PRIMARY KEY,
label VARCHAR(32) NULL,
a_device VARCHAR(100) NOT NULL,
a_port VARCHAR(64) NOT NULL DEFAULT '',
a_side VARCHAR(5) NOT NULL DEFAULT '',
b_device VARCHAR(100) NOT NULL DEFAULT '',
b_port VARCHAR(64) NOT NULL DEFAULT '',
b_side VARCHAR(5) NOT NULL DEFAULT '',
cable_type VARCHAR(32) NOT NULL DEFAULT '',
length_m DECIMAL(5,2) NULL,
color VARCHAR(32) NOT NULL DEFAULT '',
status VARCHAR(10) NOT NULL DEFAULT 'installed',
label_applied BOOLEAN NOT NULL DEFAULT FALSE,
notes TEXT NULL,
created_by VARCHAR(255) NOT NULL DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_by VARCHAR(255) NOT NULL DEFAULT '',
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uq_cable_label (label),
INDEX idx_cable_a (a_device, a_port),
INDEX idx_cable_b (b_device, b_port)
) ENGINE=InnoDB""",
"""CREATE TABLE IF NOT EXISTS cutsheet_port_labels (
item_id INT NOT NULL,
port VARCHAR(64) NOT NULL,
label VARCHAR(64) NOT NULL DEFAULT '',
applied BOOLEAN NOT NULL DEFAULT FALSE,
PRIMARY KEY (item_id, port)
) ENGINE=InnoDB""",
"""CREATE TABLE IF NOT EXISTS cutsheet_history (
id INT AUTO_INCREMENT PRIMARY KEY,
changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
changed_by VARCHAR(255) NOT NULL,
entity VARCHAR(12) NOT NULL,
entity_id INT NOT NULL,
action VARCHAR(10) NOT NULL,
summary VARCHAR(500) NOT NULL,
before_json TEXT NULL,
after_json TEXT NULL,
INDEX idx_hist_time (changed_at),
INDEX idx_hist_entity (entity, entity_id)
) ENGINE=InnoDB""",
)
_ITEM_FIELDS = ('name', 'kind', 'port_count', 'location', 'rack_u', 'rack_height', 'notes')
_CABLE_FIELDS = ('label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side',
'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes')
class CutsheetConflict(Exception):
"""A change that would clash with existing records (duplicate name or label, item in use)."""
def ensure_cutsheet_schema() -> None:
with get_conn() as conn:
with conn.cursor() as cur:
for statement in _CUTSHEET_SCHEMA:
cur.execute(statement)
def _cs_jsonable(row: Optional[dict]) -> Optional[dict]:
if row is None:
return None
out = {}
for k, v in row.items():
if hasattr(v, 'isoformat'):
v = v.isoformat() + 'Z'
elif v is not None and k == 'length_m':
v = float(v)
elif k in ('label_applied', 'applied'):
v = bool(v)
out[k] = v
return out
def _cs_log(cur, user: str, entity: str, entity_id: int, action: str, summary: str,
before: Optional[dict], after: Optional[dict]) -> None:
cur.execute(
"""INSERT INTO cutsheet_history (changed_by, entity, entity_id, action, summary, before_json, after_json)
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
(user, entity, entity_id, action, summary[:500],
json.dumps(_cs_jsonable(before)) if before else None,
json.dumps(_cs_jsonable(after)) if after else None),
)
@contextmanager
def _cs_transaction():
with get_conn() as conn:
conn.begin()
try:
with conn.cursor() as cur:
yield cur
conn.commit()
except pymysql.err.IntegrityError as e:
conn.rollback()
raise CutsheetConflict(_cs_integrity_message(e)) from e
except Exception:
conn.rollback()
raise
def _cs_integrity_message(e: Exception) -> str:
text = str(e)
if 'uq_cable_label' in text:
return 'Another cable already has that label'
if 'uq_item_name' in text:
return 'Another item already has that name'
return 'That change conflicts with an existing record'
def cutsheet_load() -> dict:
"""Every cutsheet record, for one page render or API response."""
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute('SELECT * FROM cutsheet_items ORDER BY name')
items = [_cs_jsonable(r) for r in cur.fetchall()]
cur.execute('SELECT * FROM cutsheet_cables ORDER BY id')
cables = [_cs_jsonable(r) for r in cur.fetchall()]
cur.execute(
"""SELECT l.item_id, i.name AS item_name, l.port, l.label, l.applied
FROM cutsheet_port_labels l JOIN cutsheet_items i ON i.id = l.item_id"""
)
port_labels = [_cs_jsonable(r) for r in cur.fetchall()]
return {'items': items, 'cables': cables, 'port_labels': port_labels}
def cutsheet_history(limit: int = 100) -> list:
with get_conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT id, changed_at, changed_by, entity, entity_id, action, summary
FROM cutsheet_history ORDER BY id DESC LIMIT %s""",
(limit,),
)
return [_cs_jsonable(r) for r in cur.fetchall()]
def _cs_get(cur, table: str, row_id: int) -> Optional[dict]:
cur.execute(f'SELECT * FROM {table} WHERE id=%s FOR UPDATE', (row_id,)) # nosec B608 (fixed names)
return cur.fetchone()
def cutsheet_create_item(item: dict, user: str) -> int:
with _cs_transaction() as cur:
cur.execute(
f'INSERT INTO cutsheet_items ({", ".join(_ITEM_FIELDS)}) ' # nosec B608 (fixed names)
f'VALUES ({", ".join(["%s"] * len(_ITEM_FIELDS))})',
tuple(item[f] for f in _ITEM_FIELDS),
)
item_id = cur.lastrowid
_cs_log(cur, user, 'item', item_id, 'create', f'Added {item["kind"]} {item["name"]}', None, item)
return item_id
def cutsheet_update_item(item_id: int, item: dict, user: str) -> bool:
"""Update an item. A rename carries over to the cables that reference it."""
with _cs_transaction() as cur:
before = _cs_get(cur, 'cutsheet_items', item_id)
if before is None:
return False
cur.execute(
f'UPDATE cutsheet_items SET {", ".join(f"{f}=%s" for f in _ITEM_FIELDS)} WHERE id=%s', # nosec B608
tuple(item[f] for f in _ITEM_FIELDS) + (item_id,),
)
summary = f'Edited {item["name"]}'
if before['name'] != item['name']:
for end in ('a', 'b'):
cur.execute(
f'UPDATE cutsheet_cables SET {end}_device=%s, updated_by=%s WHERE {end}_device=%s', # nosec B608
(item['name'], user, before['name']),
)
summary = f'Renamed {before["name"]} to {item["name"]}'
_cs_log(cur, user, 'item', item_id, 'update', summary, before, item)
return True
def cutsheet_delete_item(item_id: int, user: str) -> bool:
with _cs_transaction() as cur:
before = _cs_get(cur, 'cutsheet_items', item_id)
if before is None:
return False
cur.execute(
'SELECT COUNT(*) AS n FROM cutsheet_cables WHERE a_device=%s OR b_device=%s',
(before['name'], before['name']),
)
used = cur.fetchone()['n']
if used:
raise CutsheetConflict(f'{before["name"]} still has {used} cable(s); move or delete them first')
cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s', (item_id,))
cur.execute('DELETE FROM cutsheet_items WHERE id=%s', (item_id,))
_cs_log(cur, user, 'item', item_id, 'delete', f'Removed {before["name"]}', before, None)
return True
def _cable_summary(c: dict) -> str:
def end(device: str, port: str) -> str:
return ' '.join(filter(None, [device or 'unterminated', port]))
ends = f'{end(c["a_device"], c["a_port"])} ↔ {end(c["b_device"], c["b_port"])}'
return f'{c.get("label") or "unlabeled cable"}: {ends}'
def cutsheet_create_cable(cable: dict, user: str) -> int:
with _cs_transaction() as cur:
cur.execute(
f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec B608
f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
)
cable_id = cur.lastrowid
_cs_log(cur, user, 'cable', cable_id, 'create', f'Added {_cable_summary(cable)}', None, cable)
return cable_id
def cutsheet_update_cable(cable_id: int, cable: dict, user: str) -> bool:
with _cs_transaction() as cur:
before = _cs_get(cur, 'cutsheet_cables', cable_id)
if before is None:
return False
cur.execute(
f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, updated_by=%s ' # nosec B608
'WHERE id=%s',
tuple(cable[f] for f in _CABLE_FIELDS) + (user, cable_id),
)
_cs_log(cur, user, 'cable', cable_id, 'update', f'Edited {_cable_summary(cable)}', before, cable)
return True
def cutsheet_delete_cable(cable_id: int, user: str) -> bool:
with _cs_transaction() as cur:
before = _cs_get(cur, 'cutsheet_cables', cable_id)
if before is None:
return False
cur.execute('DELETE FROM cutsheet_cables WHERE id=%s', (cable_id,))
_cs_log(cur, user, 'cable', cable_id, 'delete', f'Deleted {_cable_summary(before)}', before, None)
return True
def cutsheet_set_port_label(item_id: int, port: str, label: str, applied: bool, user: str) -> bool:
with _cs_transaction() as cur:
item = _cs_get(cur, 'cutsheet_items', item_id)
if item is None:
return False
cur.execute('SELECT * FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
before = cur.fetchone()
if label:
cur.execute(
"""INSERT INTO cutsheet_port_labels (item_id, port, label, applied) VALUES (%s, %s, %s, %s)
ON DUPLICATE KEY UPDATE label=VALUES(label), applied=VALUES(applied)""",
(item_id, port, label, applied),
)
else:
cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port))
after = {'item_id': item_id, 'port': port, 'label': label, 'applied': applied} if label else None
_cs_log(cur, user, 'port_label', item_id, 'update',
f'{item["name"]} port {port} label: {label or "(cleared)"}', before, after)
return True
def cutsheet_import_cables(cables: list, user: str) -> dict:
"""Add or update cables from a CSV import, matching on label. All or nothing."""
added = updated = 0
with _cs_transaction() as cur:
for cable in cables:
before = None
if cable['label']:
cur.execute('SELECT * FROM cutsheet_cables WHERE label=%s FOR UPDATE', (cable['label'],))
before = cur.fetchone()
if before:
cur.execute(
f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, ' # nosec B608
'updated_by=%s WHERE id=%s',
tuple(cable[f] for f in _CABLE_FIELDS) + (user, before['id']),
)
_cs_log(cur, user, 'cable', before['id'], 'update', f'Imported {_cable_summary(cable)}',
before, cable)
updated += 1
else:
cur.execute(
f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec
f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)',
tuple(cable[f] for f in _CABLE_FIELDS) + (user, user),
)
_cs_log(cur, user, 'cable', cur.lastrowid, 'create', f'Imported {_cable_summary(cable)}',
None, cable)
added += 1
return {'added': added, 'updated': updated}
Reference in New Issue View Git Blame Copy Permalink