Feature/notifications

AUTHORS.md

@@ -33,6 +33,7 @@
          * PureDreamer - Developer
          * ShiZinDle - Developer
          * YairEn - Developer
+         * IdanPelled - Developer
 
 # Special thanks to
 

app/database/models.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
-
 from datetime import datetime
+import enum
 from typing import Any, Dict
 
 from sqlalchemy import (
     Boolean,
     Column,
     DateTime,
     DDL,
+    Enum,
     event,
     Float,
     ForeignKey,
@@ -203,20 +204,80 @@ class PSQLEnvironmentError(Exception):
     )
 
 
+class InvitationStatusEnum(enum.Enum):
+    UNREAD = 0
+    ACCEPTED = 1
+    DECLINED = 2
+
+
+class MessageStatusEnum(enum.Enum):
+    UNREAD = 0
+    READ = 1
+
+
 class Invitation(Base):
     __tablename__ = "invitations"
 
     id = Column(Integer, primary_key=True, index=True)
-    status = Column(String, nullable=False, default="unread")
+    creation = Column(DateTime, default=datetime.now, nullable=False)
+    status = Column(
+        Enum(InvitationStatusEnum),
+        default=InvitationStatusEnum.UNREAD,
+        nullable=False,
+    )
+
     recipient_id = Column(Integer, ForeignKey("users.id"))
     event_id = Column(Integer, ForeignKey("events.id"))
-    creation = Column(DateTime, default=datetime.now)
-
     recipient = relationship("User")
     event = relationship("Event")
 
+    def decline(self, session: Session) -> None:
+        """declines the invitation."""
+        self.status = InvitationStatusEnum.DECLINED
+        session.merge(self)
+        session.commit()
+
+    def accept(self, session: Session) -> None:
+        """Accepts the invitation by creating an
+        UserEvent association that represents
+        participantship at the event."""
+
+        association = UserEvent(
+            user_id=self.recipient.id,
+            event_id=self.event.id,
+        )
+        self.status = InvitationStatusEnum.ACCEPTED
+        session.merge(self)
+        session.add(association)
+        session.commit()
+
+    def __repr__(self):
+        return f"<Invitation ({self.event.owner} to {self.recipient})>"
+
+
+class Message(Base):
+    __tablename__ = "messages"
+
+    id = Column(Integer, primary_key=True, index=True)
+    body = Column(String, nullable=False)
+    link = Column(String)
+    creation = Column(DateTime, default=datetime.now, nullable=False)
+    status = Column(
+        Enum(MessageStatusEnum),
+        default=MessageStatusEnum.UNREAD,
+        nullable=False,
+    )
+
+    recipient_id = Column(Integer, ForeignKey("users.id"))
+    recipient = relationship("User")
+
+    def mark_as_read(self, session):
+        self.status = MessageStatusEnum.READ
+        session.merge(self)
+        session.commit()
+
     def __repr__(self):
-        return f"<Invitation " f"({self.event.owner}" f"to {self.recipient})>"
+        return f"<Message {self.id}>"
 
 
 class UserSettings(Base):

app/database/schemas.py

@@ -2,7 +2,7 @@
 from pydantic import BaseModel, validator, EmailStr, EmailError
 
 
-EMPTY_FIELD_STRING = 'field is required'
+EMPTY_FIELD_STRING = "field is required"
 MIN_FIELD_LENGTH = 3
 MAX_FIELD_LENGTH = 20
 
@@ -19,59 +19,72 @@ class UserBase(BaseModel):
     Validating fields types
     Returns a User object without sensitive information
     """
+
     username: str
     email: str
     full_name: str
+
+    language_id: Optional[int] = 1
     description: Optional[str] = None
+    target_weight: Optional[Union[int, float]] = None
 
     class Config:
         orm_mode = True
 
 
 class UserCreate(UserBase):
     """Validating fields types"""
+
     password: str
     confirm_password: str
 
     """
     Calling to field_not_empty validaion function,
     for each required field.
     """
-    _fields_not_empty_username = validator(
-        'username', allow_reuse=True)(fields_not_empty)
-    _fields_not_empty_full_name = validator(
-        'full_name', allow_reuse=True)(fields_not_empty)
-    _fields_not_empty_password = validator(
-        'password', allow_reuse=True)(fields_not_empty)
+    _fields_not_empty_username = validator("username", allow_reuse=True)(
+        fields_not_empty,
+    )
+    _fields_not_empty_full_name = validator("full_name", allow_reuse=True)(
+        fields_not_empty,
+    )
+    _fields_not_empty_password = validator("password", allow_reuse=True)(
+        fields_not_empty,
+    )
     _fields_not_empty_confirm_password = validator(
-        'confirm_password', allow_reuse=True)(fields_not_empty)
-    _fields_not_empty_email = validator(
-        'email', allow_reuse=True)(fields_not_empty)
-
-    @validator('confirm_password')
+        "confirm_password",
+        allow_reuse=True,
+    )(fields_not_empty)
+    _fields_not_empty_email = validator("email", allow_reuse=True)(
+        fields_not_empty,
+    )
+
+    @validator("confirm_password")
     def passwords_match(
-            cls, confirm_password: str,
-            values: UserBase) -> Union[ValueError, str]:
+        cls,
+        confirm_password: str,
+        values: UserBase,
+    ) -> Union[ValueError, str]:
         """Validating passwords fields identical."""
-        if 'password' in values and confirm_password != values['password']:
+        if "password" in values and confirm_password != values["password"]:
             raise ValueError("doesn't match to password")
         return confirm_password
 
-    @validator('username')
+    @validator("username")
     def username_length(cls, username: str) -> Union[ValueError, str]:
         """Validating username length is legal"""
         if not (MIN_FIELD_LENGTH < len(username) < MAX_FIELD_LENGTH):
             raise ValueError("must contain between 3 to 20 charactars")
         return username
 
-    @validator('password')
+    @validator("password")
     def password_length(cls, password: str) -> Union[ValueError, str]:
         """Validating username length is legal"""
         if not (MIN_FIELD_LENGTH < len(password) < MAX_FIELD_LENGTH):
             raise ValueError("must contain between 3 to 20 charactars")
         return password
 
-    @validator('email')
+    @validator("email")
     def confirm_mail(cls, email: str) -> Union[ValueError, str]:
         """Validating email is valid mail address."""
         try:
@@ -86,5 +99,6 @@ class User(UserBase):
     Validating fields types
     Returns a User object without sensitive information
     """
+
     id: int
     is_active: bool

app/internal/notification.py

@@ -0,0 +1,176 @@
+from operator import attrgetter
+from typing import Iterator, List, Union, Callable
+
+from fastapi import HTTPException
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_406_NOT_ACCEPTABLE
+
+from app.database.models import (
+    Invitation,
+    Message,
+    InvitationStatusEnum,
+    MessageStatusEnum,
+)
+from app.internal.utils import create_model
+
+
+WRONG_NOTIFICATION_ID = (
+    "The notification id you have entered is wrong\n."
+    "If you did not enter the notification id manually, report this exception."
+)
+
+NOTIFICATION_TYPE = Union[Invitation, Message]
+
+UNREAD_STATUS = {
+    InvitationStatusEnum.UNREAD,
+    MessageStatusEnum.UNREAD,
+}
+
+ARCHIVED = {
+    InvitationStatusEnum.DECLINED,
+    MessageStatusEnum.READ,
+}
+
+
+async def get_message_by_id(
+    message_id: int,
+    session: Session,
+) -> Union[Message, None]:
+    """Returns an invitation by an id.
+    if id does not exist, returns None.
+    """
+    return session.query(Message).filter_by(id=message_id).first()
+
+
+def _is_unread(notification: NOTIFICATION_TYPE) -> bool:
+    """Returns True if notification is unread, False otherwise."""
+    return notification.status in UNREAD_STATUS
+
+
+def _is_archived(notification: NOTIFICATION_TYPE) -> bool:
+    """Returns True if notification should be
+    in archived page, False otherwise.
+    """
+    return notification.status in ARCHIVED
+
+
+def is_owner(user, notification: NOTIFICATION_TYPE) -> bool:
+    """Checks if user is owner of the notification.
+
+    Args:
+        notification: a NOTIFICATION_TYPE object.
+        user: user schema object.
+
+    Returns:
+        True or raises HTTPException.
+    """
+    if notification.recipient_id == user.user_id:
+        return True
+
+    msg = "The notification you are trying to access is not yours."
+    raise HTTPException(
+        status_code=HTTP_401_UNAUTHORIZED,
+        detail=msg,
+    )
+
+
+def raise_wrong_id_error() -> None:
+    """Raises HTTPException.
+
+    Returns:
+        None
+    """
+    raise HTTPException(
+        status_code=HTTP_406_NOT_ACCEPTABLE,
+        detail=WRONG_NOTIFICATION_ID,
+    )
+
+
+def filter_notifications(
+    session: Session,
+    user_id: int,
+    func: Callable[[NOTIFICATION_TYPE], bool],
+) -> Iterator[NOTIFICATION_TYPE]:
+    """Filters notifications by "func"."""
+    yield from filter(func, get_all_notifications(session, user_id))
+
+
+def get_unread_notifications(
+    session: Session,
+    user_id: int,
+) -> Iterator[NOTIFICATION_TYPE]:
+    """Returns all unread notifications."""
+    yield from filter_notifications(session, user_id, _is_unread)
+
+
+def get_archived_notifications(
+    session: Session,
+    user_id: int,
+) -> List[NOTIFICATION_TYPE]:
+    """Returns all archived notifications."""
+    yield from filter_notifications(session, user_id, _is_archived)
+
+
+def get_all_notifications(
+    session: Session,
+    user_id: int,
+) -> List[NOTIFICATION_TYPE]:
+    """Returns all notifications."""
+    invitations: List[Invitation] = get_all_invitations(
+        session,
+        recipient_id=user_id,
+    )
+    messages: List[Message] = get_all_messages(session, user_id)
+
+    notifications = invitations + messages
+    return sort_notifications(notifications)
+
+
+def sort_notifications(
+    notification: List[NOTIFICATION_TYPE],
+) -> List[NOTIFICATION_TYPE]:
+    """Sorts the notifications by the creation date."""
+    return sorted(notification, key=attrgetter("creation"), reverse=True)
+
+
+def create_message(
+    session: Session,
+    msg: str,
+    recipient_id: int,
+    link=None,
+) -> Message:
+    """Creates a new message."""
+    return create_model(
+        session,
+        Message,
+        body=msg,
+        recipient_id=recipient_id,
+        link=link,
+    )
+
+
+def get_all_messages(session: Session, recipient_id: int) -> List[Message]:
+    """Returns all messages."""
+    condition = Message.recipient_id == recipient_id
+    return session.query(Message).filter(condition).all()
+
+
+def get_all_invitations(session: Session, **param) -> List[Invitation]:
+    """Returns all invitations filter by param."""
+    try:
+        invitations = session.query(Invitation).filter_by(**param).all()
+    except SQLAlchemyError:
+        return []
+    else:
+        return invitations
+
+
+def get_invitation_by_id(
+    invitation_id: int,
+    session: Session,
+) -> Union[Invitation, None]:
+    """Returns an invitation by an id.
+    if id does not exist, returns None.
+    """
+    return session.query(Invitation).filter_by(id=invitation_id).first()