77 lines
3.2 KiB
Python
77 lines
3.2 KiB
Python
# modules/calendar/service.py
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_ # Import or_
|
|
from datetime import datetime
|
|
from modules.calendar.models import CalendarEvent
|
|
from core.exceptions import not_found_exception
|
|
from modules.calendar.schemas import CalendarEventCreate, CalendarEventUpdate # Import schemas
|
|
|
|
def create_calendar_event(db: Session, user_id: int, event_data: CalendarEventCreate):
|
|
# Ensure tags is None if not provided or empty list, matching model
|
|
tags_to_store = event_data.tags if event_data.tags else None
|
|
event = CalendarEvent(
|
|
**event_data.model_dump(exclude={'tags'}), # Use model_dump and exclude tags initially
|
|
tags=tags_to_store, # Set tags separately
|
|
user_id=user_id
|
|
)
|
|
db.add(event)
|
|
db.commit()
|
|
db.refresh(event)
|
|
return event
|
|
|
|
def get_calendar_events(db: Session, user_id: int, start: datetime | None, end: datetime | None):
|
|
query = db.query(CalendarEvent).filter(CalendarEvent.user_id == user_id)
|
|
|
|
# If start and end dates are provided, filter for events overlapping the range.
|
|
# An event overlaps if: event_start < query_end AND (event_end IS NULL OR event_end > query_start)
|
|
if start and end:
|
|
query = query.filter(
|
|
CalendarEvent.start < end, # Event starts before the query window ends
|
|
or_(
|
|
CalendarEvent.end == None, # Event has no end date (considered single point in time at start)
|
|
CalendarEvent.end > start # Event ends after the query window starts
|
|
)
|
|
)
|
|
# If only start is provided, filter events starting on or after start
|
|
elif start:
|
|
query = query.filter(CalendarEvent.start >= start)
|
|
# If only end is provided, filter events ending on or before end (or starting before end if no end date)
|
|
elif end:
|
|
query = query.filter(
|
|
or_(
|
|
CalendarEvent.end <= end,
|
|
(CalendarEvent.end == None and CalendarEvent.start < end)
|
|
)
|
|
)
|
|
|
|
return query.order_by(CalendarEvent.start).all() # Order by start time
|
|
|
|
def get_calendar_event_by_id(db: Session, user_id: int, event_id: int):
|
|
event = db.query(CalendarEvent).filter(
|
|
CalendarEvent.id == event_id,
|
|
CalendarEvent.user_id == user_id
|
|
).first()
|
|
if not event:
|
|
raise not_found_exception()
|
|
return event
|
|
|
|
def update_calendar_event(db: Session, user_id: int, event_id: int, event_data: CalendarEventUpdate):
|
|
event = get_calendar_event_by_id(db, user_id, event_id) # Reuse get_by_id for check
|
|
# Use model_dump with exclude_unset=True to only update provided fields
|
|
update_data = event_data.model_dump(exclude_unset=True)
|
|
|
|
for key, value in update_data.items():
|
|
# Ensure tags is handled correctly (set to None if empty list provided)
|
|
if key == 'tags' and isinstance(value, list) and not value:
|
|
setattr(event, key, None)
|
|
else:
|
|
setattr(event, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(event)
|
|
return event
|
|
|
|
def delete_calendar_event(db: Session, user_id: int, event_id: int):
|
|
event = get_calendar_event_by_id(db, user_id, event_id) # Reuse get_by_id for check
|
|
db.delete(event)
|
|
db.commit() |