Datenzugriffsschicht PostgreSQL
This commit is contained in:
parent
ccfd1614a5
commit
c516df7650
|
|
@ -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),
|
||||
)
|
||||
Loading…
Reference in New Issue