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

calendar_backend/methods/list_calendar.py

@@ -4,14 +4,16 @@
 import time
 from datetime import date as date_
 from datetime import datetime
+from typing import 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 sqlalchemy.orm import Session
 
-from calendar_backend.models import Group
+from calendar_backend.models import Event as DB_Event
+from calendar_backend.models import Group, Lecturer, Room
 from calendar_backend.settings import get_settings
 
 from . import utils
@@ -21,6 +23,76 @@
 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(dbsession: Session, file: UploadFile = File(...)) -> List:
+    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 = []
+    result = []
+    for element in cal_obj.walk():
+        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 обязателен
+        if not data.get("group_ids"):
+            raise HTTPException(status_code=403, detail="Невозможно создать событие без группы!")
+        events.append(data)
+
+    for data in events:
+        existing_events_query = (
+            DB_Event.get_all(session=dbsession)
+            .filter(DB_Event.name == data.get("name"))
+            .filter(DB_Event.start_ts == data.get("start_ts"))
+            .filter(DB_Event.end_ts == data.get("end_ts"))
+        )
+        is_unique = True
+        for existing_event in existing_events_query.all():
+            if (
+                {column.id for column in existing_event.group} == set(data["group_id"])
+                and {column.id for column in existing_event.room} == set(data["room_id"])
+                and {column.id for column in existing_event.lecturer} == set(data["lecturer_id"])
+            ):
+                is_unique = False
+
+        if is_unique:
+            rooms = [Room.get(room_id, session=dbsession) for room_id in data.pop("room_id", [])]
+            lecturers = [Lecturer.get(lecturer_id, session=dbsession) for lecturer_id in data.pop("lecturer_id", [])]
+            groups = [Group.get(group_id, session=dbsession) for group_id in data.pop("group_id", [])]
+            result.append(
+                DB_Event.create(
+                    **data,
+                    room=rooms,
+                    lecturer=lecturers,
+                    group=groups,
+                    session=dbsession,
+                )
+            )
+    dbsession.commit()
+    return result
+
+
 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,7 +3,7 @@
 from typing import Literal
 
 from auth_lib.fastapi import UnionAuth
-from fastapi import APIRouter, Depends, Query, status
+from fastapi import APIRouter, Depends, File, Query, UploadFile, status
 from fastapi.responses import FileResponse, JSONResponse
 from fastapi_sqlalchemy import db
 from pydantic import TypeAdapter
@@ -33,6 +33,13 @@ async def get_event_by_id(id: int) -> EventGet:
     return EventGet.model_validate(Event.get(id, session=db.session))
 
 
+@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(db.session, file)
+    return [EventGet.model_validate(event) for event in events]
+
+
 async def _get_timetable(start: date, end: date, group_id, lecturer_id, room_id, detail, limit, offset):
     if bool(group_id) + bool(lecturer_id) + bool(room_id) != 1:
         raise NotEnoughCriteria("Exactly one argument group_id, lecturer_id or room_id required")

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"}],