didier/database/crud/deadlines.py

42 lines
1.4 KiB
Python
Raw Normal View History

import datetime
2022-08-13 00:41:47 +02:00
from typing import Optional
2022-08-13 00:07:48 +02:00
from zoneinfo import ZoneInfo
from dateutil.parser import parse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
2022-08-29 20:24:42 +02:00
from database.schemas import Deadline, UforaCourse
2022-08-13 00:07:48 +02:00
__all__ = ["add_deadline", "get_deadlines"]
async def add_deadline(session: AsyncSession, course_id: int, name: str, date_str: str):
"""Add a new deadline"""
date_dt = parse(date_str, dayfirst=True).replace(tzinfo=ZoneInfo("Europe/Brussels"))
2022-08-13 00:07:48 +02:00
2022-08-13 00:41:47 +02:00
# If we only have a day, assume it's the end of the day
2022-08-13 00:07:48 +02:00
if date_dt.hour == date_dt.minute == date_dt.second == 0:
date_dt.replace(hour=23, minute=59, second=59)
deadline = Deadline(course_id=course_id, name=name, deadline=date_dt)
session.add(deadline)
await session.commit()
async def get_deadlines(
session: AsyncSession, *, after: Optional[datetime.date] = None, course: Optional[UforaCourse] = None
) -> list[Deadline]:
"""Get a list of all upcoming deadlines"""
2022-08-13 00:41:47 +02:00
statement = select(Deadline)
if after is not None:
statement = statement.where(Deadline.deadline > after)
2022-08-13 00:41:47 +02:00
if course is not None:
statement = statement.where(Deadline.course_id == course.course_id)
statement = statement.options(selectinload(Deadline.course))
2022-08-13 00:07:48 +02:00
return (await session.execute(statement)).scalars().all()