File: //proc/1233/root/home/arjun/projects/buyercall/buyercall/blueprints/issue/models.py
from collections import OrderedDict
from re import I
import uuid
from sqlalchemy import or_, and_
from sqlalchemy.sql import text
from sqlalchemy.dialects.postgresql import UUID
from buyercall.lib.util_sqlalchemy import ResourceMixin
from buyercall.extensions import db
import pendulum, pytz
class Issue(ResourceMixin, db.Model):
STATUS = OrderedDict([
('unread', 'Unread'),
('open', 'Open'),
('contacted', 'Contacted'),
('closed', 'Closed')
])
LABEL = OrderedDict([
('login', 'I cannot access my account'),
('form', 'I require a dealer form and id'),
('phone', 'I require a phone number for calling and texting'),
('data', 'I am not seeing my data or the data is incorrect'),
('billing', 'I have a billing question'),
('email', 'I am not receiving e-mails'),
('request', 'I want to request a feature'),
('other', 'Other')
])
__tablename__ = 'issues'
id = db.Column(db.Integer, primary_key=True)
sid = db.Column(UUID(as_uuid=True), unique=True, default=uuid.uuid4, index=True)
status = db.Column(db.Enum(*STATUS.keys(), name='status_types'),
index=True, nullable=False, server_default='unread')
label = db.Column(db.Enum(*LABEL.keys(), name='label_types'),
index=True, nullable=False, server_default='login')
email = db.Column(db.String(255), index=True, nullable=False,
server_default='')
question = db.Column(db.Text())
partnership_id = db.Column(db.Integer, db.ForeignKey('partnerships.id',
onupdate='CASCADE',
ondelete='CASCADE'),
index=True, nullable=True)
partnership_account_id = db.Column(db.Integer, db.ForeignKey('partnership_accounts.id',
onupdate='CASCADE',
ondelete='CASCADE'),
index=True, nullable=True)
@classmethod
def search(cls, query):
"""
Search a resource by 1 or more fields.
:param query: Search query
:type query: str
:return: SQLAlchemy filter
"""
if not query:
return text('')
search_query = '%{0}%'.format(query)
return text(or_(Issue.email.ilike(search_query)))
@classmethod
def unread_to_open(cls, issue):
"""
Change unread issues to open.
:param issue: Issue instance
:type issue: Issue instance
:return: Issue instance
"""
issue.status = 'open'
issue.save()
return issue
@classmethod
def set_as_contacted(cls, issue):
"""
Change an unknown issue type to contacted.
:param issue: Issue instance
:type issue: Issue instance
:return: Issue instance
"""
issue.status = 'contacted'
issue.save()
return issue
@classmethod
def get_by_partnership(cls, pid, count_only=False, from_date=None, to_date=None):
"""
Get issues by partnership id.
:param: count_only - Boolean (Is only count of issues needed)
:param: from_date - Datetime
:param: to_date - Datetime
:return: Issue objects or objects count
"""
if from_date and to_date:
users = cls.query\
.filter(and_(cls.partnership_id == pid, cls.created_on >= from_date, cls.created_on < to_date))
else:
users = cls.query\
.filter(cls.partnership_id == pid)
return users.count() if count_only else users.all()
@classmethod
def get_by_partnership_datetime_expr(cls, pid, count_only=False, date_range_expr=None):
"""
:param date_range_expr: type:string, possible values: this_month, last_month
"""
if date_range_expr:
if date_range_expr == 'this_month':
today = pendulum.today(tz=pytz.utc)
first_day = today.replace(day=1)
return cls.get_by_partnership(pid, count_only=count_only, from_date=first_day, to_date=today)
if date_range_expr == 'last_month':
today = pendulum.today(tz=pytz.utc)
pre_month = today.month - 1
first_day_this_month = today.replace(day=1)
first_day_pre_month = today.replace(month=pre_month, day=1)
last_day_pre_month = first_day_this_month # Time considers as 00:00:00
return cls.get_by_partnership(pid, count_only=count_only, from_date=first_day_pre_month, to_date=last_day_pre_month)
else:
return cls.get_by_partnership(pid, count_only=count_only)