"""BAN YARO — Zuchtkartei (Hunde, Gesundheitstests, Gentests, Titel, Stammbaum, IK)""" import logging from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field 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 = Field(..., min_length=1, max_length=200) rufname: Optional[str] = Field(None, max_length=80) geschlecht: str = Field(..., max_length=20) # maennlich|weiblich geburtsdatum: Optional[str] = Field(None, max_length=32) sterbedatum: Optional[str] = Field(None, max_length=32) chip_nr: Optional[str] = Field(None, max_length=50) taetowiernummer: Optional[str] = Field(None, max_length=50) zuchtbuchnummer: Optional[str] = Field(None, max_length=100) farbe: Optional[str] = Field(None, max_length=100) vater_id: Optional[int] = None mutter_id: Optional[int] = None zuechter_name: Optional[str] = Field(None, max_length=200) eigentuemer_name: Optional[str] = Field(None, max_length=200) is_public: int = 1 notiz: Optional[str] = Field(None, max_length=5000) foto_url: Optional[str] = Field(None, max_length=500) class HundUpdate(BaseModel): name: Optional[str] = Field(None, max_length=200) rufname: Optional[str] = Field(None, max_length=80) geschlecht: Optional[str] = Field(None, max_length=20) geburtsdatum: Optional[str] = Field(None, max_length=32) sterbedatum: Optional[str] = Field(None, max_length=32) chip_nr: Optional[str] = Field(None, max_length=50) taetowiernummer: Optional[str] = Field(None, max_length=50) zuchtbuchnummer: Optional[str] = Field(None, max_length=100) farbe: Optional[str] = Field(None, max_length=100) vater_id: Optional[int] = None mutter_id: Optional[int] = None zuechter_name: Optional[str] = Field(None, max_length=200) eigentuemer_name: Optional[str] = Field(None, max_length=200) is_public: Optional[int] = None notiz: Optional[str] = Field(None, max_length=5000) foto_url: Optional[str] = Field(None, max_length=500) class HealthTestCreate(BaseModel): test_typ: str = Field(..., max_length=50) # HD|ED|OCD|augen|herz|patella|ZTP|custom test_name: Optional[str] = Field(None, max_length=200) ergebnis: Optional[str] = Field(None, max_length=500) untersuch_am: Optional[str] = Field(None, max_length=32) gueltig_bis: Optional[str] = Field(None, max_length=32) untersucher: Optional[str] = Field(None, max_length=200) labor: Optional[str] = Field(None, max_length=200) zertifikat_nr: Optional[str] = Field(None, max_length=100) is_public: int = 1 class HealthTestUpdate(BaseModel): test_typ: Optional[str] = Field(None, max_length=50) test_name: Optional[str] = Field(None, max_length=200) ergebnis: Optional[str] = Field(None, max_length=500) untersuch_am: Optional[str] = Field(None, max_length=32) gueltig_bis: Optional[str] = Field(None, max_length=32) untersucher: Optional[str] = Field(None, max_length=200) labor: Optional[str] = Field(None, max_length=200) zertifikat_nr: Optional[str] = Field(None, max_length=100) is_public: Optional[int] = None class GeneticTestCreate(BaseModel): marker_name: str = Field(..., max_length=100) # MDR1|PRA-prcd|DM|vWD|HUU etc. marker_kategorie: Optional[str] = Field(None, max_length=50) # krankheit|farbe|eigenschaft genotyp: Optional[str] = Field(None, max_length=20) # +/+|+/-|-/- ergebnis_klasse: Optional[str] = Field(None, max_length=50) # clear|carrier|affected getestet_am: Optional[str] = Field(None, max_length=32) labor: Optional[str] = Field(None, max_length=200) zertifikat_nr: Optional[str] = Field(None, max_length=100) is_public: int = 1 class GeneticTestUpdate(BaseModel): marker_name: Optional[str] = Field(None, max_length=100) marker_kategorie: Optional[str] = Field(None, max_length=50) genotyp: Optional[str] = Field(None, max_length=20) ergebnis_klasse: Optional[str] = Field(None, max_length=50) getestet_am: Optional[str] = Field(None, max_length=32) labor: Optional[str] = Field(None, max_length=200) zertifikat_nr: Optional[str] = Field(None, max_length=100) is_public: Optional[int] = None class TitelCreate(BaseModel): titel_typ: str = Field(..., max_length=50) # ausstellung|arbeit|sport|zucht|champion|custom titel_name: str = Field(..., min_length=1, max_length=200) verliehen_am: Optional[str] = Field(None, max_length=32) ort: Optional[str] = Field(None, max_length=200) richter: Optional[str] = Field(None, max_length=200) ausstellung: Optional[str] = Field(None, max_length=200) formwert: Optional[str] = Field(None, max_length=100) is_public: int = 1 class TitelUpdate(BaseModel): titel_typ: Optional[str] = Field(None, max_length=50) titel_name: Optional[str] = Field(None, max_length=200) verliehen_am: Optional[str] = Field(None, max_length=32) ort: Optional[str] = Field(None, max_length=200) richter: Optional[str] = Field(None, max_length=200) ausstellung: Optional[str] = Field(None, max_length=200) formwert: Optional[str] = Field(None, max_length=100) 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"]) # Genetische Risiken für Welfare-Check genetic_risks = [] try: vater_gen = conn.execute( "SELECT marker_name, ergebnis_klasse FROM dog_genetic_tests WHERE hund_id=?", (body.vater_id,) ).fetchall() mutter_gen = conn.execute( "SELECT marker_name, ergebnis_klasse FROM dog_genetic_tests WHERE hund_id=?", (body.mutter_id,) ).fetchall() mutter_map = {r["marker_name"]: r["ergebnis_klasse"] for r in mutter_gen} RISIKO = { ("carrier","carrier"): "25% betroffen, 50% Träger", ("carrier","affected"): "50% betroffen, 50% Träger", ("affected","carrier"): "50% betroffen, 50% Träger", ("affected","affected"): "100% betroffen", ("clear","affected"): "0% betroffen, 100% Träger", ("affected","clear"): "0% betroffen, 100% Träger", } for vg in vater_gen: ms = mutter_map.get(vg["marker_name"]) if ms: risk = RISIKO.get((vg["ergebnis_klasse"], ms)) if risk: genetic_risks.append({"marker": vg["marker_name"], "offspring_risk": risk}) except Exception: pass # Züchter-Profil für Welfare profile = conn.execute( "SELECT id FROM breeder_profiles WHERE user_id=?", (user["id"],) ).fetchone() bid = profile["id"] if profile else None from welfare_check import check_welfare welfare = check_welfare( conn, bid or 0, vater_id=body.vater_id, mutter_id=body.mutter_id, ik_prozent=ik_prozent, genetic_risks=genetic_risks, ) return { "ik_prozent": ik_prozent, "ik_rating": rating, "gemeinsame_vorfahren": gemeinsame_vorfahren, "welfare": welfare, } # ------------------------------------------------------------------ # 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)