Add basic update event

app/routers/event.py

@@ -1,12 +1,13 @@
+from datetime import datetime
 from operator import attrgetter
-from typing import List
+from typing import Dict, List, Optional, Any
 
-from fastapi import APIRouter, Request
-
-from app.database.models import Event
-from app.database.models import UserEvent
+from app.database.models import Event, UserEvent
 from app.dependencies import templates
 from app.internal.utils import create_model
+from fastapi import APIRouter, Request
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
 
 router = APIRouter(
     prefix="/event",
@@ -27,6 +28,57 @@ async def eventview(request: Request, id: int):
                                       {"request": request, "event_id": id})
 
 
+def by_id(db: Session, event_id: int) -> Event:
+    """Select event by id"""
+
+    return db.query(Event).filter(Event.id == event_id).first()
+
+
+def is_date_before(start_date: datetime, end_date: datetime) -> bool:
+    """Check if the start date is earlier than the end date"""
+
+    return start_date < end_date
+
+
+def is_it_possible_to_change_dates(
+        db: Session, old_event: Event, event: Dict[str, Any]) -> bool:
+    return is_date_before(
+        event.get('start', old_event.start),
+        event.get('end', old_event.end))
+
+
+def get_items_that_can_be_updated(event: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract only that keys to update"""
+
+    return {i: event[i] for i in (
+        'title', 'start', 'end', 'content', 'location') if i in event}
+
+
+def update_event(event_id: int, event: Dict, db: Session
+                 ) -> Optional[Event]:
+
+    # TODO Check if the user is the owner of the event.
+
+    event_to_update = get_items_that_can_be_updated(event)
+    if not event_to_update:
+        return None
+    try:
+        old_event = by_id(db=db, event_id=event_id)
+        if old_event is None or not is_it_possible_to_change_dates(
+                db, old_event, event_to_update):
+            return None
+
+        # Update database
+        db.query(Event).filter(Event.id == event_id).update(
+            event_to_update, synchronize_session=False)
+        db.commit()
+
+        # TODO: Send emails to recipients.
+    except (AttributeError, SQLAlchemyError, TypeError):
+        return None
+    return by_id(db=db, event_id=event_id)
+
+
 def create_event(db, title, start, end, owner_id, content=None, location=None):
     """Creates an event and an association."""
 

app/routers/profile.py

@@ -43,8 +43,6 @@ async def profile(
         session.commit()
         user = session.query(User).filter_by(id=1).first()
 
-    session.close()
-
     return templates.TemplateResponse("profile.html", {
         "request": request,
         "user": user,
@@ -64,8 +62,6 @@ async def update_user_fullname(
     user.full_name = new_fullname
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
     response = RedirectResponse(url=url, status_code=HTTP_302_FOUND)
     return response
@@ -83,8 +79,6 @@ async def update_user_email(
     user.email = new_email
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
     return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 
@@ -101,8 +95,6 @@ async def update_profile(
     user.description = new_description
     session.commit()
 
-    session.close()
-
     url = router.url_path_for("profile")
     return RedirectResponse(url=url, status_code=HTTP_302_FOUND)
 

requirements.txt

@@ -38,6 +38,8 @@ iniconfig==1.1.1
 Jinja2==2.11.2
 joblib==1.0.0
 lazy-object-proxy==1.5.2
+mypy==0.790
+mypy-extensions==0.4.3
 MarkupSafe==1.1.1
 nltk==3.5
 packaging==20.8
@@ -68,11 +70,13 @@ smtpdfix==0.2.6
 sniffio==1.2.0
 sortedcontainers==2.3.0
 soupsieve==2.1
+sortedcontainers==2.3.0
 SQLAlchemy==1.3.22
 starlette==0.13.6
 text-unidecode==1.3
 toml==0.10.2
 tqdm==4.56.0
+typed-ast==1.4.2
 typing-extensions==3.7.4.3
 urllib3==1.26.2
 uvicorn==0.13.3

tests/test_event.py

@@ -1,5 +1,16 @@
+from datetime import datetime
+
+import pytest
+from app.database.models import Event
+from app.routers.event import update_event
 from starlette.status import HTTP_404_NOT_FOUND
 
+INVALID_UPDATE_OPTIONS = [
+    {}, {"test": "test"}, {"start": "20.01.2020"},
+    {"start": datetime(2020, 2, 2), "end": datetime(2020, 1, 1)},
+    {"start": datetime(2030, 2, 2)}, {"end": datetime(1990, 1, 1)},
+]
+
 
 class TestEvent:
 
@@ -17,5 +28,36 @@ def test_eventview_without_id(self, client):
         response = client.get("/event/view")
         assert response.status_code == HTTP_404_NOT_FOUND
 
+    @staticmethod
+    @pytest.mark.parametrize("data", INVALID_UPDATE_OPTIONS)
+    def test_invalid_update(event, data, session):
+        assert update_event(event_id=event.id,
+                            event=data, db=session) is None
+
+    @staticmethod
+    def test_successful_update(event, session):
+        data = {
+            "title": "successful",
+            "start": datetime(2021, 1, 20),
+            "end": datetime(2021, 1, 21),
+        }
+        assert isinstance(update_event(1, data, session), Event)
+        assert "successful" in update_event(
+            event_id=event.id, event=data, db=session).title
+
+    @staticmethod
+    def test_update_db_close(event):
+        data = {"title": "Problem connecting to db", }
+        assert update_event(event_id=event.id,
+                            event=data, db=None) is None
+
+    @staticmethod
+    def test_update_event_does_not_exist(event, session):
+        data = {
+            "content": "An update test for an event does not exist"
+        }
+        assert update_event(
+            event_id=500, event=data, db=session) is None
+
     def test_repr(self, event):
         assert event.__repr__() == f'<Event {event.id}>'