Добавлена ручка /event/ical для получения событий из 3'd party источников

calendar_backend/methods/list_calendar.py

@@ -4,15 +4,19 @@
 import time
 from datetime import date as date_
 from datetime import datetime
+from typing import Dict, List
 
 import pytz
-from fastapi import HTTPException
+from fastapi import File, HTTPException, UploadFile
 from fastapi.responses import FileResponse
 from icalendar import Calendar, Event, vText
+from pydantic import ValidationError
 from sqlalchemy.orm import Session
 
 from calendar_backend.models import Group
+from calendar_backend.routes.models.event import EventRepeatedPost
 from calendar_backend.settings import get_settings
+from calendar_backend.utils.services import EventService
 
 from . import utils
 
@@ -21,6 +25,61 @@
 logger = logging.getLogger(__name__)
 
 
+def _get_list_from_ical_obj(element, field: str) -> List:
+    items = element.get(field)
+    if not items:
+        return []
+    elif isinstance(items, list):
+        return [int(i) for i in items]
+    elif "," in (str_items := str(items)):
+        return [int(i) for i in items.split(",")]
+    else:
+        return [int(items)]
+
+
+async def create_event_from_icalendar(file: UploadFile = File(...)) -> List[Dict]:
+    extension = file.filename.split(".")[-1]
+    available_exts = ["ical", "ics"]
+    if extension not in available_exts:
+        raise HTTPException(status_code=422, detail="Не поддерживаемый фармат файла!")
+
+    raw_file: bytes = file.read()
+    str_file: str = raw_file.decode("utf-8")
+    cal_obj = Calendar.from_ical(str_file)
+    events = []
+    for element in cal_obj.walk("VEVENT"):
+        data = {}
+        data["name"] = element.get("summary")
+        data["start_ts"] = element.get("dtstart")
+        data["end_ts"] = element.get("dtend")
+        group_ids_field, lecturer_ids_field, room_ids_field = "X-FF-GROUP-IDS", "X-FF-LECTURER-IDS", "X-FF-ROOM-IDS"
+        data["group_ids"] = _get_list_from_ical_obj(element, group_ids_field)
+        data["lecturer_ids"] = _get_list_from_ical_obj(element, lecturer_ids_field)
+        data["room_ids"] = _get_list_from_ical_obj(element, room_ids_field)
+        # решено, что group_id обязателен
+        try:
+            if not data.get("group_ids"):
+                raise HTTPException(status_code=403, detail="Невозможно создать событие без группы!")
+
+            if rrule := element.get("rrule"):
+                interval = rrule.get("interval", [None])
+                until = rrule.get("until", [None])
+                data["repeat_timedelta_days"] = interval[0]
+                data["repeat_until_ts"] = until[0]
+
+                event = EventRepeatedPost.model_validate(data)
+                repeating_events = await EventService.reproduce_repeating_event(event)
+                events.extend(repeating_events)
+            else:
+                events.append(data)
+
+        except (HTTPException, ValidationError):
+            # вероятно следует логировать какие именно события не удалось получить
+            break
+
+    return events
+
+
 async def get_user_calendar(group_id: int, session: Session, start_date: date_, end_date: date_) -> Calendar:
     """
     Returns event iCalendar object

calendar_backend/routes/event/event.py

@@ -3,8 +3,8 @@
 from typing import Literal
 
 from auth_lib.fastapi import UnionAuth
-from fastapi import APIRouter, Depends, Query, status
-from fastapi.responses import FileResponse, JSONResponse
+from fastapi import APIRouter, Depends, File, Query, UploadFile
+from fastapi.responses import FileResponse
 from fastapi_sqlalchemy import db
 from pydantic import TypeAdapter
 
@@ -21,6 +21,7 @@
     GetListEvent,
 )
 from calendar_backend.settings import get_settings
+from calendar_backend.utils.services import EventService
 
 
 settings = get_settings()
@@ -109,77 +110,29 @@ async def create_event(event: EventPost, _=Depends(UnionAuth(scopes=["timetable.
 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
+    """Создает множество повторяющихся событий."""
+    list_events = await EventService.reproduce_repeating_event(event)
+    result = await EventService.bulk_create_events(db.session, list_events)
     adapter = TypeAdapter(list[EventGet])
-    return adapter.validate_python(events)
+    return adapter.validate_python(result)
 
 
 @router.post("/bulk", response_model=list[EventGet])
 async def create_events(
     events: list[EventPost], _=Depends(UnionAuth(scopes=["timetable.event.create"]))
 ) -> list[EventGet]:
-    result = []
-    for event in events:
-        event_dict = event.model_dump()
-        existing_events_query = (
-            Event.get_all(session=db.session)
-            .filter(Event.name == event_dict.get("name"))
-            .filter(Event.start_ts == event_dict.get("start_ts"))
-            .filter(Event.end_ts == event_dict.get("end_ts"))
-        )
-        is_unique = True
-        for existing_event in existing_events_query.all():
-            if (
-                {column.id for column in existing_event.group} == set(event_dict["group_id"])
-                and {column.id for column in existing_event.room} == set(event_dict["room_id"])
-                and {column.id for column in existing_event.lecturer} == set(event_dict["lecturer_id"])
-            ):
-                is_unique = False
-        if is_unique:
-            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", [])]
-            result.append(
-                Event.create(
-                    **event_dict,
-                    room=rooms,
-                    lecturer=lecturers,
-                    group=groups,
-                    session=db.session,
-                )
-            )
-    db.session.commit()
+    """Создает множество событий."""
+    list_events = [event.model_dump() for event in events]
+    result = await EventService.bulk_create_events(db.session, list_events)
+    adapter = TypeAdapter(list[EventGet])
+    return adapter.validate_python(result)
+
+
+@router.post("/ical", response_model=EventPost)
+async def create_events_from_3dparty(file: UploadFile = File(...)):
+    """Создает события на основе .ical или .ics файлов полученных из 3'd party источников."""
+    events = await list_calendar.create_event_from_icalendar(file)
+    result = await EventService.bulk_create_events(db.session, events)
     adapter = TypeAdapter(list[EventGet])
     return adapter.validate_python(result)
 

calendar_backend/utils/services.py

@@ -0,0 +1,82 @@
+from datetime import timedelta
+from typing import Dict, List
+
+from fastapi import HTTPException, status
+from sqlalchemy.orm import Session
+
+from calendar_backend.models import Event, Group, Lecturer, Room
+from calendar_backend.routes.models.event import EventRepeatedPost
+
+
+class EventService:
+    """Сервис для работы с логикой создания событий"""
+
+    @classmethod
+    async def bulk_create_events(cls, dbsession: Session, events: List[Dict]) -> List[Event]:
+        result = []
+        for event_dict in events:
+            existing_events_query = (
+                Event.get_all(session=dbsession)
+                .filter(Event.name == event_dict.get("name"))
+                .filter(Event.start_ts == event_dict.get("start_ts"))
+                .filter(Event.end_ts == event_dict.get("end_ts"))
+            )
+            is_unique = True
+            for existing_event in existing_events_query.all():
+                if (
+                    {column.id for column in existing_event.group} == set(event_dict["group_id"])
+                    and {column.id for column in existing_event.room} == set(event_dict["room_id"])
+                    and {column.id for column in existing_event.lecturer} == set(event_dict["lecturer_id"])
+                ):
+                    is_unique = False
+            if is_unique:
+                rooms = [Room.get(room_id, session=dbsession) for room_id in event_dict.pop("room_id", [])]
+                lecturers = [
+                    Lecturer.get(lecturer_id, session=dbsession) for lecturer_id in event_dict.pop("lecturer_id", [])
+                ]
+                groups = [Group.get(group_id, session=dbsession) for group_id in event_dict.pop("group_id", [])]
+                result.append(
+                    Event.create(
+                        **event_dict,
+                        room=rooms,
+                        lecturer=lecturers,
+                        group=groups,
+                        session=dbsession,
+                    )
+                )
+        dbsession.commit()
+        return result
+
+    @classmethod
+    async def reproduce_repeating_event(cls, event: EventRepeatedPost) -> List[Dict]:
+        if event.repeat_timedelta_days <= 0:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Timedelta must be a positive integer")
+        if event.repeat_until_ts > event.start_ts + timedelta(days=1095):
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Due to disk utilization limits, events with duration > 3 years is restricted",
+            )
+        events = []
+        event_dict = event.model_dump()
+
+        rooms = [room_id for room_id in event_dict.pop("room_id", [])]
+        lecturers = [lecturer_id for lecturer_id in event_dict.pop("lecturer_id", [])]
+        groups = [group_id 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_get["name"] = event_dict["name"]
+            event_get["start_ts"] = cur_start_ts
+            event_get["end_ts"] = cur_end_ts
+            event_get["room_id"] = rooms
+            event_get["lecturer_id"] = lecturers
+            event_get["group_id"] = groups
+            events.append(event_get)
+            cur_start_ts += repeat_timedelta_days
+            cur_end_ts += repeat_timedelta_days
+
+        return events

tests/conftest.py

@@ -21,7 +21,7 @@ def client():
 
 @pytest.fixture()
 def client_auth(mocker: MockerFixture):
-    user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__')
+    user_mock = mocker.patch('auth_lib.fastapi.UnionAuth.__call__', autospec=True)
     user_mock.return_value = {
         "session_scopes": [{"id": 0, "name": "string", "comment": "string"}],
         "user_scopes": [{"id": 0, "name": "string", "comment": "string"}],