Lagerorte per 10-Zeichen-Code statt fortlaufender ID; iOS-Einlagern ohne Kamerazwang

Lagerort-IDs sind jetzt ein zufaelliger 10-Zeichen-Code (wie die Einzelstueck-UIDs) statt einer fortlaufenden Zahl - so kollidiert die Stammdaten-Sicherung zwischen zwei Instanzen praktisch nie mehr, und der Code ist zugleich der Inhalt des QR /l/<code>. Alle Fremdschluessel (lots, movements, items, Mindestbestaende, parent_id) ziehen mit; die Umstellung laeuft einmalig und transaktional beim Serverstart (_migrate_locations_to_code) und rollt bei Fehlern komplett zurueck. Vor dem Deploy ein DB-Backup machen.

iOS-Einlagern oeffnet nicht mehr sofort die Kamera, sondern ein Formular mit Artikelsuche; die Kamera kommt erst per Button. Im Formular laesst sich der Lagerort zusaetzlich per /l/-QR scannen.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-27 12:15:25 +02:00
parent 6e48cd0a7a
commit d518b4aa23
23 changed files with 507 additions and 220 deletions

View File

@@ -39,6 +39,95 @@ from .services.group_codes import backfill as backfill_group_codes
settings = get_settings()
def _migrate_locations_to_code(conn) -> None:
"""Einmalige Umstellung: Lagerort-ID von fortlaufender Zahl auf 10-Zeichen-Code.
Läuft nur, solange ``locations.id`` noch eine Integer-Spalte ist danach ist
die Umstellung erledigt und der Block überspringt sich selbst. Alles passiert
innerhalb der umgebenden Transaktion: Bricht ein Schritt ab, wird komplett
zurückgerollt und die Datenbank bleibt auf dem alten (funktionierenden) Stand.
"""
id_type = conn.execute(text(
"SELECT data_type FROM information_schema.columns "
"WHERE table_name = 'locations' AND column_name = 'id'"
)).scalar()
if id_type not in ("integer", "bigint", "smallint"):
return # frische Installation (schon VARCHAR) oder bereits umgestellt
from .models import generate_location_code
# 1. Jedem bestehenden Lagerort einen eindeutigen Code geben (id bleibt vorerst).
conn.execute(text("ALTER TABLE locations ADD COLUMN IF NOT EXISTS code VARCHAR(10)"))
vergeben: set[str] = set()
for (lid,) in conn.execute(text("SELECT id FROM locations")).all():
code = generate_location_code()
while code in vergeben:
code = generate_location_code()
vergeben.add(code)
conn.execute(text("UPDATE locations SET code = :c WHERE id = :i"), {"c": code, "i": lid})
# 2. Fremdschlüssel-Spalten (Integer) auf den Code umziehen. Beim DROP COLUMN
# fallen die alten FK-/Unique-Constraints automatisch mit weg.
kinder = [
# (Tabelle, ON DELETE, NOT NULL danach, Unique-Constraint danach)
("lots", "SET NULL", False, None),
("movements", "SET NULL", False, None),
("items", "SET NULL", False, None),
("product_location_min_stock", "CASCADE", True, ("uq_prod_loc_min", "product_id")),
("group_location_min_stock", "CASCADE", True, ("uq_group_loc_min", "group_id")),
]
vorhanden = []
for tabelle, ond, nn, uq in kinder:
spalte_da = conn.execute(text(
"SELECT 1 FROM information_schema.columns "
"WHERE table_name = :t AND column_name = 'location_id'"
), {"t": tabelle}).scalar()
if spalte_da is None:
continue # Spalte existiert (noch) nicht nichts umzuziehen
vorhanden.append((tabelle, ond, nn, uq))
conn.execute(text(f"ALTER TABLE {tabelle} ADD COLUMN location_code VARCHAR(10)"))
conn.execute(text(
f"UPDATE {tabelle} t SET location_code = l.code "
f"FROM locations l WHERE t.location_id = l.id"
))
conn.execute(text(f"ALTER TABLE {tabelle} DROP COLUMN location_id"))
conn.execute(text(f"ALTER TABLE {tabelle} RENAME COLUMN location_code TO location_id"))
# 2b. Selbstverweis parent_id ebenso auf den Code umziehen.
conn.execute(text("ALTER TABLE locations ADD COLUMN parent_code VARCHAR(10)"))
conn.execute(text(
"UPDATE locations c SET parent_code = p.code FROM locations p WHERE c.parent_id = p.id"
))
conn.execute(text("ALTER TABLE locations DROP COLUMN parent_id"))
# 3. Integer-PK durch den Code ersetzen (DROP COLUMN zieht PK und Sequenz mit).
conn.execute(text("ALTER TABLE locations DROP COLUMN id"))
conn.execute(text("ALTER TABLE locations RENAME COLUMN code TO id"))
conn.execute(text("ALTER TABLE locations ALTER COLUMN id SET NOT NULL"))
conn.execute(text("ALTER TABLE locations ADD PRIMARY KEY (id)"))
conn.execute(text("ALTER TABLE locations RENAME COLUMN parent_code TO parent_id"))
conn.execute(text(
"ALTER TABLE locations ADD CONSTRAINT locations_parent_id_fkey "
"FOREIGN KEY (parent_id) REFERENCES locations(id) ON DELETE SET NULL"
))
# 4. Fremdschlüssel (und Pflicht-/Unique-Bedingungen) neu setzen jetzt auf den Code.
for tabelle, ond, nn, uq in vorhanden:
if nn:
# Verwaiste Mindestbestände (Lagerort gelöscht) vorher entfernen.
conn.execute(text(f"DELETE FROM {tabelle} WHERE location_id IS NULL"))
conn.execute(text(f"ALTER TABLE {tabelle} ALTER COLUMN location_id SET NOT NULL"))
conn.execute(text(
f"ALTER TABLE {tabelle} ADD CONSTRAINT {tabelle}_location_id_fkey "
f"FOREIGN KEY (location_id) REFERENCES locations(id) ON DELETE {ond}"
))
if uq is not None:
name, andere = uq
conn.execute(text(
f"ALTER TABLE {tabelle} ADD CONSTRAINT {name} UNIQUE ({andere}, location_id)"
))
def _ensure_schema() -> None:
"""Schonende Migration: fehlende Spalten auf bestehenden Tabellen nachziehen.
@@ -79,7 +168,9 @@ def _ensure_schema() -> None:
"REFERENCES shops(id) ON DELETE SET NULL",
"ALTER TABLE products ADD COLUMN IF NOT EXISTS product_url VARCHAR(1024)",
# Bewegungen: Lagerort (Gegenstands-Buchungen) und Entnahmegrund.
"ALTER TABLE movements ADD COLUMN IF NOT EXISTS location_id INTEGER "
# location_id ist ein Lagerort-Code (VARCHAR(10)), siehe
# _migrate_locations_to_code deshalb hier gleich als VARCHAR anlegen.
"ALTER TABLE movements ADD COLUMN IF NOT EXISTS location_id VARCHAR(10) "
"REFERENCES locations(id) ON DELETE SET NULL",
"ALTER TABLE movements ADD COLUMN IF NOT EXISTS reason VARCHAR(16)",
# Einzelstück-Verwaltung (Items mit UID/QR) je Produkt.
@@ -90,6 +181,9 @@ def _ensure_schema() -> None:
"ALTER TABLE items ADD COLUMN IF NOT EXISTS currency VARCHAR(3)",
]
with engine.begin() as conn:
# Zuerst die Lagerort-ID auf den Code umstellen (einmalig, idempotent),
# damit die folgenden ADD-COLUMN-Verweise auf die neue VARCHAR-id passen.
_migrate_locations_to_code(conn)
for stmt in stmts:
conn.execute(text(stmt))
# Solange user_id eindeutig war, ging genau ein Dashboard je Benutzer.