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

ipaddress router wip

Browse Source
This commit is contained in:
Valentin CZERYBA 2024-04-01 23:06:59 +02:00
parent 06b50247fd
commit 6ef378ab7a
2 changed files with 36 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
app/models/ipaddress.py
View File

@ -28,6 +28,10 @@ class IpAddressIn(BaseModel):
status: int = 0
class IpAddressIDS(BaseModel):
ids: list[str]
class IpAddressRepository(AbstractRepository[IpAdress]):
class Meta:
collection_name = "ipaddress"

67
app/routers/ipaddress.py
View File

@ -18,7 +18,7 @@ async def read_events(authorize: Annotated[bool, Depends(permissions_checker.Per
detail="skip should be greater than 0 and limit should be greater than 1. Limit should be greater than skip"
)
limit = limit + skip
listEvents = []
listIp = []
ip_repository = ipaddress.IpAddressRepository(database=database.database)
object_search = {}
if status is not None:
@ -34,65 +34,62 @@ async def read_events(authorize: Annotated[bool, Depends(permissions_checker.Per
object_search = {"$and":[{"ip":{"$regex": ip}}, {"status":{"$eq":status}}]}
for event_index in event_repository.find_by(object_search, 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)
listEvents.append(event)
return listEvents
for ip_index in ip_repository.find_by(object_search, limit=limit, skip=skip):
ip = ipaddress.IpAddressOut(id=ip_index.id, username=ip_index.username, status=ip_index.status, ip=ip_index.ip, socialnetwork=ip_index.socialnetwork)
listIp.append(ip)
return listIp
@router.get("/events/me",tags=["events"])
@router.get("/ipaddress/me",tags=["ipaddress"])
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"]))]):
event_repository = events.EventRepository(database=database.database)
listOrganizers = []
ip_repository = ipaddress.IpAddressRepository(database=database.database)
listIp = []
for event_index in event_repository.find_by({"organizers":{"$eq": current_user.username}}, 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)
listOrganizers.append(event)
for ip_index in ip_repository.find_by({"username":{"$eq": current_user.username}}, limit=limit, skip=skip):
ip = ipaddress.IpAddressOut(id=ip_index.id, username=ip_index.username, status=ip_index.status, ip=ip_index.ip, socialnetwork=ip_index.socialnetwork)
listIp.append(ip)
return listIp
content = {"organizers":listOrganizers}
response = JSONResponse(content=content)
return response
@router.get("/events/count", tags=["events"])
@router.get("/ipaddress/count", tags=["ipaddress"])
async def read_events_count(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin", "User"]))]):
count = database.database.get_collection("events").estimated_document_count()
count = database.database.get_collection("ipadress").estimated_document_count()
content = {"count":count}
response = JSONResponse(content=content)
return response
@router.get("/events/{item_id}", tags=["events"], response_model=events.Event)
@router.get("/ipaddress/{item_id}", tags=["ipaddress"], response_model=ipaddress.IpAddress)
async def read_events_id(item_id : str, authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))]):
event_repository = events.EventRepository(database=database.database)
event = event_repository.find_one_by_id(ObjectId(item_id))
return event
ip_repository = ipaddressIpAddressRepository(database=database.database)
ip = ip_repository.find_one_by_id(ObjectId(item_id))
return ip
@router.delete("/events/me/{item_id}", tags=["events"])
@router.delete("/ipaddress/me/{item_id}", tags=["ipaddress"])
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):
event_repository = events.EventRepository(database=database.database)
event = event_repository.find_one_by_id(ObjectId(item_id))
ip_repository = ipaddress.IpAddressRepository(database=database.database)
ip = ip_repository.find_one_by_id(ObjectId(item_id))
if remove is True:
event.deleted_at = datetime.today()
event.status = -1
content = {"message": "event is deleted"}
ip.deleted_at = datetime.today()
ip.status = -1
content = {"message": "ip is deleted"}
else:
event.status = 0
event.deleted_at = datetime.today()
content = {"message": "event is disabled"}
event_repository.save(event)
ip.status = 0
ip.deleted_at = datetime.today()
content = {"message": "ip is disabled"}
ip_repository.save(ip)
response = JSONResponse(content=content)
return response
@router.delete("/events/groups",tags=["events"])
async def delete_events_groups(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove: bool = False, eventids: events.EventIDS | None = None):
if len(eventids.ids) == 0:
@router.delete("/ipaddress/groups",tags=["ipaddress"])
async def delete_events_groups(authorize: Annotated[bool, Depends(permissions_checker.PermissionChecker(roles=["Admin"]))], remove: bool = False, ipaddressids: ipaddress.IpAddressIDS | None = None):
if len(ipaddressids.ids) == 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="eventids should be greater than 0"
)
event_repository = event.EventRepository(database=database.database)
ip_repository = ipaddress.IpAddressRepository(database=database.database)
for i in eventids.ids:
event = event_repository.find_one_by_id(ObjectId(i))
if remove is True: