128 lines
3.9 KiB
Python
128 lines
3.9 KiB
Python
"""
|
|
Vereins-Mandanten: Mitgliedschaften, aktiver Vereinskontext, einfache Berechtigungen.
|
|
|
|
Siehe .claude/docs/technical/MULTI_TENANCY_RBAC_ARCHITECTURE.md
|
|
"""
|
|
from typing import Any, Dict, List, Optional, Set
|
|
|
|
from fastapi import HTTPException
|
|
|
|
def is_platform_admin(role: Optional[str]) -> bool:
|
|
return (role or "").lower() in ("admin", "superadmin")
|
|
|
|
|
|
def club_ids_for_profile(cur, profile_id: int) -> Set[int]:
|
|
cur.execute(
|
|
"""
|
|
SELECT club_id FROM club_members
|
|
WHERE profile_id = %s AND status = 'active'
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
return {r["club_id"] for r in cur.fetchall()}
|
|
|
|
|
|
def assert_club_member(cur, profile_id: int, club_id: int) -> None:
|
|
cur.execute(
|
|
"""
|
|
SELECT 1 FROM club_members
|
|
WHERE profile_id = %s AND club_id = %s AND status = 'active'
|
|
LIMIT 1
|
|
""",
|
|
(profile_id, club_id),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(status_code=403, detail="Keine Mitgliedschaft in diesem Verein")
|
|
|
|
|
|
def has_club_role(cur, profile_id: int, club_id: int, *role_codes: str) -> bool:
|
|
if not role_codes:
|
|
return False
|
|
cur.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM club_members cm
|
|
INNER JOIN club_member_roles r ON r.club_member_id = cm.id
|
|
WHERE cm.profile_id = %s AND cm.club_id = %s AND cm.status = 'active'
|
|
AND r.role_code IN ({ph})
|
|
LIMIT 1
|
|
""".format(ph=",".join(["%s"] * len(role_codes))),
|
|
(profile_id, club_id, *role_codes),
|
|
)
|
|
return cur.fetchone() is not None
|
|
|
|
|
|
def can_manage_club_org(cur, profile_id: int, club_id: int, global_role: Optional[str]) -> bool:
|
|
"""Sparten/Gruppen/Struktur: Vereinsadmin oder Plattform-Admin."""
|
|
if is_platform_admin(global_role):
|
|
return True
|
|
return has_club_role(cur, profile_id, club_id, "club_admin")
|
|
|
|
|
|
def can_plan_in_club(cur, profile_id: int, club_id: int, global_role: Optional[str]) -> bool:
|
|
"""Trainingsgruppen anlegen / planen: Admin-Rollen im Verein oder Plattform."""
|
|
if is_platform_admin(global_role):
|
|
return True
|
|
return has_club_role(
|
|
cur, profile_id, club_id, "club_admin", "trainer", "content_editor", "division_lead"
|
|
)
|
|
|
|
|
|
def memberships_with_roles(cur, profile_id: int) -> List[Dict[str, Any]]:
|
|
cur.execute(
|
|
"""
|
|
SELECT c.id, c.name, c.abbreviation, c.status,
|
|
COALESCE(
|
|
ARRAY_AGG(DISTINCT r.role_code) FILTER (WHERE r.role_code IS NOT NULL),
|
|
ARRAY[]::varchar[]
|
|
) AS roles
|
|
FROM club_members cm
|
|
INNER JOIN clubs c ON c.id = cm.club_id
|
|
LEFT JOIN club_member_roles r ON r.club_member_id = cm.id
|
|
WHERE cm.profile_id = %s AND cm.status = 'active'
|
|
GROUP BY c.id, c.name, c.abbreviation, c.status
|
|
ORDER BY c.name
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
out: List[Dict[str, Any]] = []
|
|
for row in cur.fetchall():
|
|
d = dict(row)
|
|
roles = d.get("roles") or []
|
|
if hasattr(roles, "tolist"):
|
|
roles = roles.tolist()
|
|
d["roles"] = list(roles)
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
def exercise_visible_to_profile(
|
|
cur,
|
|
profile_id: int,
|
|
visibility: str,
|
|
exercise_club_id: Optional[int],
|
|
created_by: Optional[int],
|
|
global_role: Optional[str],
|
|
) -> bool:
|
|
if is_platform_admin(global_role):
|
|
return True
|
|
if visibility == "official":
|
|
return True
|
|
if created_by is not None and created_by == profile_id:
|
|
return True
|
|
if visibility == "private":
|
|
return False
|
|
if visibility == "club":
|
|
if exercise_club_id is None:
|
|
return False
|
|
cur.execute(
|
|
"""
|
|
SELECT 1 FROM club_members
|
|
WHERE profile_id = %s AND club_id = %s AND status = 'active'
|
|
LIMIT 1
|
|
""",
|
|
(profile_id, exercise_club_id),
|
|
)
|
|
return cur.fetchone() is not None
|
|
return False
|