42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
from fastapi import FastAPI, Depends, HTTPException,status
|
|
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
|
import server.backend.schema.pydantic as pydantic
|
|
import server.backend.database.db as db
|
|
from server.backend.auth.JWT import signJWT, decodeJWT
|
|
api = FastAPI()
|
|
security = HTTPBearer()
|
|
|
|
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
|
|
token = credentials.credentials
|
|
user = decodeJWT(token)
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
return user
|
|
async def check_roles(user=Depends(get_current_user)):
|
|
if user.get("admin") != True:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access denied")
|
|
return user
|
|
|
|
@api.post("/update", response_model=pydantic.UserUpdate)
|
|
async def update_user(data: pydantic.UserUpdate,user=Depends(get_current_user)):
|
|
data = await db.update_user(data)
|
|
return data
|
|
|
|
@api.post("/create", response_model=pydantic.UserAccess)
|
|
async def create_user(user_info: pydantic.UserCreate,user=Depends(check_roles)):
|
|
await db.create_user(user_info)
|
|
return user_info
|
|
|
|
@api.get("/list")
|
|
async def list_users(user=Depends(check_roles)):
|
|
list_of_users = await db.list_users()
|
|
return list_of_users
|
|
|
|
@api.post("/auth",response_model=pydantic.Token)
|
|
async def auth(code:pydantic.UserAccess):
|
|
login = await db.login_user(code)
|
|
if login == None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Forbidden")
|
|
token = signJWT(login)
|
|
return {"access_token": token, "token_type": "bearer"}
|