Add additional arguments to decorator

This allows for `@requires_team_membership()` to be able to
retrieve the team_id from a JSON or query parameter.
master
John Montagu, the 4th Earl of Sandvich 2024-11-21 18:18:43 -08:00
parent 3ff1ef7683
commit 061499b822
Signed by: sandvich
GPG Key ID: 9A39BE37E602B22D
4 changed files with 46 additions and 26 deletions

View File

@ -72,7 +72,7 @@ class CreateEventJson(BaseModel):
) )
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def create_event(player_team: PlayerTeam, json: CreateEventJson, **_): def create_event(player_team: PlayerTeam, json: CreateEventJson, **_):
event = Event() event = Event()
event.team_id = player_team.team_id event.team_id = player_team.team_id
@ -107,7 +107,7 @@ def create_event(player_team: PlayerTeam, json: CreateEventJson, **_):
@api_events.patch("/<int:event_id>/players") @api_events.patch("/<int:event_id>/players")
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def set_event_players(player_team: PlayerTeam, event_id: int, **_): def set_event_players(player_team: PlayerTeam, event_id: int, **_):
assert_team_authority(player_team, None) assert_team_authority(player_team, None)

View File

@ -1,5 +1,7 @@
from functools import wraps from functools import wraps
from typing import Optional
from flask import abort, make_response, request from flask import abort, make_response, request
from sqlalchemy.sql.operators import json_path_getitem_op
from app_db import db from app_db import db
from models.auth_session import AuthSession from models.auth_session import AuthSession
from models.player import Player from models.player import Player
@ -28,29 +30,47 @@ def requires_authentication(f):
return f(*args, **kwargs) return f(*args, **kwargs)
return decorator return decorator
def requires_team_membership(f): def requires_team_membership(
@wraps(f) path_param: Optional[str] = None,
def decorator(*args, **kwargs): json_param: Optional[str] = None,
player: Player | None = kwargs["player"] query_param: Optional[str] = None
team_id: int = kwargs["team_id"] ):
def wrapper(f):
@wraps(f)
def decorator(*args, **kwargs):
player: Player | None = kwargs["player"]
if not player: team_id: int
abort(401) if path_param:
team_id = kwargs[path_param]
elif json_param:
team_id = getattr(kwargs["json"], json_param)
elif query_param:
team_id = getattr(kwargs["query"], query_param)
else:
team_id = kwargs["team_id"]
player_team = db.session.query( if not player:
PlayerTeam abort(401)
).where(
PlayerTeam.player == player
).where(
PlayerTeam.team_id == team_id
).one_or_none()
if not player_team: if not team_id:
abort(404, "Player is not a member of this team") abort(500)
kwargs["player_team"] = player_team player_team = db.session.query(
return f(*args, **kwargs) PlayerTeam
return decorator ).where(
PlayerTeam.player == player
).where(
PlayerTeam.team_id == team_id
).one_or_none()
if not player_team:
abort(404, "Player is not a member of this team")
kwargs["player_team"] = player_team
return f(*args, **kwargs)
return decorator
return wrapper
def assert_team_authority( def assert_team_authority(
player_team: PlayerTeam, player_team: PlayerTeam,

View File

@ -55,7 +55,7 @@ def get_integrations(player: Player, team_id: int, **_):
operation_id="create_integration" operation_id="create_integration"
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def create_integration(player_team: PlayerTeam, integration_type: str, **_): def create_integration(player_team: PlayerTeam, integration_type: str, **_):
assert_team_authority(player_team) assert_team_authority(player_team)
@ -81,7 +81,7 @@ def create_integration(player_team: PlayerTeam, integration_type: str, **_):
operation_id="delete_integration" operation_id="delete_integration"
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def delete_integration(player_team: PlayerTeam, integration_id: int, **_): def delete_integration(player_team: PlayerTeam, integration_id: int, **_):
assert_team_authority(player_team) assert_team_authority(player_team)
@ -109,7 +109,7 @@ def delete_integration(player_team: PlayerTeam, integration_id: int, **_):
operation_id="update_integration" operation_id="update_integration"
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def update_integration( def update_integration(
player_team: PlayerTeam, player_team: PlayerTeam,
integration_id: int, integration_id: int,

View File

@ -22,7 +22,7 @@ api_team_invite = Blueprint("team_invite", __name__)
operation_id="get_invites" operation_id="get_invites"
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def get_invites(team_id: int, **_): def get_invites(team_id: int, **_):
invites = db.session.query( invites = db.session.query(
TeamInvite TeamInvite
@ -48,7 +48,7 @@ def get_invites(team_id: int, **_):
operation_id="create_invite" operation_id="create_invite"
) )
@requires_authentication @requires_authentication
@requires_team_membership @requires_team_membership()
def create_invite(team_id: int, **_): def create_invite(team_id: int, **_):
team_id_shifted = int(team_id) << 48 team_id_shifted = int(team_id) << 48
random_value_shifted = int(randint(0, (1 << 16) - 1)) << 32 random_value_shifted = int(randint(0, (1 << 16) - 1)) << 32