rene/banyaro
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
banyaro/backend/routes/zucht_hunde.py
rene 91340be5a3 Feature: Vollständige Züchter-Rolle — Antrag, Würfe, Stammbaum, Genetik 
Basis-Features (Schritte 1–11):
- Züchter-Antrag mit Dokument-Upload, Admin-Prüfung, E-Mail-Benachrichtigungen
- Öffentliches Züchter-Profil + Karten-Marker (lila, certificate-Icon)
- Wurfverwaltung: Würfe, Welpen, Gewichtsverlauf, Foto-System
- Wurfbörse (öffentlich) mit Filtersuche nach Rasse/Status
- Läufigkeits-Tracker: Deckdatum + Wurftermin (+63 Tage, nur für Züchter)
- Interessenten-Chat: Kontakt-Button in Wurfbörse und Züchter-Profil
- Sidebar-Einträge: Zuchtkartei + Wurfverwaltung für Züchter/Admin

Stammbaum & Genetik (Schritte 1–8):
- Zuchtkartei: Hunde-Stammdaten mit Vater/Mutter-Verknüpfung
- Stammbaum-Visualisierung: 4 Generationen, horizontales CSS-Grid
- Gesundheitstests (HD, ED, OCD, Augen…) mit farbigen Ergebnis-Badges
- Genetische Tests (MDR1, PRA, DM…): clear/carrier/affected
- Titel & Auszeichnungen (CAC, CACIB, IPO…)
- Probeverpaarung: IK-Berechnung nach Wright + Ampel-Bewertung
- Teilen-Link für öffentliche Hunde-Profile
- Kaufvertrag: druckbares HTML-Dokument pro Welpe

Technisch: 4 neue Route-Dateien, 5 neue Page-Module, 11 neue DB-Tabellen,
icons shield-check + certificate + tree-structure im Sprite — SW by-v465, APP_VER 444
2026-04-28 18:25:21 +02:00

779 lines
29 KiB
Python
Raw Blame History

"""BAN YARO — Zuchtkartei (Hunde, Gesundheitstests, Gentests, Titel, Stammbaum, IK)"""
import logging
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from typing import Optional
from database import db
from auth import get_current_user, get_current_user_optional
router = APIRouter()
logger = logging.getLogger(__name__)
# ------------------------------------------------------------------
# Dependency: nur verifizierte Züchter + Admins
# ------------------------------------------------------------------
def _require_breeder(user=Depends(get_current_user)):
if user["rolle"] not in ("breeder", "admin"):
raise HTTPException(403, "Nur für Züchter.")
return user
# ------------------------------------------------------------------
# Hilfsfunktionen: Ownership
# ------------------------------------------------------------------
def _get_breeder_profile_id(user_id: int, conn) -> Optional[int]:
"""Gibt die breeder_profiles.id des Users zurück, oder None."""
row = conn.execute(
"SELECT id FROM breeder_profiles WHERE user_id=?", (user_id,)
).fetchone()
return row["id"] if row else None
def _check_hund_owner(hund_id: int, user: dict, conn) -> dict:
"""Gibt den Hund zurück wenn der User Eigentümer oder Admin ist."""
row = conn.execute(
"""SELECT zh.*, bp.user_id AS owner_user_id
FROM zucht_hunde zh
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE zh.id=?""",
(hund_id,)
).fetchone()
if not row:
raise HTTPException(404, "Hund nicht gefunden.")
if user["rolle"] != "admin" and row["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
return dict(row)
def _check_hund_access(hund_id: int, user: Optional[dict], conn) -> dict:
"""Zugriff auf Hund: öffentlich wenn is_public=1, sonst nur Owner/Admin."""
row = conn.execute(
"""SELECT zh.*, bp.user_id AS owner_user_id
FROM zucht_hunde zh
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE zh.id=?""",
(hund_id,)
).fetchone()
if not row:
raise HTTPException(404, "Hund nicht gefunden.")
is_owner = user and (
user["rolle"] == "admin" or row["owner_user_id"] == user["id"]
)
if not row["is_public"] and not is_owner:
raise HTTPException(404, "Hund nicht gefunden.")
return dict(row)
# ------------------------------------------------------------------
# Stammbaum-Algorithmus
# ------------------------------------------------------------------
def _build_tree(conn, hund_id, depth: int):
if depth == 0 or hund_id is None:
return None
row = conn.execute(
"SELECT * FROM zucht_hunde WHERE id=?", (hund_id,)
).fetchone()
if not row:
return None
d = dict(row)
d["vater"] = _build_tree(conn, d["vater_id"], depth - 1)
d["mutter"] = _build_tree(conn, d["mutter_id"], depth - 1)
return d
# ------------------------------------------------------------------
# Inzucht-Koeffizient (Wright's Formel)
# ------------------------------------------------------------------
def _get_ancestors(conn, hund_id, depth: int, path: list) -> dict:
"""Gibt {ancestor_id: [paths]} zurück."""
if depth == 0 or hund_id is None:
return {}
row = conn.execute(
"SELECT vater_id, mutter_id, name FROM zucht_hunde WHERE id=?", (hund_id,)
).fetchone()
if not row:
return {}
result = {hund_id: [path]}
for parent_id in [row["vater_id"], row["mutter_id"]]:
if parent_id:
sub = _get_ancestors(conn, parent_id, depth - 1, path + [hund_id])
for aid, paths in sub.items():
result.setdefault(aid, []).extend(paths)
return result
def _calculate_ik(conn, vater_id, mutter_id, generations: int = 8) -> float:
fa = _get_ancestors(conn, vater_id, generations, [])
ma = _get_ancestors(conn, mutter_id, generations, [])
common = set(fa.keys()) & set(ma.keys())
ik = 0.0
for aid in common:
for pf in fa[aid]:
for pm in ma[aid]:
ik += 0.5 ** (len(pf) + len(pm) + 1)
return round(ik * 100, 2)
def _ik_rating(ik: float) -> str:
if ik < 2.5:
return "optimal"
if ik < 6.25:
return "akzeptabel"
if ik < 12.5:
return "erhoeht"
return "kritisch"
# ------------------------------------------------------------------
# Pydantic-Schemas
# ------------------------------------------------------------------
class HundCreate(BaseModel):
name: str
rufname: Optional[str] = None
geschlecht: str # maennlich|weiblich
geburtsdatum: Optional[str] = None
sterbedatum: Optional[str] = None
chip_nr: Optional[str] = None
taetowiernummer: Optional[str] = None
zuchtbuchnummer: Optional[str] = None
farbe: Optional[str] = None
vater_id: Optional[int] = None
mutter_id: Optional[int] = None
zuechter_name: Optional[str] = None
eigentuemer_name: Optional[str] = None
is_public: int = 1
notiz: Optional[str] = None
foto_url: Optional[str] = None
class HundUpdate(BaseModel):
name: Optional[str] = None
rufname: Optional[str] = None
geschlecht: Optional[str] = None
geburtsdatum: Optional[str] = None
sterbedatum: Optional[str] = None
chip_nr: Optional[str] = None
taetowiernummer: Optional[str] = None
zuchtbuchnummer: Optional[str] = None
farbe: Optional[str] = None
vater_id: Optional[int] = None
mutter_id: Optional[int] = None
zuechter_name: Optional[str] = None
eigentuemer_name: Optional[str] = None
is_public: Optional[int] = None
notiz: Optional[str] = None
foto_url: Optional[str] = None
class HealthTestCreate(BaseModel):
test_typ: str # HD|ED|OCD|augen|herz|patella|ZTP|custom
test_name: Optional[str] = None
ergebnis: Optional[str] = None
untersuch_am: Optional[str] = None
gueltig_bis: Optional[str] = None
untersucher: Optional[str] = None
labor: Optional[str] = None
zertifikat_nr: Optional[str] = None
is_public: int = 1
class HealthTestUpdate(BaseModel):
test_typ: Optional[str] = None
test_name: Optional[str] = None
ergebnis: Optional[str] = None
untersuch_am: Optional[str] = None
gueltig_bis: Optional[str] = None
untersucher: Optional[str] = None
labor: Optional[str] = None
zertifikat_nr: Optional[str] = None
is_public: Optional[int] = None
class GeneticTestCreate(BaseModel):
marker_name: str # MDR1|PRA-prcd|DM|vWD|HUU etc.
marker_kategorie: Optional[str] = None # krankheit|farbe|eigenschaft
genotyp: Optional[str] = None # +/+|+/-|-/-
ergebnis_klasse: Optional[str] = None # clear|carrier|affected
getestet_am: Optional[str] = None
labor: Optional[str] = None
zertifikat_nr: Optional[str] = None
is_public: int = 1
class GeneticTestUpdate(BaseModel):
marker_name: Optional[str] = None
marker_kategorie: Optional[str] = None
genotyp: Optional[str] = None
ergebnis_klasse: Optional[str] = None
getestet_am: Optional[str] = None
labor: Optional[str] = None
zertifikat_nr: Optional[str] = None
is_public: Optional[int] = None
class TitelCreate(BaseModel):
titel_typ: str # ausstellung|arbeit|sport|zucht|champion|custom
titel_name: str
verliehen_am: Optional[str] = None
ort: Optional[str] = None
richter: Optional[str] = None
ausstellung: Optional[str] = None
formwert: Optional[str] = None
is_public: int = 1
class TitelUpdate(BaseModel):
titel_typ: Optional[str] = None
titel_name: Optional[str] = None
verliehen_am: Optional[str] = None
ort: Optional[str] = None
richter: Optional[str] = None
ausstellung: Optional[str] = None
formwert: Optional[str] = None
is_public: Optional[int] = None
class TrialMatingBody(BaseModel):
vater_id: int
mutter_id: int
# ==================================================================
# HUNDE CRUD
# ==================================================================
# ------------------------------------------------------------------
# GET /api/zuchthunde — eigene Hunde
# ------------------------------------------------------------------
@router.get("/zuchthunde")
async def list_eigene_hunde(user=Depends(_require_breeder)):
with db() as conn:
if user["rolle"] == "admin":
profile_id = _get_breeder_profile_id(user["id"], conn)
if profile_id is None:
# Admin ohne Profil sieht alle Hunde
rows = conn.execute(
"SELECT * FROM zucht_hunde ORDER BY name ASC"
).fetchall()
return [dict(r) for r in rows]
else:
profile_id = _get_breeder_profile_id(user["id"], conn)
if profile_id is None:
raise HTTPException(404, "Kein Züchter-Profil vorhanden.")
rows = conn.execute(
"SELECT * FROM zucht_hunde WHERE breeder_id=? ORDER BY name ASC",
(profile_id,)
).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------
# POST /api/zuchthunde — Hund anlegen
# ------------------------------------------------------------------
@router.post("/zuchthunde", status_code=201)
async def create_hund(body: HundCreate, user=Depends(_require_breeder)):
with db() as conn:
profile_id = _get_breeder_profile_id(user["id"], conn)
if profile_id is None and user["rolle"] != "admin":
raise HTTPException(404, "Kein Züchter-Profil vorhanden.")
cur = conn.execute(
"""INSERT INTO zucht_hunde
(breeder_id, name, rufname, geschlecht, geburtsdatum, sterbedatum,
chip_nr, taetowiernummer, zuchtbuchnummer, farbe,
vater_id, mutter_id, zuechter_name, eigentuemer_name,
is_public, notiz, foto_url)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
(
profile_id,
body.name, body.rufname, body.geschlecht,
body.geburtsdatum, body.sterbedatum,
body.chip_nr, body.taetowiernummer, body.zuchtbuchnummer,
body.farbe, body.vater_id, body.mutter_id,
body.zuechter_name, body.eigentuemer_name,
body.is_public, body.notiz, body.foto_url,
)
)
row = conn.execute(
"SELECT * FROM zucht_hunde WHERE id=?", (cur.lastrowid,)
).fetchone()
return dict(row)
# ==================================================================
# FIXE ROUTEN vor {id} — Route-Reihenfolge kritisch!
# ==================================================================
# ------------------------------------------------------------------
# POST /api/zuchthunde/trial-mating — Probeverpaarung / IK-Berechnung
# ------------------------------------------------------------------
@router.post("/zuchthunde/trial-mating")
async def trial_mating(body: TrialMatingBody, user=Depends(_require_breeder)):
with db() as conn:
vater = conn.execute(
"SELECT id, name FROM zucht_hunde WHERE id=?", (body.vater_id,)
).fetchone()
if not vater:
raise HTTPException(404, "Vater nicht gefunden.")
mutter = conn.execute(
"SELECT id, name FROM zucht_hunde WHERE id=?", (body.mutter_id,)
).fetchone()
if not mutter:
raise HTTPException(404, "Mutter nicht gefunden.")
ik_prozent = _calculate_ik(conn, body.vater_id, body.mutter_id, generations=8)
rating = _ik_rating(ik_prozent)
# Gemeinsame Vorfahren mit Namen ermitteln
fa = _get_ancestors(conn, body.vater_id, 8, [])
ma = _get_ancestors(conn, body.mutter_id, 8, [])
common_ids = set(fa.keys()) & set(ma.keys())
gemeinsame_vorfahren = []
for aid in common_ids:
anc = conn.execute(
"SELECT id, name FROM zucht_hunde WHERE id=?", (aid,)
).fetchone()
if not anc:
continue
# Minimale Pfadlängen für Anzeige
min_gen_vater = min(len(p) for p in fa[aid])
min_gen_mutter = min(len(p) for p in ma[aid])
gemeinsame_vorfahren.append({
"id": anc["id"],
"name": anc["name"],
"gen_vater": min_gen_vater,
"gen_mutter": min_gen_mutter,
})
gemeinsame_vorfahren.sort(key=lambda x: x["gen_vater"] + x["gen_mutter"])
return {
"ik_prozent": ik_prozent,
"ik_rating": rating,
"gemeinsame_vorfahren": gemeinsame_vorfahren,
}
# ------------------------------------------------------------------
# PUT /api/zuchthunde/health-tests/{tid}
# ------------------------------------------------------------------
@router.put("/zuchthunde/health-tests/{tid}")
async def update_health_test(tid: int, body: HealthTestUpdate, user=Depends(_require_breeder)):
with db() as conn:
test = conn.execute(
"""SELECT ht.*, bp.user_id AS owner_user_id
FROM dog_health_tests ht
JOIN zucht_hunde zh ON zh.id = ht.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE ht.id=?""",
(tid,)
).fetchone()
if not test:
raise HTTPException(404, "Gesundheitstest nicht gefunden.")
if user["rolle"] != "admin" and test["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
fields, params = [], []
for field, value in body.model_dump(exclude_none=True).items():
fields.append(f"{field}=?")
params.append(value)
if not fields:
raise HTTPException(400, "Keine Felder zum Aktualisieren.")
params.append(tid)
conn.execute(
f"UPDATE dog_health_tests SET {', '.join(fields)} WHERE id=?", params
)
row = conn.execute(
"SELECT * FROM dog_health_tests WHERE id=?", (tid,)
).fetchone()
return dict(row)
# ------------------------------------------------------------------
# DELETE /api/zuchthunde/health-tests/{tid}
# ------------------------------------------------------------------
@router.delete("/zuchthunde/health-tests/{tid}", status_code=204)
async def delete_health_test(tid: int, user=Depends(_require_breeder)):
with db() as conn:
test = conn.execute(
"""SELECT ht.id, bp.user_id AS owner_user_id
FROM dog_health_tests ht
JOIN zucht_hunde zh ON zh.id = ht.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE ht.id=?""",
(tid,)
).fetchone()
if not test:
raise HTTPException(404, "Gesundheitstest nicht gefunden.")
if user["rolle"] != "admin" and test["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
conn.execute("DELETE FROM dog_health_tests WHERE id=?", (tid,))
return None
# ------------------------------------------------------------------
# PUT /api/zuchthunde/genetic-tests/{tid}
# ------------------------------------------------------------------
@router.put("/zuchthunde/genetic-tests/{tid}")
async def update_genetic_test(tid: int, body: GeneticTestUpdate, user=Depends(_require_breeder)):
with db() as conn:
test = conn.execute(
"""SELECT gt.*, bp.user_id AS owner_user_id
FROM dog_genetic_tests gt
JOIN zucht_hunde zh ON zh.id = gt.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE gt.id=?""",
(tid,)
).fetchone()
if not test:
raise HTTPException(404, "Gentest nicht gefunden.")
if user["rolle"] != "admin" and test["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
fields, params = [], []
for field, value in body.model_dump(exclude_none=True).items():
fields.append(f"{field}=?")
params.append(value)
if not fields:
raise HTTPException(400, "Keine Felder zum Aktualisieren.")
params.append(tid)
conn.execute(
f"UPDATE dog_genetic_tests SET {', '.join(fields)} WHERE id=?", params
)
row = conn.execute(
"SELECT * FROM dog_genetic_tests WHERE id=?", (tid,)
).fetchone()
return dict(row)
# ------------------------------------------------------------------
# DELETE /api/zuchthunde/genetic-tests/{tid}
# ------------------------------------------------------------------
@router.delete("/zuchthunde/genetic-tests/{tid}", status_code=204)
async def delete_genetic_test(tid: int, user=Depends(_require_breeder)):
with db() as conn:
test = conn.execute(
"""SELECT gt.id, bp.user_id AS owner_user_id
FROM dog_genetic_tests gt
JOIN zucht_hunde zh ON zh.id = gt.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE gt.id=?""",
(tid,)
).fetchone()
if not test:
raise HTTPException(404, "Gentest nicht gefunden.")
if user["rolle"] != "admin" and test["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
conn.execute("DELETE FROM dog_genetic_tests WHERE id=?", (tid,))
return None
# ------------------------------------------------------------------
# PUT /api/zuchthunde/titles/{tid}
# ------------------------------------------------------------------
@router.put("/zuchthunde/titles/{tid}")
async def update_titel(tid: int, body: TitelUpdate, user=Depends(_require_breeder)):
with db() as conn:
titel = conn.execute(
"""SELECT dt.*, bp.user_id AS owner_user_id
FROM dog_titles dt
JOIN zucht_hunde zh ON zh.id = dt.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE dt.id=?""",
(tid,)
).fetchone()
if not titel:
raise HTTPException(404, "Titel nicht gefunden.")
if user["rolle"] != "admin" and titel["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
fields, params = [], []
for field, value in body.model_dump(exclude_none=True).items():
fields.append(f"{field}=?")
params.append(value)
if not fields:
raise HTTPException(400, "Keine Felder zum Aktualisieren.")
params.append(tid)
conn.execute(
f"UPDATE dog_titles SET {', '.join(fields)} WHERE id=?", params
)
row = conn.execute(
"SELECT * FROM dog_titles WHERE id=?", (tid,)
).fetchone()
return dict(row)
# ------------------------------------------------------------------
# DELETE /api/zuchthunde/titles/{tid}
# ------------------------------------------------------------------
@router.delete("/zuchthunde/titles/{tid}", status_code=204)
async def delete_titel(tid: int, user=Depends(_require_breeder)):
with db() as conn:
titel = conn.execute(
"""SELECT dt.id, bp.user_id AS owner_user_id
FROM dog_titles dt
JOIN zucht_hunde zh ON zh.id = dt.hund_id
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE dt.id=?""",
(tid,)
).fetchone()
if not titel:
raise HTTPException(404, "Titel nicht gefunden.")
if user["rolle"] != "admin" and titel["owner_user_id"] != user["id"]:
raise HTTPException(403, "Kein Zugriff.")
conn.execute("DELETE FROM dog_titles WHERE id=?", (tid,))
return None
# ==================================================================
# {id}-ROUTEN — nach den fixen Pfaden!
# ==================================================================
# ------------------------------------------------------------------
# GET /api/zuchthunde/{id} — Hund-Detail
# ------------------------------------------------------------------
@router.get("/zuchthunde/{hund_id}")
async def get_hund(hund_id: int, user=Depends(get_current_user_optional)):
with db() as conn:
hund = _check_hund_access(hund_id, user, conn)
return hund
# ------------------------------------------------------------------
# PUT /api/zuchthunde/{id} — bearbeiten
# ------------------------------------------------------------------
@router.put("/zuchthunde/{hund_id}")
async def update_hund(hund_id: int, body: HundUpdate, user=Depends(_require_breeder)):
with db() as conn:
_check_hund_owner(hund_id, user, conn)
fields, params = [], []
for field, value in body.model_dump(exclude_none=True).items():
fields.append(f"{field}=?")
params.append(value)
if not fields:
raise HTTPException(400, "Keine Felder zum Aktualisieren.")
params.append(hund_id)
conn.execute(
f"UPDATE zucht_hunde SET {', '.join(fields)} WHERE id=?", params
)
row = conn.execute(
"SELECT * FROM zucht_hunde WHERE id=?", (hund_id,)
).fetchone()
return dict(row)
# ------------------------------------------------------------------
# DELETE /api/zuchthunde/{id} — löschen (cascade)
# ------------------------------------------------------------------
@router.delete("/zuchthunde/{hund_id}", status_code=204)
async def delete_hund(hund_id: int, user=Depends(_require_breeder)):
with db() as conn:
_check_hund_owner(hund_id, user, conn)
conn.execute("DELETE FROM dog_health_tests WHERE hund_id=?", (hund_id,))
conn.execute("DELETE FROM dog_genetic_tests WHERE hund_id=?", (hund_id,))
conn.execute("DELETE FROM dog_titles WHERE hund_id=?", (hund_id,))
conn.execute("DELETE FROM zucht_hunde WHERE id=?", (hund_id,))
return None
# ------------------------------------------------------------------
# GET /api/zuchthunde/{id}/pedigree — Stammbaum
# ------------------------------------------------------------------
@router.get("/zuchthunde/{hund_id}/pedigree")
async def get_pedigree(
hund_id: int,
generations: int = Query(default=4, ge=1, le=8),
user=Depends(get_current_user_optional),
):
with db() as conn:
_check_hund_access(hund_id, user, conn)
tree = _build_tree(conn, hund_id, generations)
if not tree:
raise HTTPException(404, "Hund nicht gefunden.")
return tree
# ==================================================================
# GESUNDHEITSTESTS
# ==================================================================
# ------------------------------------------------------------------
# GET /api/zuchthunde/{id}/health-tests
# ------------------------------------------------------------------
@router.get("/zuchthunde/{hund_id}/health-tests")
async def list_health_tests(hund_id: int, user=Depends(get_current_user_optional)):
with db() as conn:
_check_hund_access(hund_id, user, conn)
is_owner = user and (
user["rolle"] == "admin"
or conn.execute(
"""SELECT 1 FROM zucht_hunde zh
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE zh.id=? AND bp.user_id=?""",
(hund_id, user["id"])
).fetchone() is not None
)
q = "SELECT * FROM dog_health_tests WHERE hund_id=?"
params = [hund_id]
if not is_owner:
q += " AND is_public=1"
rows = conn.execute(q + " ORDER BY untersuch_am DESC", params).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------
# POST /api/zuchthunde/{id}/health-tests
# ------------------------------------------------------------------
@router.post("/zuchthunde/{hund_id}/health-tests", status_code=201)
async def create_health_test(
hund_id: int, body: HealthTestCreate, user=Depends(_require_breeder)
):
with db() as conn:
_check_hund_owner(hund_id, user, conn)
cur = conn.execute(
"""INSERT INTO dog_health_tests
(hund_id, test_typ, test_name, ergebnis, untersuch_am, gueltig_bis,
untersucher, labor, zertifikat_nr, is_public)
VALUES (?,?,?,?,?,?,?,?,?,?)""",
(
hund_id, body.test_typ, body.test_name, body.ergebnis,
body.untersuch_am, body.gueltig_bis, body.untersucher,
body.labor, body.zertifikat_nr, body.is_public,
)
)
row = conn.execute(
"SELECT * FROM dog_health_tests WHERE id=?", (cur.lastrowid,)
).fetchone()
return dict(row)
# ==================================================================
# GENTESTS
# ==================================================================
# ------------------------------------------------------------------
# GET /api/zuchthunde/{id}/genetic-tests
# ------------------------------------------------------------------
@router.get("/zuchthunde/{hund_id}/genetic-tests")
async def list_genetic_tests(hund_id: int, user=Depends(get_current_user_optional)):
with db() as conn:
_check_hund_access(hund_id, user, conn)
is_owner = user and (
user["rolle"] == "admin"
or conn.execute(
"""SELECT 1 FROM zucht_hunde zh
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE zh.id=? AND bp.user_id=?""",
(hund_id, user["id"])
).fetchone() is not None
)
q = "SELECT * FROM dog_genetic_tests WHERE hund_id=?"
params = [hund_id]
if not is_owner:
q += " AND is_public=1"
rows = conn.execute(q + " ORDER BY getestet_am DESC", params).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------
# POST /api/zuchthunde/{id}/genetic-tests
# ------------------------------------------------------------------
@router.post("/zuchthunde/{hund_id}/genetic-tests", status_code=201)
async def create_genetic_test(
hund_id: int, body: GeneticTestCreate, user=Depends(_require_breeder)
):
with db() as conn:
_check_hund_owner(hund_id, user, conn)
cur = conn.execute(
"""INSERT INTO dog_genetic_tests
(hund_id, marker_name, marker_kategorie, genotyp, ergebnis_klasse,
getestet_am, labor, zertifikat_nr, is_public)
VALUES (?,?,?,?,?,?,?,?,?)""",
(
hund_id, body.marker_name, body.marker_kategorie, body.genotyp,
body.ergebnis_klasse, body.getestet_am, body.labor,
body.zertifikat_nr, body.is_public,
)
)
row = conn.execute(
"SELECT * FROM dog_genetic_tests WHERE id=?", (cur.lastrowid,)
).fetchone()
return dict(row)
# ==================================================================
# TITEL
# ==================================================================
# ------------------------------------------------------------------
# GET /api/zuchthunde/{id}/titles
# ------------------------------------------------------------------
@router.get("/zuchthunde/{hund_id}/titles")
async def list_titles(hund_id: int, user=Depends(get_current_user_optional)):
with db() as conn:
_check_hund_access(hund_id, user, conn)
is_owner = user and (
user["rolle"] == "admin"
or conn.execute(
"""SELECT 1 FROM zucht_hunde zh
LEFT JOIN breeder_profiles bp ON bp.id = zh.breeder_id
WHERE zh.id=? AND bp.user_id=?""",
(hund_id, user["id"])
).fetchone() is not None
)
q = "SELECT * FROM dog_titles WHERE hund_id=?"
params = [hund_id]
if not is_owner:
q += " AND is_public=1"
rows = conn.execute(q + " ORDER BY verliehen_am DESC", params).fetchall()
return [dict(r) for r in rows]
# ------------------------------------------------------------------
# POST /api/zuchthunde/{id}/titles
# ------------------------------------------------------------------
@router.post("/zuchthunde/{hund_id}/titles", status_code=201)
async def create_titel(
hund_id: int, body: TitelCreate, user=Depends(_require_breeder)
):
with db() as conn:
_check_hund_owner(hund_id, user, conn)
cur = conn.execute(
"""INSERT INTO dog_titles
(hund_id, titel_typ, titel_name, verliehen_am, ort,
richter, ausstellung, formwert, is_public)
VALUES (?,?,?,?,?,?,?,?,?)""",
(
hund_id, body.titel_typ, body.titel_name, body.verliehen_am,
body.ort, body.richter, body.ausstellung, body.formwert,
body.is_public,
)
)
row = conn.execute(
"SELECT * FROM dog_titles WHERE id=?", (cur.lastrowid,)
).fetchone()
return dict(row)
Reference in a new issue View git blame Copy permalink