add route + test (#113)

* add route + test

* black

* add repeated insert

* black

Author: parfenovma

calendar_backend/exceptions.py

@@ -2,8 +2,11 @@
 
 
 class ObjectNotFound(Exception):
-    def __init__(self, type: Type, ids: int | list[int]):
-        super().__init__(f"Objects of type {type.__name__} {ids=} not found")
+    def __init__(self, type: Type, ids: int | list[int] = [], name: str | None = None):
+        msg = f"Objects of type {type.__name__} {ids=} not found"
+        if name:
+            msg = f"Objects of type {type.__name__} with {name=} not found"
+        super().__init__(msg)
 
 
 class NotEnoughCriteria(Exception):

calendar_backend/models/db.py

@@ -1,5 +1,4 @@
-"""Database common classes and methods
-"""
+"""Database common classes and methods"""
 
 from __future__ import annotations
 

calendar_backend/routes/event/event.py

@@ -1,18 +1,25 @@
 import logging
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta
 from typing import Literal
 
 from auth_lib.fastapi import UnionAuth
-from fastapi import APIRouter, Depends, Query
-from fastapi.responses import FileResponse
+from fastapi import APIRouter, Depends, Query, status
+from fastapi.responses import FileResponse, JSONResponse
 from fastapi_sqlalchemy import db
 from pydantic import TypeAdapter
 
 from calendar_backend.exceptions import NotEnoughCriteria
 from calendar_backend.methods import list_calendar
 from calendar_backend.models import Event, Group, Lecturer, Room
 from calendar_backend.routes.models import EventGet
-from calendar_backend.routes.models.event import EventPatch, EventPost, GetListEvent
+from calendar_backend.routes.models.event import (
+    EventPatch,
+    EventPatchName,
+    EventPatchResult,
+    EventPost,
+    EventRepeatedPost,
+    GetListEvent,
+)
 from calendar_backend.settings import get_settings
 
 
@@ -98,6 +105,44 @@ async def create_event(event: EventPost, _=Depends(UnionAuth(scopes=["timetable.
     return EventGet.model_validate(event_get)
 
 
+@router.post("/repeating", response_model=list[EventGet])
+async def create_repeating_event(
+    event: EventRepeatedPost,  # _=Depends(UnionAuth(scopes=["timetable.event.create"]))
+) -> list[EventGet]:
+    if event.repeat_timedelta_days <= 0:
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST, content={"detail": f"Timedelta must be a positive integer"}
+        )
+    if event.repeat_until_ts > event.start_ts + timedelta(days=1095):
+        return JSONResponse(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            content={"detail": "Due to disk utilization limits, events with duration > 3 years is restricted"},
+        )
+    events = []
+    event_dict = event.model_dump()
+    rooms = [Room.get(room_id, session=db.session) for room_id in event_dict.pop("room_id", [])]
+    lecturers = [Lecturer.get(lecturer_id, session=db.session) for lecturer_id in event_dict.pop("lecturer_id", [])]
+    groups = [Group.get(group_id, session=db.session) for group_id in event_dict.pop("group_id", [])]
+    repeat_timedelta_days = timedelta(days=event.repeat_timedelta_days)
+    cur_start_ts = event_dict["start_ts"]
+    cur_end_ts = event_dict["end_ts"]
+    while cur_start_ts <= event.repeat_until_ts:
+        event_get = Event.create(
+            name=event_dict["name"],
+            start_ts=cur_start_ts,
+            end_ts=cur_end_ts,
+            room=rooms,
+            lecturer=lecturers,
+            group=groups,
+            session=db.session,
+        )
+        events.append(event_get)
+        cur_start_ts += repeat_timedelta_days
+        cur_end_ts += repeat_timedelta_days
+    adapter = TypeAdapter(list[EventGet])
+    return adapter.validate_python(events)
+
+
 @router.post("/bulk", response_model=list[EventGet])
 async def create_events(
     events: list[EventPost], _=Depends(UnionAuth(scopes=["timetable.event.create"]))
@@ -139,6 +184,17 @@ async def create_events(
     return adapter.validate_python(result)
 
 
+@router.patch("/patch_name", response_model=EventPatchResult, summary="Batch update events by name")
+async def patch_event_by_name(
+    event_inp: EventPatchName, _=Depends(UnionAuth(scopes=["timetable.event.update"]))
+) -> EventPatchResult:
+    updated = (
+        db.session.query(Event).filter(Event.name == event_inp.old_name).update(values={"name": event_inp.new_name})
+    )
+    db.session.commit()
+    return EventPatchResult(old_name=event_inp.old_name, new_name=event_inp.new_name, updated=updated)
+
+
 @router.patch("/{id}", response_model=EventGet)
 async def patch_event(
     id: int, event_inp: EventPatch, _=Depends(UnionAuth(scopes=["timetable.event.update"]))

calendar_backend/routes/models/event.py

@@ -19,6 +19,17 @@ def __repr__(self):
         )
 
 
+class EventPatchName(Base):
+    old_name: str
+    new_name: str
+
+
+class EventPatchResult(Base):
+    old_name: str
+    new_name: str
+    updated: int
+
+
 class EventPost(Base):
     name: str
     room_id: list[int]
@@ -35,6 +46,25 @@ def __repr__(self):
         )
 
 
+class EventRepeatedPost(Base):
+    name: str
+    room_id: list[int]
+    group_id: list[int]
+    lecturer_id: list[int]
+    start_ts: datetime.datetime
+    end_ts: datetime.datetime
+    repeat_timedelta_days: int = 7  # set one week by default
+    repeat_until_ts: datetime.datetime
+
+    def __repr__(self):
+        return (
+            f"Lesson(name={self.name},\n"
+            f" room={self.room_id}, group={self.group_id},\n"
+            f" lecturer={self.lecturer_id}, start_ts={self.start_ts}, end_ts={self.end_ts})\n"
+            f" repeats every {self.repeat_timedelta_days} days until {repeat_until_ts}\n"
+        )
+
+
 class Event(Base):
     id: int
     name: str

tests/event/event.py

@@ -331,3 +331,72 @@ def test_delete_from_to(client_auth: TestClient, dbsession: Session, room_factor
     for row in (obj1, obj2):
         dbsession.delete(row)
     dbsession.commit()
+
+
+def test_update_by_name(client_auth: TestClient, dbsession: Session, room_factory, group_factory, lecturer_factory):
+    room_path1 = room_factory(client_auth)
+    group_path1 = group_factory(client_auth)
+    lecturer_path1 = lecturer_factory(client_auth)
+    room_path2 = room_factory(client_auth)
+    group_path2 = group_factory(client_auth)
+    lecturer_path2 = lecturer_factory(client_auth)
+    room_id1 = int(room_path1.split("/")[-1])
+    group_id1 = int(group_path1.split("/")[-1])
+    lecturer_id1 = int(lecturer_path1.split("/")[-1])
+    room_id2 = int(room_path2.split("/")[-1])
+    group_id2 = int(group_path2.split("/")[-1])
+    lecturer_id2 = int(lecturer_path2.split("/")[-1])
+    request_obj = [
+        {
+            "name": "string",
+            "room_id": [room_id1],
+            "group_id": [group_id1],
+            "lecturer_id": [lecturer_id1],
+            "start_ts": "2022-08-26T22:32:38.575Z",
+            "end_ts": "2022-08-26T22:32:38.575Z",
+        },
+        {
+            "name": "string",
+            "room_id": [room_id2],
+            "group_id": [group_id2],
+            "lecturer_id": [lecturer_id2],
+            "start_ts": "2022-08-26T22:32:38.575Z",
+            "end_ts": "2022-08-26T22:32:38.575Z",
+        },
+    ]
+    response = client_auth.post(f"{RESOURCE}bulk", json=request_obj)
+    created = response.json()
+    assert response.status_code == status.HTTP_200_OK, response.json()
+    name_to_patch = "not_existing_name"
+    response = client_auth.patch(
+        f"{RESOURCE}patch_name", json={"old_name": "not_existing_name", "new_name": "some_name"}
+    )
+    assert response.status_code == status.HTTP_200_OK, response.json()
+    assert response.json()["updated"] == 0  # no events w name "not_existing_name"
+    response = client_auth.patch(f"{RESOURCE}patch_name", json={"old_name": "string", "new_name": "some_name"})
+    assert response.status_code == status.HTTP_200_OK, response.json()
+    assert response.json()["updated"] > 0  # at least 2 events w name "string" (due to our post request)
+
+
+def test_create_repeated_events(
+    client_auth: TestClient, dbsession: Session, room_factory, group_factory, lecturer_factory
+):
+    room_path1 = room_factory(client_auth)
+    group_path1 = group_factory(client_auth)
+    lecturer_path1 = lecturer_factory(client_auth)
+    room_id1 = int(room_path1.split("/")[-1])
+    group_id1 = int(group_path1.split("/")[-1])
+    lecturer_id1 = int(lecturer_path1.split("/")[-1])
+    request_obj = {
+        "name": "string",
+        "room_id": [room_id1],
+        "group_id": [group_id1],
+        "lecturer_id": [lecturer_id1],
+        "start_ts": "2022-08-26T22:32:38.575Z",
+        "end_ts": "2022-08-26T22:32:38.575Z",
+        "repeat_timedelta_days": 7,
+        "repeat_until_ts": "2023-08-26T22:32:38.575Z",
+    }
+    response = client_auth.post(f"{RESOURCE}repeating", json=request_obj)
+    created = response.json()
+    assert response.status_code == status.HTTP_200_OK, response.json()