ap7pd/event-management-backend
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Allow filtering events by date

Browse source
This commit is contained in:
Peter Vacho 2024-12-28 12:16:00 +01:00
parent f83f0d1ae1
commit 9186c5fc3d
Signed by: school
GPG key ID: 8CFC3837052871B4
1 changed files with 54 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

60
src/api/events.py
View file

@ -2,7 +2,7 @@ from datetime import date, datetime, time
from typing import Annotated, final
from beanie import DeleteRules, Link, PydanticObjectId
from fastapi import APIRouter, Body, HTTPException, status
from fastapi import APIRouter, Body, HTTPException, Query, status
from fastapi.responses import Response
from pydantic import BaseModel, StringConstraints, field_validator
from pydantic_extra_types.color import Color
@ -160,20 +160,55 @@ class PartialEventUpdateData(_BaseEventData):
@base_router.get("/users/{user_id}/events")
async def get_user_events(user_id: PydanticObjectId, user: CurrentUserDep) -> list[EventData]:
"""Get all events that belong to given user."""
async def get_user_events(
user_id: PydanticObjectId,
user: CurrentUserDep,
start_date_from: Annotated[date | None, Query()] = None,
start_date_to: Annotated[date | None, Query()] = None,
end_date_from: Annotated[date | None, Query()] = None,
end_date_to: Annotated[date | None, Query()] = None,
) -> list[EventData]:
"""Get all events that belong to given user.
Optionally, it's possible to use query params to filter the events
based on the date of the event. The dates are specified in ISO 8601 format.
E.g.: /users/{user_id}/events?start_date_from=2025-01-01&start_date_to=2025-01-10
"""
if user_id != user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only access your own categories",
)
events = await (Event.find(expr(Event.user).id == user.id, fetch_links=True)).to_list()
# Initial query (all events for the user)
query = Event.find(expr(Event.user).id == user.id)
# Filter by date
if start_date_from is not None:
query = query.find(expr(Event.start_date) >= start_date_from)
if start_date_to is not None:
query = query.find(expr(Event.start_date) <= start_date_to)
if end_date_from is not None:
query = query.find(expr(Event.end_date) >= end_date_from)
if end_date_to is not None:
query = query.find(expr(Event.end_date) <= end_date_to)
query = query.find(fetch_links=True)
events = await query.to_list()
return [EventData.from_event(event) for event in events]
@base_router.get("/users/{user_id}/events/invited")
async def get_user_invited_events(user_id: PydanticObjectId, user: CurrentUserDep) -> list[EventData]:
async def get_user_invited_events(
user_id: PydanticObjectId,
user: CurrentUserDep,
start_date_from: Annotated[date | None, Query()] = None,
start_date_to: Annotated[date | None, Query()] = None,
end_date_from: Annotated[date | None, Query()] = None,
end_date_to: Annotated[date | None, Query()] = None,
) -> list[EventData]:
"""Get all events that given user was invited to.
These are the events that the user has already accepted the invite for.
@ -185,7 +220,20 @@ async def get_user_invited_events(user_id: PydanticObjectId, user: CurrentUserDe
detail="You can only access your own invite categories",
)
events = await (Event.find(expr(Event.attendees).id == user_id, fetch_links=True)).to_list()
# Initial query (all events the user is attending)
query = Event.find(expr(Event.attendees).id == user_id, fetch_links=True)
# Filter by date
if start_date_from is not None:
query = query.find(expr(Event.start_date) >= start_date_from)
if start_date_to is not None:
query = query.find(expr(Event.start_date) <= start_date_to)
if end_date_from is not None:
query = query.find(expr(Event.end_date) >= end_date_from)
if end_date_to is not None:
query = query.find(expr(Event.end_date) <= end_date_to)
events = await query.to_list()
return [EventData.from_event(event) for event in events]