126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
from datetime import datetime, timedelta
|
|
from typing import List, Optional
|
|
|
|
import jwt
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from passlib.context import CryptContext
|
|
from sqlalchemy.orm import Session
|
|
|
|
import crud
|
|
from database import SessionLocal, engine
|
|
from enums import MinigameEnum, CourseEnum
|
|
from models import Base
|
|
from schemas import highscores, users, courseprogress
|
|
|
|
app = FastAPI()
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
# JWT authentication setup
|
|
jwt_secret = "secret_key"
|
|
ALGORITHM = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = 30
|
|
|
|
bearer_scheme = HTTPBearer()
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello world!"}
|
|
|
|
|
|
@app.get("/users", response_model=List[users.User])
|
|
async def read_users(db: Session = Depends(get_db)):
|
|
users = crud.get_users(db)
|
|
return users
|
|
|
|
|
|
@app.post("/users", response_model=users.User)
|
|
async def create_user(user: users.UserCreate, db: Session = Depends(get_db)):
|
|
db_user = crud.get_user_by_username(db, username=user.username)
|
|
if db_user:
|
|
raise HTTPException(status_code=400, detail="Username already registered")
|
|
return crud.create_user(
|
|
db=db, username=user.username, hashed_password=pwd_context.hash(user.password)
|
|
)
|
|
|
|
|
|
@app.get("/highscores", response_model=List[users.UserHighScore])
|
|
async def read_high_scores(
|
|
db: Session = Depends(get_db),
|
|
minigame: Optional[MinigameEnum] = None,
|
|
n_highest: Optional[int] = None,
|
|
):
|
|
high_scores = crud.get_high_scores(db, minigame, n_highest)
|
|
return high_scores
|
|
|
|
|
|
@app.post("/highscores", response_model=highscores.HighScore)
|
|
async def create_high_score(
|
|
high_score: highscores.HighScoreCreate, db: Session = Depends(get_db)
|
|
):
|
|
return crud.create_high_score(db=db, high_score=high_score)
|
|
|
|
|
|
#### TESTING!! DELETE LATER
|
|
|
|
|
|
async def get_current_user(
|
|
token: HTTPAuthorizationCredentials = Depends(bearer_scheme),
|
|
):
|
|
try:
|
|
payload = jwt.decode(token.credentials, jwt_secret, algorithms=[ALGORITHM])
|
|
username = payload.get("sub")
|
|
if username is None:
|
|
raise HTTPException(status_code=401, detail="Invalid JWT token")
|
|
return username
|
|
except jwt.exceptions.DecodeError:
|
|
raise HTTPException(status_code=401, detail="Invalid JWT token")
|
|
|
|
|
|
@app.get("/protected")
|
|
async def protected_route(current_user = Depends(get_current_user)):
|
|
return {"message": f"Hello, {current_user}!"}
|
|
|
|
|
|
def authenticate_user(user: users.UserCreate, db: Session = Depends(get_db)):
|
|
db_user = crud.get_user_by_username(db=db, username=user.username)
|
|
if not db_user:
|
|
return False
|
|
hashed_password = db_user.hashed_password
|
|
if not hashed_password or not pwd_context.verify(user.password, hashed_password):
|
|
return False
|
|
return db_user
|
|
|
|
|
|
@app.post("/login")
|
|
async def login(user: users.UserCreate, db: Session = Depends(get_db)):
|
|
user = authenticate_user(user, db)
|
|
if not user:
|
|
raise HTTPException(status_code=401, detail="Invalid username or password")
|
|
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
access_token_payload = {
|
|
"sub": user.username,
|
|
"exp": datetime.utcnow() + access_token_expires,
|
|
}
|
|
access_token = jwt.encode(access_token_payload, jwt_secret, algorithm=ALGORITHM)
|
|
return {"access_token": access_token}
|
|
|
|
|
|
@app.get("/courseprogress", response_model=List[courseprogress.CourseProgressBase])
|
|
async def get_course_progress(course: Optional[CourseEnum] = CourseEnum.All, current_user = Depends(get_current_user), db: Session = Depends(get_db)):
|
|
user = crud.get_user_by_username(db, current_user)
|
|
return crud.get_course_progress(db = db, user = user, course = course)
|