delete old dep

This commit is contained in:
IluaAir
2025-09-28 22:20:34 +03:00
parent ddaa5b0c74
commit 64dcc77518

View File

@@ -1,6 +1,6 @@
from typing import Annotated from typing import Annotated
from fastapi import Depends, HTTPException, Path from fastapi import Depends, HTTPException
from fastapi.security import ( from fastapi.security import (
HTTPAuthorizationCredentials, HTTPAuthorizationCredentials,
HTTPBearer, HTTPBearer,
@@ -12,13 +12,11 @@ from src.api.dependacies.db_dep import sessionDep
from src.core.auth_manager import AuthManager from src.core.auth_manager import AuthManager
from src.core.settings import settings from src.core.settings import settings
from src.schemas.auth import TokenData from src.schemas.auth import TokenData
from src.services.tasks import TaskService
from src.services.users import UserService from src.services.users import UserService
http_bearer = HTTPBearer(auto_error=False) http_bearer = HTTPBearer(auto_error=False)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api.v1_login_url}/login") oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.api.v1_login_url}/login")
# AccessTokenDep = Annotated[str, Depends(oauth2_scheme)]
AccessTokenDep = Annotated[HTTPAuthorizationCredentials, Depends(http_bearer)] AccessTokenDep = Annotated[HTTPAuthorizationCredentials, Depends(http_bearer)]
@@ -65,37 +63,3 @@ async def get_admin_user(db: sessionDep, current_user: ActiveUser):
AdminUser = Annotated[TokenData, Depends(get_admin_user)] AdminUser = Annotated[TokenData, Depends(get_admin_user)]
async def user_or_admin(db: sessionDep, current_user: ActiveUser, owner_id: int):
if current_user.id == owner_id:
return current_user
else:
admin = await get_admin_user(db, current_user)
return admin
async def CurrentOrAdminOwner(
db: sessionDep, current_user: ActiveUser, id: Annotated[int, Path()]
):
authorized_user = await user_or_admin(db, current_user, id)
if not authorized_user:
raise HTTPException(status_code=403, detail="Not authorized")
return authorized_user
async def CurrentOrAdminTask(
db: sessionDep,
id: Annotated[int, Path()],
current_user: ActiveUser,
):
task = await TaskService(db).get_task(id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return await CurrentOrAdminOwner(db, current_user, task.user_id)
OwnerDep = Annotated[TokenData, Depends(CurrentOrAdminOwner)]
TaskOwnerDep = Annotated[TokenData, Depends(CurrentOrAdminTask)]