rene
1ff66a7083
PYDANTIC max_length (38 Routen, ~400 Field-Constraints): Schützt vor DoS durch Riesen-Payloads (10MB Thread-Titel etc.). Pragmatische Limits: - Titel/Name: 200 · Beschreibung/Body: 10000 · Notiz: 5000 - Email: 254 (RFC 5321) · URL: 500 · Slug/Kategorie: 100 - Hund-Name/Rasse: 80 · Hund-Bio: 2000 Top-betroffen: forum.py, diary.py, health.py, dogs.py, expenses.py, notes.py, auth.py, profile.py. Manuelle len()-Checks in profile, chat, ki entfernt (jetzt durch Field abgedeckt). PYTEST COVERAGE (+19 Tests, 37 grün + 1 xfail): - test_security.py: require_owner (Places GET/PATCH/DELETE mit Fremduser → 403), JWT-Blacklist (Logout invalidiert Token), Login-Lockout (5 Fehlversuche → 429 + Retry-After Header) - test_race.py: Invoice-Counter (20 parallele Threads, alle unique), Founder-Number (atomare Vergabe, voll bei 100) - test_validation.py: Forum-Titel 30k Zeichen → 422, Diary-Text 50k → 422 (verifiziert Pydantic max_length-Sweep) A11Y (Tap-Targets ≥44×44 + Dark-Mode-Kontrast): - #header-user-btn 36→44px, .header-back 40→44, .header-menu-btn 40→44 - dog-profile Wrapped-Slider Prev/Next 40→44 - forum-Lightbox Close 40→44 - --c-text-muted Light: #B0A090 (2.37:1 FAIL) → #7F6B58 (4.74:1 PASS) - --c-text-muted Dark: #806A58 (3.58:1 FAIL) → #A08878 (5.46:1 PASS) - Branding-Farben unangetastet
822 lines
33 KiB
Python
822 lines
33 KiB
Python
"""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)
|