| from datetime import datetime |
| from cbh.api.account.models import AccountModel, AccountShorten |
| from cbh.api.common.db_requests import get_all_objs |
| from cbh.api.common.schemas import FilterRequest |
| from cbh.api.events.dto import EventType |
| from cbh.api.events.models import EventModel |
| from cbh.api.events.schemas import CreateEventRequest, EventFilter |
| from cbh.core.config import settings |
|
|
|
|
| async def create_event_obj( |
| event: CreateEventRequest, account: AccountModel |
| ) -> EventModel: |
| """ |
| Create a new event. |
| """ |
| event = EventModel( |
| reason=event.reason, |
| type=EventType.CUSTOM, |
| startDate=event.startDate, |
| endDate=event.endDate, |
| coach=AccountShorten(**account.model_dump()), |
| ) |
| await settings.DB_CLIENT.events.insert_one(event.to_mongo()) |
| return event |
|
|
|
|
| async def filter_events_objs( |
| request: FilterRequest[EventFilter], |
| account: AccountModel, |
| ) -> tuple[list[EventModel], int]: |
| filters = {"isActive": True, "coach.id": account.id, "type": EventType.CUSTOM.value} |
| if request.filter.startDate: |
| start_dt = datetime.combine( |
| request.filter.startDate, datetime.min.time() |
| ).isoformat() |
| filters["startDate"] = {"$gte": start_dt} |
| if request.filter.endDate: |
| end_dt = datetime.combine( |
| request.filter.endDate, datetime.max.time() |
| ).isoformat() |
| filters["endDate"] = {"$lte": end_dt} |
| if request.filter.searchTerm: |
| filters["$or"] = [ |
| {"reason": {"$regex": request.filter.searchTerm, "$options": "i"}}, |
| ] |
| sorting = ("_id", -1) |
| if request.sortBy: |
| sorting = (request.sortBy.name, request.sortBy.order.value) |
| events, total_count = await get_all_objs( |
| EventModel, |
| request.pageSize, |
| request.pageIndex, |
| additional_filter=filters, |
| sort=sorting, |
| ) |
| return events, total_count |
|
|
