Class TaskEvent

shared/database/task/task_event.py:8–206

AKA: Streaming TaskEvents track all the actions or occurences related with a task. For example a change of status, archiving a task, creating a task or any other action related with a task object.

from shared.database.project import Project

class TaskEvent(Base, SerializerMixin):
    """
        AKA: Streaming

        TaskEvents track all the actions or occurences related with a task. For example a change of status,
        archiving a task, creating a task or any other action related with a task object.
    """
    __tablename__ = 'task_event'
    id = Column(BIGINT, primary_key = True)

    # The job context on where this sync happened.
    job_id = Column(Integer, ForeignKey('job.id'))
    job = relationship("Job", foreign_keys = [job_id])

    # For knowing in what project did the sync occurred.
    project_id = Column(Integer, ForeignKey('project.id'))
    project = relationship("Project", foreign_keys = [project_id])

    # For knowing which task was created.
    task_id = Column(Integer, ForeignKey('task.id'))
    task = relationship("Task", foreign_keys = [task_id])

    event_type = Column(String())  # ["completed", "archived",  "in_review", "created", "task_comment_created"]

    comment_id = Column(Integer, ForeignKey('discussion_comment.id'))
    comment = relationship("DiscussionComment", foreign_keys = [comment_id])

    time_created = Column(DateTime, default = datetime.datetime.utcnow)
    time_updated = Column(DateTime, onupdate = datetime.datetime.utcnow)

    user_reviewer_id = Column(Integer, ForeignKey('userbase.id'))
    user_reviewer = relationship("User", foreign_keys = [user_reviewer_id])

    user_assignee_id = Column(Integer, ForeignKey('userbase.id'))
    user_assignee = relationship("User", foreign_keys = [user_assignee_id])

    member_created_id = Column(Integer, ForeignKey('member.id'))
    member_created = relationship("Member", foreign_keys = [member_created_id])

    member_updated_id = Column(Integer, ForeignKey('member.id'))
    member_updated = relationship("Member", foreign_keys = [member_updated_id])

    def serialize(self):
        data = self.to_dict(rules = (
            '-member_created',
            '-member_updated',
            '-job',
            '-user_assignee',
            '-user_reviewer',
            '-task',
            '-comment',
            '-project'))

        if self.comment_id:
            data['comment'] = self.comment.serialize()
        return data

    @staticmethod