Anaz
/
After_Chido_Api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

minors corrections

main
Anaz 2025-01-11 10:01:13 +04:00
parent 7f0d1209e9
commit f0ae47dd20
9 changed files with 45 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
api/v1/need_requests.py
View File

@ -34,8 +34,8 @@ async def update_need(
if not need:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Need request not found")
# Vérifie si l'utilisateur est l'auteur ou un administrateur
if need.requester_email != current_user.email and not await admin_required(db=db):
# V<EFBFBD>rifie si l'utilisateur est l'auteur ou un administrateur
if need.requester_email != current_user.email and not await AuthService.admin_required(db=db):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to update this need request",
@ -55,8 +55,8 @@ async def delete_need(
if not need:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Need request not found")
# Vérifie si l'utilisateur est l'auteur ou un administrateur
if need.requester_email != current_user.email and not await admin_required(db=db):
# V<EFBFBD>rifie si l'utilisateur est l'auteur ou un administrateur
if need.requester_email != current_user.email and not await AuthService.admin_required(db=db):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have permission to delete this need request",

4
models/db.py
View File

@ -16,8 +16,8 @@ users_table = Table(
Column('organization', String(255)),
Column('hashed_password', String(255), nullable=False),
Column('role', String(50), nullable=False),
Column('is_active', Boolean, default=True),
Column('is_banned', Boolean, default=False)
Column('is_banned', Boolean, default=False),
Column('is_deleted', Boolean, default=False)
)
# Table des rôles

4
models/schemas.py
View File

@ -33,8 +33,8 @@ class UserResponse(BaseModel):
date_of_birth: str
organization: Optional[str] = None
role: RoleResponse
is_active: bool
is_banned: bool
is_deleted: bool
class Role(BaseModel):
id: int
@ -43,7 +43,7 @@ class Role(BaseModel):
class Config:
orm_mode = True
class UserUpdateRole(BaseModel):
email: EmailStr
new_role: str

31
services/auth_service.py
View File

@ -7,7 +7,7 @@ from fastapi.security import OAuth2PasswordBearer
from config.settings import settings
from models.schemas import TokenData, UserCreate, UserResponse
from config.database import get_db
from models.db import users_table
from models.db import users_table, roles_table, role_permissions_table, permissions_table
from sqlalchemy import select, update, insert
from sqlalchemy.ext.asyncio import AsyncSession
from utils.logging import logger
@ -19,7 +19,7 @@ logger.info("Test log message")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Configuration pour OAuth2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
class AuthService:
@staticmethod
@ -81,9 +81,28 @@ class AuthService:
query = select(users_table).where(users_table.c.email == email)
result = await db.execute(query)
user = result.fetchone()
user = result.mappings().fetchone()
if user is None:
raise credentials_exception
# Récupérer le rôle et ses permissions
role_query = select(roles_table.c.id, roles_table.c.name, permissions_table.c.name.label("permission")).join(
role_permissions_table, role_permissions_table.c.role_id == roles_table.c.id
).join(
permissions_table, role_permissions_table.c.permission_id == permissions_table.c.id
).where(roles_table.c.name == user["role"])
role_result = await db.execute(role_query)
role_data = role_result.mappings().all()
if not role_data:
raise credentials_exception
role = {
"id": role_data[0]["id"],
"name": role_data[0]["name"],
"permissions": [r["permission"] for r in role_data]
}
# Préparez la réponse avec tous les champs requis
return {
@ -92,9 +111,9 @@ class AuthService:
"full_name": user["full_name"],
"phone": user["phone"],
"date_of_birth": user["date_of_birth"].isoformat(),
"role": user["role"],
"is_active": not user["is_blocked"],
"is_banned": user["is_deleted"],
"role": role,
"is_banned": user["is_banned"],
"is_deleted": user["is_deleted"],
}
@staticmethod

1
services/message_service.py
View File

@ -1,3 +1,4 @@
from fastapi import HTTPException
from sqlalchemy import insert
from models.schemas import TechnicalIssue
from config.database import get_db

12
services/need_request_service.py
View File

@ -9,7 +9,7 @@ from config.settings import settings
from jose import jwt, JWTError
from datetime import datetime
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
class NeedRequestService:
@ -53,7 +53,7 @@ class NeedRequestService:
@staticmethod
async def update_need(need_id: int, data: dict, db, token: str):
user = await NeedRequestService.verify_requester_or_admin(need_id, token, db)
if "deleted" in data: # Empêche la mise à jour directe du champ `deleted`
if "deleted" in data: # Emp<EFBFBD>che la mise <20> jour directe du champ `deleted`
raise HTTPException(status_code=400, detail="Invalid update field")
query = (
@ -97,7 +97,7 @@ class NeedRequestService:
detail="You do not have permission to perform this action.",
)
try:
# Décodage du token JWT
# D<EFBFBD>codage du token JWT
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
email: str = payload.get("sub")
if email is None:
@ -106,21 +106,21 @@ class NeedRequestService:
except JWTError:
raise credentials_exception
# Récupère l'utilisateur depuis la base de données
# R<EFBFBD>cup<EFBFBD>re l'utilisateur depuis la base de donn<6E>es
user_query = select(users_table).where(users_table.c.email == token_data.email)
result = await db.execute(user_query)
user = result.fetchone()
if user is None:
raise credentials_exception
# Récupère la demande de besoin
# R<EFBFBD>cup<EFBFBD>re la demande de besoin
need_query = select(need_requests_table).where(need_requests_table.c.id == need_id)
result = await db.execute(need_query)
need = result.fetchone()
if need is None:
raise HTTPException(status_code=404, detail="Need request not found")
# Vérifie si l'utilisateur est l'auteur ou un administrateur
# V<EFBFBD>rifie si l'utilisateur est l'auteur ou un administrateur
if need["requester_email"] != user["email"] and user["role"] != "admin":
raise credentials_exception

2
services/person_report_service.py
View File

@ -7,7 +7,7 @@ from typing import Optional
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
class PersonReportService:

2
services/role_service.py
View File

@ -8,7 +8,7 @@ from config.settings import settings
from jose import jwt, JWTError
from sqlalchemy.ext.asyncio import AsyncSession
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
class RoleService:

6
services/user_service.py
View File

@ -1,4 +1,4 @@
from sqlalchemy import update, select
from sqlalchemy import select, update, insert
from fastapi import Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from models.schemas import UserCreate, UserResponse, UserUpdateRole, UserBlockBan
@ -12,11 +12,11 @@ class UserService:
@staticmethod
async def list_users(status: Optional[str] = None, db=Depends(get_db)):
query = select(person_reports_table)
query = select(users_table)
if status:
query = query.where(users_table.c.status == status)
result = await db.execute(query)
users = result.fetchall()
users = result.mappings().all()
return [UserResponse(**user) for user in users]
@staticmethod