73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
from datetime import timedelta
|
|
|
|
from fastapi import FastAPI, HTTPException, status, Depends
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.security import OAuth2PasswordRequestForm
|
|
from fastapi.responses import JSONResponse
|
|
from schemas.user_schemas import UserCreateInfo, UserInfo
|
|
from auth import validate_user, ACCESS_TOKEN_EXPIRE_MINUTES, create_access_token, get_current_user
|
|
from db import db
|
|
from db.users_manip import register_user
|
|
|
|
app = FastAPI()
|
|
|
|
db.init(app, True)
|
|
|
|
|
|
|
|
# TODO: Задание через конфиг
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:5173"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
|
|
@app.get("/test")
|
|
async def test():
|
|
await register_user(UserCreateInfo(username="test", password="test"))
|
|
return {"message": "Hello World"}
|
|
|
|
|
|
@app.post("/login")
|
|
async def login(user: OAuth2PasswordRequestForm = Depends()):
|
|
user = await validate_user(user)
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect username or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
access_token = create_access_token(
|
|
data={"sub": user.username}, expires_delta=access_token_expires
|
|
)
|
|
token = jsonable_encoder(access_token)
|
|
content = {"message": "You've successfully logged in. Welcome back!"}
|
|
response = JSONResponse(content=content)
|
|
response.set_cookie(
|
|
"Authorization",
|
|
value=f"Bearer {token}",
|
|
httponly=True,
|
|
max_age=1800,
|
|
expires=1800,
|
|
samesite="lax",
|
|
secure=False,
|
|
)
|
|
|
|
return response
|
|
|
|
|
|
@app.get("/getCurrentUserInfo", response_model=UserInfo, dependencies=[Depends(get_current_user)])
|
|
async def get_current_user_info(current_user: UserInfo = Depends(get_current_user)):
|
|
return current_user
|