MH.Dmitrii/The_DisExcel_project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

create, delete, patch user

Browse Source
This commit is contained in:
MH.Dmitrii
2026-08-14 13:24:55 +03:00
parent d3244666af
commit 33aa3cb7a4
5 changed files with 168 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
makefile
+3
View File
@@ -13,6 +13,9 @@ help:
##
## Init section
##
.PHONY: run
run: ## Run dev local application
${VENV} uvicorn main:app --host 0.0.0.0 --port 8000 --reload
.PHONY: run-dev
run-dev: ## Run dev application
src/database/users/crud.py
+85 -3
View File
@@ -1,9 +1,14 @@
from uuid import UUID
from sqlalchemy import select
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import async_sessionmaker
from src.models.database_models.model import User, engine
from src.models.database_models.model import (
Permissions,
PermissionsGroups,
User,
engine,
)
from src.models.pydantic_models.model import UserOutDB
@@ -13,17 +18,94 @@ class UsersCrudActions:
self.Session=async_sessionmaker(bind=engine)
async def get_user_by_email(self, email:str)->UserOutDB|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.email==email)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
return UserOutDB.model_validate(response)
async def get_user_by_id(self, id:UUID)->UserOutDB|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.id==id)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
return UserOutDB.model_validate(response)
return UserOutDB.model_validate(response)
async def create_user(self, data:dict)->UserOutDB|None:
async with self.Session() as session, session.begin():
groups_name=data.pop("group", None)
permissions_name=data.pop("direct_permissions", None)
new_user=User(**data)
if groups_name:
query=select(PermissionsGroups).where(PermissionsGroups.group.in_(groups_name))
response=(await session.scalars(query)).all()
if response is None:
new_user.group=[]
else:
new_user.group=list(response)
else:
new_user.group=[]
if permissions_name:
query=select(Permissions).where(Permissions.permission.in_(permissions_name))
response=(await session.scalars(query)).all()
if response is None:
new_user.direct_permissions=[]
else:
new_user.direct_permissions=list(response)
else:
new_user.direct_permissions=[]
session.add(new_user)
await session.flush()
return UserOutDB.model_validate(new_user)
async def delete_user(self, user_email:str)->bool|None:
async with self.Session() as session, session.begin():
query=select(User).where(User.email == user_email)
response=(await session.scalars(query)).one_or_none()
if response is None:
return None
response.status=False
return True
async def update_user(self, user_email:str, data:dict)->UserOutDB|None:
async with self.Session() as session, session.begin():
groups_name=data.pop("group", None)
permissions_name=data.pop("direct_permissions", None)
query = update(User).where(User.email == user_email).values(**data).returning(User)
user_edit=(await session.scalars(query)).one_or_none()
if user_edit is None:
return None
if groups_name is not None:
query=select(PermissionsGroups).where(PermissionsGroups.group.in_(permissions_name))
groups=(await session.scalars(query)).all()
user_edit.group=list(groups)
if permissions_name is not None:
query=select(Permissions).where(Permissions.permission.in_(permissions_name))
groups=(await session.scalars(query)).all()
user_edit.direct_permissions=list(groups)
await session.flush()
return UserOutDB.model_validate(user_edit)
src/models/pydantic_models/model.py
+18 -4
View File
@@ -2,14 +2,28 @@ from datetime import datetime
from typing import Annotated
from uuid import UUID
from pydantic import BaseModel, EmailStr, Field
from pydantic import AfterValidator, BaseModel, EmailStr, Field
def validate_password(password: str) -> str:
PUNCTUATION: set[str] = {"$", "@", "#", "%", "!", "^", "&", "*", "(", ")", "-", "_", "+", "=", "{", "}", "[", "]"}
if len(password) < 8 or len(password) > 72:
raise ValueError("Password must be 8-72 characters")
if (
not any(c.isupper() for c in password)
or not any(c.islower() for c in password)
or not any(c.isdigit() for c in password)
or not any(c in PUNCTUATION for c in password)
):
raise ValueError("Password must contain uppercase, lowercase, digit and special char")
return password
PasswordStr = Annotated[str, AfterValidator(validate_password)]
class Base(BaseModel):
model_config = {"from_attributes": True}
class PermissionsCreate(Base):
permission:Annotated[str, Field(..., max_length=30, description="permission name")]
@@ -35,8 +49,7 @@ class UserCreate(Base):
last_name:Annotated[str, Field(...,max_length=64, description="last name of the user")]
middle_name:Annotated[str, Field(...,max_length=64, description="middle name of the user")]
email:Annotated[EmailStr, Field(...,min_length=5, max_length=255, description="email of the user")]
plain_password:Annotated[str, Field(...,min_length=8,max_length=72, description="plain password of the user")]
status:Annotated[bool, Field(..., description="status of the user")]
plain_password:Annotated[PasswordStr, Field(..., description="plain password of the user")]
direct_permissions:Annotated[list[str], Field(..., description="permissions of the user")]
group:Annotated[list[str], Field(..., description="permissions groups of the user")]
@@ -63,6 +76,7 @@ class UserUpdate(Base):
last_name:Annotated[str|None, Field(None, max_length=64,description="last name of the user")]
middle_name:Annotated[str|None, Field(None, max_length=64,description="middle name of the user")]
email:Annotated[EmailStr|None, Field(None, min_length=5, max_length=255, description="email of the user")]
plain_password:Annotated[PasswordStr|None, Field(None, description="plain password of the user")]
status:Annotated[bool|None, Field(None, description="status of the user")]
direct_permissions:Annotated[list[str]|None, Field(None, description="permissions of the user")]
group:Annotated[list[str]|None, Field(None, description="permissions groups of the user")]
src/service/users_crud/users_crud.py
+47 -1
View File
@@ -1,6 +1,7 @@
from src.database.users.crud import UsersCrudActions
from src.errors.http_errors.errors import Errors
from src.models.pydantic_models.model import UserOut
from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
from src.service.auth.jwt import HashService
class CrudService:
@@ -8,12 +9,57 @@ class CrudService:
def __init__(self) -> None:
self.errors=Errors()
self.crud_db_actions=UsersCrudActions()
self.hash_service=HashService()
async def get_user_by_email(self, email:str)->UserOut:
user_entity=await self.crud_db_actions.get_user_by_email(email)
if not user_entity:
raise self.errors.not_found_error(detail="User wasn't found")
return UserOut.model_validate(user_entity)
async def create_user(self, data:UserCreate)->UserOut:
user_data=UserCreate.model_dump(data)
user_data["hashed_password"]=user_data.pop("plain_password")
user_data["hashed_password"]=self.hash_service.plain_to_hash(user_data["hashed_password"])
user_entity=await self.crud_db_actions.create_user(user_data)
if not user_entity:
raise self.errors.validation_error(detail="User creation gone wrong")
return UserOut.model_validate(user_entity)
async def delete_user(self, email:str)->bool:
user_entity=await self.crud_db_actions.delete_user(email)
if not user_entity:
raise self.errors.not_found_error(detail="User not found")
return user_entity
async def update_user(self, email:str, data:UserUpdate)->UserOut:
user_data=UserUpdate.model_dump(data, exclude_unset=True)
if not user_data:
raise self.errors.bad_request_error(detail="User info to update can not be empty")
if user_data.get("plain_password"):
user_data["hashed_password"]=user_data.pop("plain_password")
user_data["hashed_password"]=self.hash_service.plain_to_hash(user_data["hashed_password"])
user_entity=await self.crud_db_actions.update_user(email, user_data)
if not user_entity:
raise self.errors.not_found_error(detail="User not found")
return UserOut.model_validate(user_entity)
async def crud_service()->CrudService:
return CrudService()
src/web/protected_routes/protected_user_action_routes.py
+15 -2
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter, Depends
from src.models.pydantic_models.model import UserOut
from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
from src.service.users_crud.users_crud import CrudService, crud_service
from src.web.protected_routes.auth_routes import get_current_user
@@ -8,4 +8,17 @@ router=APIRouter(prefix="/user")
@router.get("/get_by_email")
async def get_current_user_by_email(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->UserOut: # noqa: B008
return await crud.get_user_by_email(email)
return await crud.get_user_by_email(email)
@router.post("/create_user")
async def create_user(data:UserCreate, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->UserOut: #noqa: B008
return await crud.create_user(data)
@router.post("/delete_user")
async def delete_user(email:str, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->bool: #noqa: B008
return await crud.delete_user(email)
@router.post("/patch_user")
async def patch_user(email:str, data:UserUpdate, crud:CrudService=Depends(crud_service), current_user=Depends(get_current_user))->UserOut: #noqa: B008
return await crud.update_user(email, data)