sandvich
/
availabili.tf
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

Migrate to spectree for pydantic integration

master
John Montagu, the 4th Earl of Sandvich 2024-11-03 14:29:59 -08:00
parent b76d2786e5
commit 283624706e
Signed by: sandvich
GPG Key ID: 9A39BE37E602B22D
6 changed files with 98 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend-flask/app.py
View File

@ -1,11 +1,11 @@
from flask import Blueprint, Flask, make_response, request
from flask_sqlalchemy import SQLAlchemy
from flask_cors import CORS
import login
import schedule
import team
from models import init_db
from spec import spec
app = Flask(__name__)
CORS(app, origins=["http://localhost:5173"], supports_credentials=True)
@ -28,3 +28,4 @@ def debug_set_cookie():
return res, 200
app.register_blueprint(api)
spec.register(app)

38
backend-flask/login.py
View File

@ -15,43 +15,6 @@ STEAM_OPENID_URL = "https://steamcommunity.com/openid/login"
def index():
return "test"
def get_steam_login_url(return_to):
"""Build the Steam OpenID URL for login"""
params = {
"openid.ns": "http://specs.openid.net/auth/2.0",
"openid.mode": "checkid_setup",
"openid.return_to": return_to,
"openid.identity": "http://specs.openid.net/auth/2.0/identifier_select",
"openid.claimed_id": "http://specs.openid.net/auth/2.0/identifier_select",
}
return f"{STEAM_OPENID_URL}?{urllib.parse.urlencode(params)}"
#@api_login.get("/steam/")
#def steam_login():
# return_to = url_for("api.login.steam_login_callback", _external=True)
# steam_login_url = get_steam_login_url(return_to)
# return redirect(steam_login_url)
#
#@api_login.get("/steam/callback/")
#def steam_login_callback():
# params = request.args.to_dict()
# params["openid.mode"] = "check_authentication"
# response = requests.post(STEAM_OPENID_URL, data=params)
#
# # Check if authentication was successful
# if "is_valid:true" in response.text:
# claimed_id = request.args.get("openid.claimed_id")
# steam_id = extract_steam_id_from_response(claimed_id)
# print("User logged in as", steam_id)
#
# player = create_or_get_user_from_steam_id(int(steam_id))
# auth_session = create_auth_session_for_player(player)
#
# resp = make_response("Logged in")
# resp.set_cookie("auth", auth_session.key, secure=True, httponly=True)
# return resp
# return "no"
@api_login.post("/authenticate")
def steam_authenticate():
params = request.get_json()
@ -64,7 +27,6 @@ def steam_authenticate():
steam_id = int(extract_steam_id_from_response(claimed_id))
print("User logged in as", steam_id)
#player = create_or_get_user_from_steam_id(int(steam_id))
player = db.session.query(
Player
).where(

13
backend-flask/models.py
View File

@ -8,6 +8,8 @@ from sqlalchemy import TIMESTAMP, BigInteger, Boolean, Enum, ForeignKey, Foreign
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy_utc import UtcDateTime
import spec
class Base(DeclarativeBase):
pass
@ -34,6 +36,11 @@ class Player(db.Model):
created_at: Mapped[datetime] = mapped_column(TIMESTAMP, server_default=func.now())
class PlayerSpec(spec.BaseModel):
steam_id: str
username: str
#teams: list["PlayerTeamSpec"]
class Team(db.Model):
__tablename__ = "teams"
@ -45,6 +52,12 @@ class Team(db.Model):
created_at: Mapped[datetime] = mapped_column(TIMESTAMP, server_default=func.now())
class TeamSpec(spec.BaseModel):
id: int
team_name: str
discord_webhook_url: str | None
#players: list[PlayerTeamSpec] | None
class PlayerTeam(db.Model):
__tablename__ = "players_teams"

7
backend-flask/requirements.txt
View File

@ -1,14 +1,19 @@
flask
# CORS
Flask-CORS
# ORM
sqlalchemy
Flask-SQLAlchemy
SQLAlchemy-Utc
# form/data validation
pydantic
Flask-Pydantic
spectree # generates OpenAPI documents for us to make TypeScript API clients
# based on our pydantic models
# DB migrations
alembic
Flask-Migrate

55
backend-flask/schedule.py
View File

@ -1,28 +1,25 @@
import datetime
from typing import cast
from flask import Blueprint, abort, jsonify, make_response, request
import pydantic
from flask_pydantic import validate
from models import Player, PlayerTeam, PlayerTeamAvailability, PlayerTeamRole, db
from middleware import requires_authentication
import models
import utc
from spec import spec, BaseModel
api_schedule = Blueprint("schedule", __name__, url_prefix="/schedule")
class ViewScheduleForm(pydantic.BaseModel):
class ViewScheduleForm(BaseModel):
window_start: datetime.datetime
team_id: int
window_size_days: int = 7
@api_schedule.get("/")
@validate(query=ViewScheduleForm)
@spec.validate()
@requires_authentication
def get(query: ViewScheduleForm, *args, **kwargs):
def get(query: ViewScheduleForm, player: Player, **kwargs):
window_start = query.window_start
window_end = window_start + datetime.timedelta(days=query.window_size_days)
player: Player = kwargs["player"]
availability_regions = db.session.query(
PlayerTeamAvailability
@ -65,7 +62,7 @@ def get(query: ViewScheduleForm, *args, **kwargs):
"availability": availability
}
class PutScheduleForm(pydantic.BaseModel):
class PutScheduleForm(BaseModel):
window_start: datetime.datetime
window_size_days: int = 7
team_id: int
@ -91,17 +88,14 @@ def find_consecutive_blocks(arr: list[int]) -> list[tuple[int, int, int]]:
return blocks
@api_schedule.put("/")
@validate(body=PutScheduleForm, get_json_params={})
@spec.validate()
@requires_authentication
def put(body: PutScheduleForm, **kwargs):
window_start = body.window_start.replace(tzinfo=utc.utc)
window_end = window_start + datetime.timedelta(days=body.window_size_days)
player: Player = kwargs["player"]
if not player:
abort(400)
def put(json: PutScheduleForm, player: Player, **kwargs):
window_start = json.window_start
window_end = window_start + datetime.timedelta(days=json.window_size_days)
# TODO: add error message
if len(body.availability) != 168:
if len(json.availability) != 168:
abort(400, {
"error": "Availability must be length " + str(168)
})
@ -111,7 +105,7 @@ def put(body: PutScheduleForm, **kwargs):
).where(
PlayerTeamAvailability.player_id == player.steam_id
).where(
PlayerTeamAvailability.team_id == body.team_id
PlayerTeamAvailability.team_id == json.team_id
).where(
PlayerTeamAvailability.start_time.between(window_start, window_end) |
PlayerTeamAvailability.end_time.between(window_start, window_end)
@ -148,7 +142,7 @@ def put(body: PutScheduleForm, **kwargs):
# create time regions inside our window based on the availability array
availability_blocks = []
for block in find_consecutive_blocks(body.availability):
for block in find_consecutive_blocks(json.availability):
availability_value = block[0]
hour_start = block[1]
hour_end = block[2]
@ -163,7 +157,7 @@ def put(body: PutScheduleForm, **kwargs):
new_availability.start_time = abs_start
new_availability.end_time = abs_end
new_availability.player_id = player.steam_id
new_availability.team_id = body.team_id
new_availability.team_id = json.team_id
availability_blocks.append(new_availability)
@ -182,28 +176,15 @@ def put(body: PutScheduleForm, **kwargs):
db.session.commit()
return make_response({ }, 300)
class ViewAvailablePlayersForm(pydantic.BaseModel):
class ViewAvailablePlayersForm(BaseModel):
start_time: datetime.datetime
team_id: int
@api_schedule.get("/view-available")
@validate()
@spec.validate()
@requires_authentication
def view_available(query: ViewAvailablePlayersForm, **kwargs):
start_time = query.start_time.replace(tzinfo=utc.utc)
player: Player = kwargs["player"]
#q = (
# db.select(PlayerTeamAvailability)
# .filter(
# (PlayerTeamAvailability.player_id == player.steam_id) &
# (PlayerTeamAvailability.team_id == query.team_id) &
# (PlayerTeamAvailability.start_time == start_time)
# )
#)
#availability: Sequence[PlayerTeamAvailability] = \
# db.session.execute(q).scalars().all()
def view_available(query: ViewAvailablePlayersForm, player: Player, **kwargs):
start_time = query.start_time
availability = db.session.query(
PlayerTeamAvailability

82
backend-flask/team.py
View File

@ -1,53 +1,89 @@
import datetime
from typing import List
from flask import Blueprint, jsonify, request
from flask import Blueprint, abort, jsonify, make_response, request
import pydantic
from flask_pydantic import validate
from models import Player, PlayerTeam, Team, db
from spectree import Response
from models import Player, PlayerTeam, Team, TeamSpec, db
from middleware import requires_authentication
import models
from spec import spec, BaseModel
api_team = Blueprint("team", __name__, url_prefix="/team")
@api_team.get("/view/")
@api_team.get("/view/<team_id>/")
class CreateTeamJson(BaseModel):
team_name: str
webhook_url: str
timezone: str
class ViewTeamResponse(BaseModel):
team: models.TeamSpec
class ViewTeamsResponse(BaseModel):
teams: list[models.TeamSpec]
@api_team.get("/all/")
@spec.validate(
resp=Response(
HTTP_200=ViewTeamsResponse,
HTTP_403=None,
HTTP_404=None,
)
)
@requires_authentication
def view(team_id = None, **kwargs):
def view_teams(**kwargs):
player: Player = kwargs["player"]
response = fetch_teams_for_player(player, None)
if isinstance(response, ViewTeamsResponse):
return jsonify(response.dict())
abort(404)
q_filter = PlayerTeam.player_id == player.steam_id
if team_id is not None:
q_filter = q_filter & (PlayerTeam.team_id == team_id)
@api_team.get("/id/<team_id>/")
@spec.validate(
resp=Response(
HTTP_200=ViewTeamResponse,
HTTP_403=None,
HTTP_404=None,
)
)
@requires_authentication
def view_team(team_id: int, **kwargs):
player: Player = kwargs["player"]
response = fetch_teams_for_player(player, team_id)
if isinstance(response, ViewTeamResponse):
return jsonify(response.dict())
abort(404)
def fetch_teams_for_player(player: Player, team_id: int | None):
q = db.session.query(
Team
).join(
PlayerTeam
).join(
Player
).filter(
).where(
PlayerTeam.player_id == player.steam_id
)
def map_player_team_to_player_json(player_team: PlayerTeam):
return {
"steamId": player_team.player.steam_id,
"username": player_team.player.username,
}
if team_id is not None:
q = q.where(PlayerTeam.team_id == team_id)
def map_team_to_json(team: Team):
return {
"teamName": team.team_name,
"id": team.id,
"players": list(map(map_player_team_to_player_json, team.players)),
}
def map_team_to_spec(team: Team) -> TeamSpec:
return TeamSpec(
id=team.id,
team_name=team.team_name,
discord_webhook_url=None
)
if team_id is None:
teams = q.all()
return jsonify(list(map(map_team_to_json, teams)))
return ViewTeamsResponse(
teams=list(map(map_team_to_spec, teams))
)
else:
team = q.one_or_none()
if team:
return jsonify(map_team_to_json(team))
return jsonify(), 404
return ViewTeamResponse(
team=map_team_to_spec(team)
)