cic-stack/apps/cic-ussd/cic_ussd/db/models/task_tracker.py
2021-08-06 16:29:01 +00:00

35 lines
962 B
Python

# standard imports
import logging
# third-party imports
from sqlalchemy import Column, String
from sqlalchemy.orm.session import Session
# local imports
from cic_ussd.db.models.base import SessionBase
logg = logging.getLogger(__name__)
class TaskTracker(SessionBase):
__tablename__ = 'task_tracker'
def __init__(self, task_uuid):
self.task_uuid = task_uuid
task_uuid = Column(String, nullable=False)
@staticmethod
def add(session: Session, task_uuid: str):
"""This function persists celery tasks uuids to storage.
:param session: Database session object.
:type session: Session
:param task_uuid: The uuid for an initiated task.
:type task_uuid: str
"""
session = SessionBase.bind_session(session=session)
task_record = TaskTracker(task_uuid=task_uuid)
session.add(task_record)
session.flush()
SessionBase.release_session(session=session)