add check is_admin route

This commit is contained in:
2025-12-06 01:50:02 +05:00
parent 14f7929e0f
commit 8f0a5abfb3
2 changed files with 29 additions and 0 deletions

View File

@ -1,4 +1,5 @@
from fastapi import APIRouter, HTTPException, Body, Response from fastapi import APIRouter, HTTPException, Body, Response
from fastapi.params import Query
from app.models.user import UserCreate, UserLogin, VerifyCode from app.models.user import UserCreate, UserLogin, VerifyCode
from app.models.request import ValidateRequest from app.models.request import ValidateRequest
from app.services.auth import AuthService from app.services.auth import AuthService
@ -129,3 +130,13 @@ async def generate_code(username: str):
@router.get("/auth/verification_status/{username}") @router.get("/auth/verification_status/{username}")
async def get_verification_status(username: str): async def get_verification_status(username: str):
return await AuthService().get_verification_status(username) return await AuthService().get_verification_status(username)
@router.get("/auth/me")
async def get_me(
accessToken: str = Query(...),
clientToken: str = Query(...),
):
"""
Текущий пользователь по accessToken + clientToken.
"""
return await AuthService().get_current_user(accessToken, clientToken)

View File

@ -143,6 +143,24 @@ class AuthService:
user = await users_collection.find_one({"uuid": session["user_uuid"]}) user = await users_collection.find_one({"uuid": session["user_uuid"]})
return user and user.get("is_admin") is True return user and user.get("is_admin") is True
async def get_current_user(self, access_token: str, client_token: str):
session = await sessions_collection.find_one({
"access_token": access_token,
"client_token": client_token,
})
if not session:
raise HTTPException(status_code=401, detail="Invalid session")
user = await users_collection.find_one({"uuid": session["user_uuid"]})
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {
"username": user["username"],
"uuid": user["uuid"],
"is_admin": user.get("is_admin", False),
}
async def refresh(self, access_token: str, client_token: str): async def refresh(self, access_token: str, client_token: str):
if not await self.validate(access_token, client_token): if not await self.validate(access_token, client_token):
return None return None