Aktivierungscodes verwalten: Bezeichnung, Gueltigkeit, Nutzung, Status/Loeschen

Migration 0056: bezeichnung, gueltig_bis, verwendet_anzahl,
zuletzt_verwendet (alle optional, bestehende Codes unbegrenzt gueltig).
Abgelaufene Codes werden bei der Aktivierung abgelehnt, erfolgreiche
Aktivierungen gezaehlt. Neue Endpunkte: Status setzen, Loeschen; Anlegen
nimmt optional bezeichnung/gueltig_tage. GUI-Runde 10/2026 Punkt 2.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
adminandClaude Opus 5.5 committed 2026-10-09 11:25:38 +02:00
1 parent f1435018ec
commit 3381b2b13c
2 files changed
+119 -7

No files matched your search

+107 -7
View File
@@ -123,6 +123,15 @@ class ActivationRequest(BaseModel):
hardware: dict
class CreateActivationCodeRequest(BaseModel):
bezeichnung: str | None = None
gueltig_tage: int | None = None
class ActivationCodeStatusRequest(BaseModel):
active: bool
class ResolveRequest(BaseModel):
device_id: str
@@ -627,6 +636,7 @@ def find_activation_code(code: str):
FROM activation_codes ac
JOIN organizations o ON o.id = ac.organization_id
WHERE ac.code = %s AND ac.active
AND (ac.gueltig_bis IS NULL OR ac.gueltig_bis > now())
""",
(code,),
)
@@ -637,6 +647,23 @@ def find_activation_code(code: str):
return {"organization_id": str(row[0]), "organization_name": row[1]}
def mark_activation_code_used(code: str):
"""Nutzung zaehlen (GUI-Runde 09.10.2026) - nur nach erfolgreicher
Aktivierung, damit die Verwaltungsseite zeigt, ob ein Code noch
gebraucht wird."""
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE activation_codes
SET verwendet_anzahl = verwendet_anzahl + 1, zuletzt_verwendet = now()
WHERE code = %s
""",
(code,),
)
def find_enrollment_session(code: str):
"""
Enrollment Sessions haben kein eigenes code-Feld - ihre id dient direkt
@@ -2973,7 +3000,9 @@ def fetch_activation_codes(organization_id: str):
with conn.cursor() as cur:
cur.execute(
"""
SELECT code, active, created_at
SELECT code, active, created_at, bezeichnung, gueltig_bis,
verwendet_anzahl, zuletzt_verwendet,
(gueltig_bis IS NOT NULL AND gueltig_bis <= now()) AS abgelaufen
FROM activation_codes
WHERE organization_id = %s
ORDER BY created_at DESC
@@ -2981,11 +3010,19 @@ def fetch_activation_codes(organization_id: str):
(organization_id,),
)
return [
{"code": code, "active": active, "created_at": created_at.isoformat()}
for code, active, created_at in cur.fetchall()
{
"code": code, "active": active, "created_at": created_at.isoformat(),
"bezeichnung": bezeichnung,
"gueltig_bis": gueltig_bis.isoformat() if gueltig_bis else None,
"verwendet_anzahl": verwendet_anzahl,
"zuletzt_verwendet": zuletzt_verwendet.isoformat() if zuletzt_verwendet else None,
"abgelaufen": abgelaufen,
}
for code, active, created_at, bezeichnung, gueltig_bis, verwendet_anzahl,
zuletzt_verwendet, abgelaufen in cur.fetchall()
]
def create_activation_code(organization_id: str):
def create_activation_code(organization_id: str, bezeichnung: str | None = None, gueltig_tage: int | None = None):
code = "-".join(
secrets.token_hex(3).upper()[i:i + 3] for i in range(0, 6, 3)
)
@@ -2993,12 +3030,35 @@ def create_activation_code(organization_id: str):
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO activation_codes (code, organization_id) VALUES (%s, %s)",
(code, organization_id),
"""
INSERT INTO activation_codes (code, organization_id, bezeichnung, gueltig_bis)
VALUES (%s, %s, %s, CASE WHEN %s::int IS NULL THEN NULL ELSE now() + make_interval(days => %s::int) END)
""",
(code, organization_id, bezeichnung or None, gueltig_tage, gueltig_tage),
)
return {"code": code, "active": True}
def set_activation_code_active(organization_id: str, code: str, active: bool) -> bool:
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE activation_codes SET active = %s WHERE code = %s AND organization_id = %s",
(active, code, organization_id),
)
return cur.rowcount == 1
def delete_activation_code(organization_id: str, code: str) -> bool:
with get_database_connection() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM activation_codes WHERE code = %s AND organization_id = %s",
(code, organization_id),
)
return cur.rowcount == 1
def fetch_merkmale():
with get_database_connection() as conn:
with conn.cursor() as cur:
@@ -4891,6 +4951,9 @@ def activate(payload: ActivationRequest, request: Request):
payload.hardware,
)
if activation is not None:
mark_activation_code_used(payload.activation_code)
if enrollment_session is not None:
consume_enrollment_session(enrollment_session["session_id"])
if enrollment_session.get("ziel_oe_id"):
@@ -5644,6 +5707,7 @@ def list_activation_codes(
@app.post("/api/v1/organizations/{organization_id}/activation-codes")
def create_activation_code_endpoint(
organization_id: str,
payload: CreateActivationCodeRequest | None = None,
authorization: str | None = Header(default=None),
):
if not require_service_token(authorization):
@@ -5653,7 +5717,43 @@ def create_activation_code_endpoint(
"message": "Fehlendes oder ungültiges Service-Token.",
}
return {"success": True, "activation_code": create_activation_code(organization_id)}
bezeichnung = (payload.bezeichnung if payload else None) or None
gueltig_tage = payload.gueltig_tage if payload else None
if gueltig_tage is not None and not (1 <= gueltig_tage <= 3650):
return {"success": False, "error": "invalid_gueltigkeit", "message": "Gültigkeit muss zwischen 1 und 3650 Tagen liegen."}
return {"success": True, "activation_code": create_activation_code(organization_id, bezeichnung, gueltig_tage)}
@app.post("/api/v1/organizations/{organization_id}/activation-codes/{code}/status")
def set_activation_code_status_endpoint(
organization_id: str,
code: str,
payload: ActivationCodeStatusRequest,
authorization: str | None = Header(default=None),
):
if not require_service_token(authorization):
return {"success": False, "error": "unauthorized", "message": "Fehlendes oder ungültiges Service-Token."}
if not set_activation_code_active(organization_id, code, payload.active):
return {"success": False, "error": "not_found", "message": "Aktivierungscode nicht gefunden."}
return {"success": True}
@app.delete("/api/v1/organizations/{organization_id}/activation-codes/{code}")
def delete_activation_code_endpoint(
organization_id: str,
code: str,
authorization: str | None = Header(default=None),
):
if not require_service_token(authorization):
return {"success": False, "error": "unauthorized", "message": "Fehlendes oder ungültiges Service-Token."}
if not delete_activation_code(organization_id, code):
return {"success": False, "error": "not_found", "message": "Aktivierungscode nicht gefunden."}
return {"success": True}
@app.get("/api/v1/merkmale")