Event-CRUD-Flask-python3-API/services/EventService.py

83 lines
2.7 KiB
Python
Raw Normal View History

from sqlalchemy.sql import exists
2024-01-08 19:12:29 +00:00
from models import db, Event, UserEventAssociation, User
from services.UserService import UserService
from datetime import datetime
from flask import g
2024-01-08 13:28:00 +00:00
from sqlalchemy import func
2024-01-08 20:03:51 +00:00
from datetime import timedelta
class EventService:
@staticmethod
def create_event(data):
new_event = Event(
title=data['title'],
description=data.get('description', ''),
location=data.get('location', ''),
2024-01-07 14:21:35 +00:00
duedate=datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S'),
user_id=data['user_id']
)
db.session.add(new_event)
db.session.commit()
return new_event
@staticmethod
2024-01-08 13:28:00 +00:00
def get_upcoming_events(location=None, sort_by=None):
query = Event.query.filter(
2024-01-10 07:37:05 +00:00
Event.deleted == False
2024-01-08 13:28:00 +00:00
)
if location:
query = query.filter(Event.location.ilike(f"%{location}%"))
if sort_by == 'date':
query = query.order_by(Event.duedate)
elif sort_by == 'popularity':
2024-01-08 19:12:29 +00:00
query = query.join(UserEventAssociation, Event.id == UserEventAssociation.event_id)\
.group_by(Event.id)\
.order_by(func.count(UserEventAssociation.user_id).desc())
2024-01-08 13:28:00 +00:00
elif sort_by == 'creation':
query = query.order_by(Event.created_at)
events = query.all()
return [event.to_dict() for event in events]
@staticmethod
2024-01-07 14:21:35 +00:00
def get_event_by_id(event_id, user_id):
event = Event.query.filter_by(id=event_id,user_id=user_id, deleted=False).first()
if(event):
return event.to_dict()
else:
return []
2024-01-07 11:28:49 +00:00
@staticmethod
def update_event(event_id, data):
2024-01-08 18:15:49 +00:00
event = Event.query.filter_by(user_id=g.user_id, id=event_id).first()
2024-01-07 14:32:08 +00:00
if not event:
return None
event.title = data['title']
event.description = data.get('description', '')
event.location = data.get('location', '')
event.time = datetime.strptime(data['duedate'], '%Y-%m-%dT%H:%M:%S')
db.session.commit()
return event
@staticmethod
def delete_event(event_id):
2024-01-08 18:15:49 +00:00
event = Event.query.filter_by(id=event_id, user_id=g.user_id, deleted=False).first()
if event:
event.deleted = True
db.session.commit()
else:
return None
return event
2024-01-08 20:03:51 +00:00
@staticmethod
def get_all_upcomming_events():
2024-01-10 07:37:05 +00:00
upcoming_deadline = datetime.now() + timedelta(minutes=30)
2024-01-08 20:03:51 +00:00
events = Event.query.filter(
2024-01-10 07:37:05 +00:00
Event.duedate <= upcoming_deadline,
2024-01-08 20:03:51 +00:00
Event.deleted == False
).all()
return [event.to_dict() for event in events]