Update event - Improving the code

app/routers/event.py

@@ -1,20 +1,19 @@
-from datetime import datetime as dt
+from datetime import datetime
 from operator import attrgetter
 from typing import Any, Dict, List, Optional
 
-from fastapi import APIRouter, Depends, Request
-from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy.orm import Session
-from starlette import status
-from starlette.responses import RedirectResponse
-from starlette.status import HTTP_302_FOUND
-
 from app.database.database import get_db
 from app.database.models import Event, User, UserEvent
 from app.dependencies import templates
 from app.internal.event import validate_zoom_link
 from app.internal.utils import create_model
 from app.routers.user import create_user
+from fastapi import APIRouter, Depends, HTTPException, Request
+from loguru import logger
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+from starlette import status
+from starlette.responses import RedirectResponse
 
 router = APIRouter(
     prefix="/event",
@@ -34,10 +33,10 @@ async def create_new_event(request: Request, session=Depends(get_db)):
     data = await request.form()
     title = data['title']
     content = data['description']
-    start = dt.strptime(data['start_date'] + ' ' + data['start_time'],
-                        '%Y-%m-%d %H:%M')
-    end = dt.strptime(data['end_date'] + ' ' + data['end_time'],
-                      '%Y-%m-%d %H:%M')
+    start = datetime.strptime(data['start_date'] + ' ' + data['start_time'],
+                              '%Y-%m-%d %H:%M')
+    end = datetime.strptime(data['end_date'] + ' ' + data['end_time'],
+                            '%Y-%m-%d %H:%M')
     user = session.query(User).filter_by(id=1).first()
     user = user if user else create_user("u", "p", "e@mail.com", session)
     owner_id = user.id
@@ -51,7 +50,7 @@ async def create_new_event(request: Request, session=Depends(get_db)):
     event = create_event(session, title, start, end, owner_id, content,
                          location)
     return RedirectResponse(router.url_path_for('eventview', id=event.id),
-                            status_code=HTTP_302_FOUND)
+                            status_code=status.HTTP_302_FOUND)
 
 
 @router.get("/view/{id}")
@@ -60,43 +59,68 @@ async def eventview(request: Request, id: int):
                                       {"request": request, "event_id": id})
 
 
+UPDATE_EVENTS_FIELDS = {
+    'title': str,
+    'start': datetime,
+    'end': datetime,
+    'content': (str, type(None)),
+    'location': (str, type(None))
+}
+
+
 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: dt, end_date: dt) -> bool:
+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 is_change_dates_allowed(
+        old_event: Event, event: Dict[str, Any]) -> bool:
+    try:
+        return is_date_before(
+            event.get('start', old_event.start),
+            event.get('end', old_event.end))
+    except TypeError:
+        return False
+
+
+def is_fields_types_valid(to_check: Dict[str, Any], types: Dict[str, Any]):
+    """validate dictionary values by dictionary of types"""
+    errors = []
+    for key in to_check.keys():
+        if types[key] and not isinstance(to_check[key], types[key]):
+            errors.append(
+                f"{key} is '{type(to_check[key]).__name__}' and it should be"
+                + f"from type '{types[key].__name__}'")
+            logger.warning(errors)
+    if errors:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail=errors)
 
 
-def get_items_that_can_be_updated(event: Dict[str, Any]) -> Dict[str, Any]:
-    """Extract only that keys to update"""
+def get_event_with_editable_fields_only(event: Dict[str, Any]
+                                        ) -> Dict[str, Any]:
+    """Remove all keys that are not allowed to update"""
 
-    return {i: event[i] for i in (
-        'title', 'start', 'end', 'content', 'location') if i in event}
+    return {i: event[i] for i in UPDATE_EVENTS_FIELDS.keys() 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
+    event_to_update = get_event_with_editable_fields_only(event)
+    is_fields_types_valid(event_to_update, UPDATE_EVENTS_FIELDS)
     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):
+        old_event = by_id(db, event_id)
+        forbidden = not is_change_dates_allowed(old_event, event_to_update)
+        if not event_to_update or old_event is None or forbidden:
             return None
 
         # Update database
@@ -105,9 +129,10 @@ def update_event(event_id: int, event: Dict, db: Session
         db.commit()
 
         # TODO: Send emails to recipients.
-    except (AttributeError, SQLAlchemyError, TypeError):
+        return by_id(db, event_id)
+    except (AttributeError, SQLAlchemyError) as e:
+        logger.error(str(e))
         return None
-    return by_id(db=db, event_id=event_id)
 
 
 def create_event(db, title, start, end, owner_id, content=None, location=None):
@@ -150,12 +175,19 @@ def get_participants_emails_by_event(db: Session, event_id: int) -> List[str]:
 
 
 @router.delete("/{event_id}")
-def delete_event(request: Request,
-                 event_id: int,
+def delete_event(event_id: int,
                  db: Session = Depends(get_db)):
 
     # TODO: Check if the user is the owner of the event.
-    event = by_id(db, event_id)
+    try:
+        event = by_id(db, event_id)
+    except AttributeError as e:
+        logger.error(str(e) + "Could not connect to database")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                            detail="Could not connect to database")
+    if not event:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail="The event was not found")
     participants = get_participants_emails_by_event(db, event_id)
     try:
         # Delete event
@@ -166,11 +198,12 @@ def delete_event(request: Request,
 
         db.commit()
 
-    except (SQLAlchemyError, TypeError):
-        return templates.TemplateResponse(
-            "event/eventview.html", {"request": request, "event_id": event_id},
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
-    if participants and event.start > dt.now():
+    except (SQLAlchemyError, AttributeError) as e:
+        logger.error(str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Deletion failed")
+    if participants and event.start > datetime.now():
         pass
         # TODO: Send them a cancellation notice
         # if the deletion is successful

tests/test_event.py

@@ -1,11 +1,11 @@
 from datetime import datetime
 
 import pytest
-from starlette import status
-from starlette.status import HTTP_302_FOUND
-
 from app.database.models import Event
-from app.routers.event import by_id, update_event
+from app.routers.event import (by_id, delete_event, is_change_dates_allowed,
+                               update_event)
+from fastapi import HTTPException
+from starlette import status
 
 CORRECT_EVENT_FORM_DATA = {
     'title': 'test title',
@@ -36,7 +36,7 @@
 }
 
 INVALID_UPDATE_OPTIONS = [
-    {}, {"test": "test"}, {"start": "20.01.2020"},
+    {}, {"test": "test"},
     {"start": datetime(2020, 2, 2), "end": datetime(2020, 1, 1)},
     {"start": datetime(2030, 2, 2)}, {"end": datetime(1990, 1, 1)},
 ]
@@ -57,27 +57,37 @@ def test_eventview_without_id(self, client):
         response = client.get("/event/view")
         assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
 
-    def test_eventedit_post_correct(self, client, user):
-        response = client.post(client.app.url_path_for('create_new_event'),
-                               data=CORRECT_EVENT_FORM_DATA)
+    def test_eventedit_post_correct(self, event_test_client, user):
+        response = event_test_client.post(
+            event_test_client.app.url_path_for(
+                'create_new_event'), data=CORRECT_EVENT_FORM_DATA)
         assert response.ok
-        assert response.status_code == HTTP_302_FOUND
-        assert (client.app.url_path_for('eventview', id=1).strip('1')
-               in response.headers['location'])
-
-    def test_eventedit_post_wrong(self, client, user):
-        response = client.post(client.app.url_path_for('create_new_event'),
-                               data=WRONG_EVENT_FORM_DATA)
+        assert response.status_code == status.HTTP_302_FOUND
+        assert (event_test_client.app.url_path_for(
+            'eventview', id=1).strip('1') in response.headers['location'])
+
+    def test_eventedit_post_wrong(self, event_test_client, user):
+        response = event_test_client.post(
+            event_test_client.app.url_path_for(
+                'create_new_event'), data=WRONG_EVENT_FORM_DATA)
         assert response.json()['detail'] == 'VC type with no valid zoom link'
 
-    @staticmethod
     @pytest.mark.parametrize("data", INVALID_UPDATE_OPTIONS)
-    def test_invalid_update(event, data, session):
+    def test_invalid_update(self, event, data, session):
         assert update_event(event_id=event.id,
                             event=data, db=session) is None
 
-    @staticmethod
-    def test_successful_update(event, session):
+    def test_fields_types_invalid(self, event, session):
+        data = {"start": "20.01.2020"}
+        with pytest.raises(HTTPException):
+            response = update_event(event_id=event.id, event=data, db=session)
+            assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+    def test_not_is_change_dates_allowed(self, event):
+        data = {"start": "20.01.2020"}
+        assert is_change_dates_allowed(event, data) is False
+
+    def test_successful_update(self, event, session):
         data = {
             "title": "successful",
             "start": datetime(2021, 1, 20),
@@ -87,14 +97,12 @@ def test_successful_update(event, session):
         assert "successful" in update_event(
             event_id=event.id, event=data, db=session).title
 
-    @staticmethod
-    def test_update_db_close(event):
+    def test_update_db_close(self, 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):
+    def test_update_event_does_not_exist(self, event, session):
         data = {
             "content": "An update test for an event does not exist"
         }
@@ -105,10 +113,16 @@ def test_repr(self, event):
         assert event.__repr__() == f'<Event {event.id}>'
 
     def test_successful_deletion(self, event_test_client, session, event):
-        respons = event_test_client.delete("/event/1")
-        assert respons.ok
+        response = event_test_client.delete("/event/1")
+        assert response.ok
         assert by_id(db=session, event_id=1) is None
 
     def test_delete_failed(self, event_test_client, event):
-        respons = event_test_client.delete("/event/2")
-        assert respons.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
+        response = event_test_client.delete("/event/2")
+        assert response.status_code == status.HTTP_404_NOT_FOUND
+
+    def test_no_connection_to_db(self, event):
+        with pytest.raises(HTTPException):
+            response = delete_event(event_id=1, db=None)
+            assert response.status_code == status.\
+                HTTP_500_INTERNAL_SERVER_ERROR