agenda/backend-collector
1
0
Fork 0
Code Issues Pull Requests Projects Releases 37 Wiki Activity

change delete method events

Browse Source
This commit is contained in:
Valentin CZERYBA 2023-11-08 22:43:53 +01:00
parent 6cc43005f0
commit a85582d200
2 changed files with 82 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
app/models/events.py
View File

@ -7,9 +7,9 @@ class Event(BaseModel):
name: str name: str
place: str place: str
status: int = 0 status: int = 0
participants: list[str] | None = None participants: list[str] = []
organizers: list[str] | None = None organizers: list[str] = []
interested: list[str] | None = None interested: list[str] = []
start_date: datetime | None = None start_date: datetime | None = None
end_date: datetime | None = None end_date: datetime | None = None
created_at: datetime = datetime.today() created_at: datetime = datetime.today()
@ -22,20 +22,24 @@ class EventOut(BaseModel):
name: str name: str
place: str place: str
status: int = 0 status: int = 0
participants: list[str] | None = None participants: list[str] = []
organizers: list[str] | None = None organizers: list[str] = []
interested: list[str] | None = None interested: list[str] = []
start_date: datetime | None = None start_date: datetime | None = None
end_date: datetime | None = None
class EventIn(BaseModel): class EventIn(BaseModel):
name: str name: str
place: str place: str
status: int = 0 status: int = 0
participants: list[str] | None = None participants: list[str] = []
organizers: list[str] | None = None organizers: list[str] = []
interested: list[str] | None = None interested: list[str] = []
start_date: datetime | None = None start_date: datetime | None = None
end_date: datetime | None = None
class EventIDS(BaseModel):
ids: list[str]
class EventRepository(AbstractRepository[Event]): class EventRepository(AbstractRepository[Event]):

119
app/routers/events.py
View File

@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException, status, Response
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from datetime import datetime from datetime import datetime
from ..dependencies import users_token, permissions_checker, database from ..dependencies import users_token, permissions_checker, database
from ..models import events from ..models import events, users
from pydantic import EmailStr from pydantic import EmailStr
from typing import Annotated from typing import Annotated
from bson import ObjectId from bson import ObjectId
@ -35,88 +35,107 @@ async def read_events(authorize: Annotated[bool, Depends(permissions_checker.Per
for event_index in event_repository.find_by(object_search, limit=limit, skip=skip): for event_index in event_repository.find_by(object_search, limit=limit, skip=skip):
event = users.UserOut(id=event_index.id, name=event_index.name, place=event_index.place, status=event_index.status, start_date=event_index.start_date, end_date=event_index.end_date) event = events.EventOut(id=event_index.id, name=event_index.name, place=event_index.place, status=event_index.status, start_date=event_index.start_date, end_date=event_index.end_date)
listEvents.append(event) listEvents.append(event)
return listEvents return listEvents
@router.get("/users/me",tags=["users"], response_model=users.User, response_model_exclude=["id", "password", "roles", "status"]) @router.get("/events/me",tags=["events"])
async def read_users_me(current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))]): async def read_users_me(current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))]):
return current_user event_repository = events.EventRepository(database=database.database)
listOrganizers = []
listParticipants = []
listInterested = []
@router.get("/users/count", tags=["users"]) for event_index in event_repository.find_by({"organizers":{"$eq": current_user.id}}, limit=limit, skip=skip):
async def read_users_count(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))]): event = events.EventOut(id=event_index.id, name=event_index.name, place=event_index.place, status=event_index.status, start_date=event_index.start_date, end_date=event_index.end_date)
count = database.database.get_collection("users").estimated_document_count() listOrganizers.append(event)
for event_index in event_repository.find_by({"participants":{"$eq": current_user.id}}, limit=limit, skip=skip):
event = events.EventOut(id=event_index.id, name=event_index.name, place=event_index.place, status=event_index.status, start_date=event_index.start_date, end_date=event_index.end_date)
listParticipants.append(event)
for event_index in event_repository.find_by({"interested":{"$eq": current_user.id}}, limit=limit, skip=skip):
event = events.EventOut(id=event_index.id, name=event_index.name, place=event_index.place, status=event_index.status, start_date=event_index.start_date, end_date=event_index.end_date)
listInterested.append(event)
content = {"participants":listParticipants, "organizers":listOrganizers, "interested":listInterested}
response = JSONResponse(content=content)
return response
@router.get("/events/count", tags=["events"])
async def read_events_count(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))]):
count = database.database.get_collection("events").estimated_document_count()
content = {"count":count} content = {"count":count}
response = JSONResponse(content=content) response = JSONResponse(content=content)
return response return response
@router.get("/users/{item_id}", tags=["users"], response_model=users.User) @router.get("/events/{item_id}", tags=["events"], response_model=events.Event)
async def read_users_id(item_id : str, authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))]): async def read_events_id(item_id : str, authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))]):
user_repository = users.UserRepository(database=database.database) event_repository = event.EventRepository(database=database.database)
user = user_repository.find_one_by_id(ObjectId(item_id)) event = event_repository.find_one_by_id(ObjectId(item_id))
return user return event
@router.delete("/users/me",tags=["users"]) @router.delete("/events/me/{item_id}", tags=["events"])
async def delete_users_me(current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], remove: bool = False): async def delete_event_me(item_id: str, current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], remove: bool = False):
user_repository = users.UserRepository(database=database.database) event_repository = event.EventRepository(database=database.database)
event = event_repository.find_one_by_id(ObjectId(item_id))
if remove is True: if remove is True:
current_user.deleted_at = datetime.today() event.deleted_at = datetime.today()
current_user.status = -1 event.status = -1
content = {"message": "users are deleted"} content = {"message": "event is deleted"}
else: else:
current_user.status = 0 event.status = 0
current_user.deleted_at = datetime.today() event.deleted_at = datetime.today()
content = {"message": "users are disabled"} content = {"message": "event is disabled"}
user_repository.save(current_user) event_repository.save(event)
response = JSONResponse(content=content) response = JSONResponse(content=content)
return response return response
@router.delete("/users/groups",tags=["users"]) @router.delete("/events/groups",tags=["events"])
async def delete_users_groups(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove: bool = False, userids: users.UserIDS | None = None): async def delete_events_groups(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove: bool = False, eventids: events.EventIDS | None = None):
if len(userids.ids) == 0: if len(eventids.ids) == 0:
raise HTTPException( raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, status_code=status.HTTP_400_BAD_REQUEST,
detail="userids should be greater than 0" detail="eventids should be greater than 0"
) )
user_repository = users.UserRepository(database=database.database) event_repository = event.EventRepository(database=database.database)
for i in userids.ids: for i in eventids.ids:
user = user_repository.find_one_by_id(ObjectId(i)) event = event_repository.find_one_by_id(ObjectId(i))
if remove is True: if remove is True:
user.status = -1 event.status = -1
user.deleted_at = datetime.today() event.deleted_at = datetime.today()
content = {"message": "users are deleted "} content = {"message": "events are deleted "}
else: else:
user.status = 0 event.status = 0
user.disabled_at = datetime.today() event.disabled_at = datetime.today()
content = {"message": "users are disabled"} content = {"message": "events are disabled"}
user_repository.save(user) event_repository.save(event)
response = JSONResponse(content=content) response = JSONResponse(content=content)
return response return response
@router.delete("/users/{item_id}", tags=["users"]) @router.delete("/events/{item_id}", tags=["events"])
async def delete_users_id(item_id : str, authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove : bool = False): async def delete_events_id(item_id : str, authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove : bool = False):
user_repository = users.UserRepository(database=database.database) event_repository = event.EventRepository(database=database.database)
user = user_repository.find_one_by_id(ObjectId(item_id)) event = event_repository.find_one_by_id(ObjectId(item_id))
if remove is True: if remove is True:
user.status = -1 event.status = -1
user.deleted_at = datetime.today() event.deleted_at = datetime.today()
content = {"message": "users are deleted"} content = {"message": "events are deleted"}
else: else:
user.status = 0 event.status = 0
user.disabled_at = datetime.today() event.disabled_at = datetime.today()
content = {"message": "users are disabled"} content = {"message": "events are disabled"}
user_repository.save(user) event.save(user)
response = JSONResponse(content=content) response = JSONResponse(content=content)
return response return response
@router.put("/users/me",tags=["users"]) @router.put("/events/me/{item_id}",tags=["users"])
async def update_users_me(current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], userSingle: users.UserIn | None = None): async def update_users_me(item_id: str, current_user: Annotated[users.User, Depends(users_token.get_current_active_user)], authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))], userSingle: users.UserIn | None = None):
user_repository = users.UserRepository(database=database.database) user_repository = users.UserRepository(database=database.database)
current_user.username = userSingle.username current_user.username = userSingle.username
current_user.password = user_token.get_password_hash(userSingle.password) current_user.password = user_token.get_password_hash(userSingle.password)