cic-internal-integration/apps/cic-ussd/cic_ussd/db/models/ussd_session.py

101 lines
3.1 KiB
Python
Raw Normal View History

2021-02-06 16:13:47 +01:00
# standard imports
import logging
# third-party imports
2021-08-06 18:29:01 +02:00
from sqlalchemy import Column, desc, Integer, String
2021-02-06 16:13:47 +01:00
from sqlalchemy.dialects.postgresql import JSON
from sqlalchemy.orm.attributes import flag_modified
2021-08-06 18:29:01 +02:00
from sqlalchemy.orm.session import Session
2021-02-06 16:13:47 +01:00
# local imports
from cic_ussd.db.models.base import SessionBase
from cic_ussd.error import VersionTooLowError
logg = logging.getLogger(__name__)
class UssdSession(SessionBase):
__tablename__ = 'ussd_session'
2021-08-06 18:29:01 +02:00
data = Column(JSON)
2021-02-06 16:13:47 +01:00
external_session_id = Column(String, nullable=False, index=True, unique=True)
msisdn = Column(String, nullable=False)
2021-08-06 18:29:01 +02:00
service_code = Column(String, nullable=False)
2021-02-06 16:13:47 +01:00
state = Column(String, nullable=False)
2021-08-06 18:29:01 +02:00
user_input = Column(String)
2021-02-06 16:13:47 +01:00
version = Column(Integer, nullable=False)
def set_data(self, key, session, value):
2021-08-06 18:29:01 +02:00
if self.data is None:
self.data = {}
self.data[key] = value
2021-02-06 16:13:47 +01:00
# https://stackoverflow.com/questions/42559434/updates-to-json-field-dont-persist-to-db
2021-08-06 18:29:01 +02:00
flag_modified(self, "data")
2021-02-06 16:13:47 +01:00
session.add(self)
def get_data(self, key):
2021-08-06 18:29:01 +02:00
if self.data is not None:
return self.data.get(key)
2021-02-06 16:13:47 +01:00
else:
return None
def check_version(self, new_version):
if new_version <= self.version:
raise VersionTooLowError('New session version number is not greater than last saved version!')
def update(self, user_input, state, version, session):
self.check_version(version)
self.user_input = user_input
self.state = state
self.version = version
session.add(self)
@staticmethod
2021-08-06 18:29:01 +02:00
def has_record_for_phone_number(phone_number: str, session: Session):
"""
:param phone_number:
:type phone_number:
:param session:
:type session:
:return:
:rtype:
"""
session = SessionBase.bind_session(session=session)
ussd_session = session.query(UssdSession).filter_by(msisdn=phone_number).first()
SessionBase.release_session(session=session)
return ussd_session is not None
@staticmethod
def last_ussd_session(phone_number: str, session: Session):
"""
:param phone_number:
:type phone_number:
:param session:
:type session:
:return:
:rtype:
"""
session = SessionBase.bind_session(session=session)
ussd_session = session.query(UssdSession) \
.filter_by(msisdn=phone_number) \
.order_by(desc(UssdSession.created)) \
.first()
SessionBase.release_session(session=session)
return ussd_session
2021-02-06 16:13:47 +01:00
def to_json(self):
""" This function serializes the in db ussd session object to a JSON object
:return: A JSON object of a ussd session in db
:rtype: dict
"""
return {
2021-08-06 18:29:01 +02:00
"data": self.data,
2021-02-06 16:13:47 +01:00
"external_session_id": self.external_session_id,
"msisdn": self.msisdn,
2021-08-06 18:29:01 +02:00
"service_code": self.service_code,
2021-02-06 16:13:47 +01:00
"state": self.state,
2021-08-06 18:29:01 +02:00
"user_input": self.user_input,
2021-02-06 16:13:47 +01:00
"version": self.version
}