Lars
b8453f3f07
All checks were successful
Deploy Development / deploy (push) Successful in 33s
Test Suite / pytest-backend (push) Successful in 23s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 6s
Test Suite / playwright-tests (push) Successful in 31s
- Introduced new API endpoints for bulk lifecycle actions and bulk patching of media assets, allowing for more efficient management of multiple assets. - Updated media lifecycle permissions to restrict actions based on user roles, ensuring that only superadmins can perform critical operations like purging and force lifecycle changes. - Enhanced frontend components to support new bulk actions and improved user experience in the media library, including visibility and copyright management. - Incremented version to 0.8.49, reflecting the latest improvements in media handling and governance.
223 lines
7.4 KiB
Python
223 lines
7.4 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 is_superadmin(role: Optional[str]) -> bool:
|
|
return (role or "").lower() == "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 club_admin_shares_club_with_creator(
|
|
cur, club_admin_profile_id: int, creator_profile_id: int
|
|
) -> bool:
|
|
"""
|
|
True, wenn club_admin_profile_id in mindestens einem Verein die Rolle club_admin hat und
|
|
creator_profile_id dort ebenfalls aktives Mitglied ist (z. B. Löschen fremder privater Übungen).
|
|
"""
|
|
if club_admin_profile_id == creator_profile_id:
|
|
return False
|
|
cur.execute(
|
|
"""
|
|
SELECT 1
|
|
FROM club_members cm_admin
|
|
INNER JOIN club_member_roles r
|
|
ON r.club_member_id = cm_admin.id AND r.role_code = 'club_admin'
|
|
INNER JOIN club_members cm_creator
|
|
ON cm_creator.club_id = cm_admin.club_id
|
|
AND cm_creator.profile_id = %s
|
|
AND cm_creator.status = 'active'
|
|
WHERE cm_admin.profile_id = %s AND cm_admin.status = 'active'
|
|
LIMIT 1
|
|
""",
|
|
(creator_profile_id, club_admin_profile_id),
|
|
)
|
|
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, active_only: bool = True) -> List[Dict[str, Any]]:
|
|
status_filter = "AND cm.status = 'active'" if active_only else ""
|
|
cur.execute(
|
|
f"""
|
|
SELECT c.id, c.name, c.abbreviation, c.status,
|
|
cm.status AS membership_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 {status_filter}
|
|
GROUP BY c.id, c.name, c.abbreviation, c.status, cm.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 club_ids_for_profile_with_roles(cur, profile_id: int, *role_codes: str) -> Set[int]:
|
|
"""Vereins-IDs, in denen das Profil mindestens eine der Rollen hat."""
|
|
if not role_codes:
|
|
return set()
|
|
ph = ",".join(["%s"] * len(role_codes))
|
|
cur.execute(
|
|
f"""
|
|
SELECT DISTINCT cm.club_id
|
|
FROM club_members cm
|
|
INNER JOIN club_member_roles r ON r.club_member_id = cm.id
|
|
WHERE cm.profile_id = %s AND cm.status = 'active'
|
|
AND r.role_code IN ({ph})
|
|
""",
|
|
(profile_id, *role_codes),
|
|
)
|
|
return {int(r["club_id"]) for r in cur.fetchall() if r.get("club_id") is not None}
|
|
|
|
|
|
_GOVERNANCE_VISIBILITY = frozenset({"private", "club", "official"})
|
|
|
|
|
|
def assert_valid_governance_visibility(
|
|
cur,
|
|
profile_id: int,
|
|
role: Optional[str],
|
|
visibility: str,
|
|
club_id: Optional[int],
|
|
) -> None:
|
|
"""Pflicht club_id bei visibility=club; Mitgliedschaft außer Plattform-Admin; official nur Plattform-Admin."""
|
|
if visibility not in _GOVERNANCE_VISIBILITY:
|
|
raise HTTPException(status_code=400, detail="Ungültige visibility")
|
|
if visibility == "official" and not is_platform_admin(role):
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Nur Plattform-Admins dürfen offizielle Inhalte setzen",
|
|
)
|
|
if visibility == "club":
|
|
if club_id is None:
|
|
raise HTTPException(status_code=400, detail="club_id ist bei visibility=club erforderlich")
|
|
if is_platform_admin(role):
|
|
cur.execute("SELECT 1 FROM clubs WHERE id = %s", (club_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(status_code=400, detail="Verein nicht gefunden")
|
|
else:
|
|
assert_club_member(cur, profile_id, club_id)
|
|
|
|
|
|
def library_content_visible_to_profile(
|
|
cur,
|
|
profile_id: int,
|
|
visibility: str,
|
|
content_club_id: Optional[int],
|
|
created_by: Optional[int],
|
|
global_role: Optional[str],
|
|
) -> bool:
|
|
"""Leserechte wie Übungen für alle Objekte mit visibility/club_id/created_by (Bibliothek & Co.)."""
|
|
return exercise_visible_to_profile(
|
|
cur, profile_id, visibility, content_club_id, created_by, global_role
|
|
)
|
|
|
|
|
|
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:
|
|
"""Leserechte einer Übung. Für neue Codepfade lieber `library_content_visible_to_profile` verwenden."""
|
|
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
|