From c516df76503cd7d57b6d8ece7f2ceb2c5c4119cf Mon Sep 17 00:00:00 2001 From: mscadm Date: Mon, 13 Jul 2026 12:24:38 +0000 Subject: [PATCH] Datenzugriffsschicht PostgreSQL --- db.py | 272 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 db.py diff --git a/db.py b/db.py new file mode 100644 index 0000000..2793399 --- /dev/null +++ b/db.py @@ -0,0 +1,272 @@ +""" +Datenzugriffsschicht fuer die ITSM-Plattform (PostgreSQL, mandantenfaehig). + +Ersetzt die fruehere JSON-Datei-Ablage. Grund: Betrieb unter ISO 27001, DSGVO +und NIS 2 verlangt belastbare Zugriffskontrolle, Nachvollziehbarkeit (Audit-Log) +und geordnete Backup-/Wiederherstellungsfaehigkeit -- das leistet eine Datei- +Ablage nicht. + +ENV: + DATABASE_URL postgresql://user:pass@host:5432/dbname +""" +import os +import json +import datetime as _dt +from contextlib import contextmanager + +import psycopg2 +import psycopg2.extras +import psycopg2.pool + +DATABASE_URL = os.getenv("DATABASE_URL", "") + +_pool = None + + +def _get_pool(): + global _pool + if _pool is None: + _pool = psycopg2.pool.SimpleConnectionPool(1, 10, dsn=DATABASE_URL) + return _pool + + +@contextmanager +def get_conn(): + pool = _get_pool() + conn = pool.getconn() + try: + yield conn + conn.commit() + except Exception: + conn.rollback() + raise + finally: + pool.putconn(conn) + + +def _dict_cursor(conn): + return conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) + + +def init_db(): + """Legt das Schema an (idempotent). Wird beim App-Start aufgerufen.""" + here = os.path.dirname(os.path.abspath(__file__)) + schema_path = os.path.join(here, "schema.sql") + with open(schema_path, "r", encoding="utf-8") as f: + ddl = f.read() + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute(ddl) + + +def any_tenant_exists(): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute("SELECT 1 FROM tenants LIMIT 1") + return cur.fetchone() is not None + + +# ── Tenants / Ersteinrichtung ────────────────────────────────────────────────── +def create_tenant(name, dsb_name, dsb_email, retention_tickets_days, + retention_audit_days, sla_antwort_minuten, sla_loesung_minuten): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """INSERT INTO tenants + (name, dsb_name, dsb_email, retention_tickets_days, + retention_audit_days, sla_antwort_minuten, sla_loesung_minuten) + VALUES (%s,%s,%s,%s,%s,%s,%s) RETURNING id""", + (name, dsb_name, dsb_email, retention_tickets_days, + retention_audit_days, sla_antwort_minuten, sla_loesung_minuten), + ) + return cur.fetchone()["id"] + + +def get_tenant(tenant_id): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute("SELECT * FROM tenants WHERE id=%s", (tenant_id,)) + return cur.fetchone() + + +def update_tenant_settings(tenant_id, dsb_name, dsb_email, retention_tickets_days, + retention_audit_days, sla_antwort_minuten, sla_loesung_minuten): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """UPDATE tenants SET dsb_name=%s, dsb_email=%s, + retention_tickets_days=%s, retention_audit_days=%s, + sla_antwort_minuten=%s, sla_loesung_minuten=%s + WHERE id=%s""", + (dsb_name, dsb_email, retention_tickets_days, retention_audit_days, + sla_antwort_minuten, sla_loesung_minuten, tenant_id), + ) + + +# ── Users ─────────────────────────────────────────────────────────────────────── +def create_user(tenant_id, email, password_hash, role="admin"): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """INSERT INTO users (tenant_id, email, password_hash, role) + VALUES (%s,%s,%s,%s) RETURNING id""", + (tenant_id, email, password_hash, role), + ) + return cur.fetchone()["id"] + + +def get_user_by_email(email): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute("SELECT * FROM users WHERE email=%s AND active=TRUE", (email,)) + return cur.fetchone() + + +def touch_last_login(user_id): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute("UPDATE users SET last_login_at=now() WHERE id=%s", (user_id,)) + + +def email_exists(email): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute("SELECT 1 FROM users WHERE email=%s", (email,)) + return cur.fetchone() is not None + + +# ── Audit-Log ─────────────────────────────────────────────────────────────────── +def log_audit(tenant_id, user_id, aktion, entity_typ=None, entity_id=None, details=None, ip=None): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """INSERT INTO audit_log (tenant_id, user_id, aktion, entity_typ, entity_id, details, ip) + VALUES (%s,%s,%s,%s,%s,%s,%s)""", + (tenant_id, user_id, aktion, entity_typ, entity_id, + json.dumps(details) if details is not None else None, ip), + ) + + +def list_audit(tenant_id, limit=200): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """SELECT a.*, u.email AS user_email FROM audit_log a + LEFT JOIN users u ON u.id = a.user_id + WHERE a.tenant_id=%s ORDER BY a.zeit DESC LIMIT %s""", + (tenant_id, limit), + ) + return cur.fetchall() + + +# ── Services ──────────────────────────────────────────────────────────────────── +def create_service(tenant_id, name, beschreibung, kategorie, gebucht, endpoint): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """INSERT INTO services (tenant_id, name, beschreibung, kategorie, gebucht, endpoint) + VALUES (%s,%s,%s,%s,%s,%s) RETURNING id""", + (tenant_id, name, beschreibung, kategorie, gebucht, endpoint), + ) + return cur.fetchone()["id"] + + +def list_services(tenant_id): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute("SELECT * FROM services WHERE tenant_id=%s ORDER BY id", (tenant_id,)) + return cur.fetchall() + + +# ── Tickets ───────────────────────────────────────────────────────────────────── +def _next_ticket_nr(tenant_id): + year = _dt.datetime.utcnow().year + prefix = "TKT-%d-" % year + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """SELECT ticket_nr FROM tickets + WHERE tenant_id=%s AND ticket_nr LIKE %s + ORDER BY ticket_nr DESC LIMIT 1""", + (tenant_id, prefix + "%"), + ) + row = cur.fetchone() + seq = 1 + if row: + try: + seq = int(row[0].split("-")[-1]) + 1 + except Exception: + seq = 1 + return "%s%06d" % (prefix, seq) + + +def create_ticket(tenant_id, titel, beschreibung, service_id, prioritaet, kategorie, + zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten, + actor_label): + ticket_nr = _next_ticket_nr(tenant_id) + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """INSERT INTO tickets + (tenant_id, ticket_nr, titel, beschreibung, service_id, prioritaet, + kategorie, zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) RETURNING id""", + (tenant_id, ticket_nr, titel, beschreibung, service_id, prioritaet, + kategorie, zugewiesen_an, ersteller_id, sla_antwort_minuten, sla_loesung_minuten), + ) + tid = cur.fetchone()["id"] + cur.execute( + """INSERT INTO ticket_timeline (ticket_id, tenant_id, akteur, text) + VALUES (%s,%s,%s,%s)""", + (tid, tenant_id, actor_label, "Ticket angelegt."), + ) + return tid, ticket_nr + + +def list_tickets(tenant_id): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """SELECT t.*, s.name AS service_name FROM tickets t + LEFT JOIN services s ON s.id = t.service_id + WHERE t.tenant_id=%s ORDER BY t.updated_at DESC""", + (tenant_id,), + ) + return cur.fetchall() + + +def get_ticket(tenant_id, ticket_id): + with get_conn() as conn: + with _dict_cursor(conn) as cur: + cur.execute( + """SELECT t.*, s.name AS service_name FROM tickets t + LEFT JOIN services s ON s.id = t.service_id + WHERE t.tenant_id=%s AND t.id=%s""", + (tenant_id, ticket_id), + ) + ticket = cur.fetchone() + if not ticket: + return None + cur.execute( + "SELECT * FROM ticket_timeline WHERE ticket_id=%s ORDER BY zeit DESC", + (ticket_id,), + ) + ticket = dict(ticket) + ticket["timeline"] = cur.fetchall() + return ticket + + +def update_ticket_status(tenant_id, ticket_id, status, actor_label): + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """UPDATE tickets SET status=%s, updated_at=now(), + fortschritt = CASE WHEN %s IN ('Geloest','Geschlossen') THEN 100 ELSE fortschritt END + WHERE tenant_id=%s AND id=%s""", + (status, status, tenant_id, ticket_id), + ) + cur.execute( + """INSERT INTO ticket_timeline (ticket_id, tenant_id, akteur, text) + VALUES (%s,%s,%s,%s)""", + (ticket_id, tenant_id, actor_label, "Status geaendert auf '%s'." % status), + )