HEX
Server: Apache/2.4.52 (Ubuntu)
System: Linux spn-python 5.15.0-89-generic #99-Ubuntu SMP Mon Oct 30 20:42:41 UTC 2023 x86_64
User: arjun (1000)
PHP: 8.1.2-1ubuntu2.20
Disabled: NONE
Upload Files
File: //home/arjun/projects/buyercall_new/buyercall/buyercall/blueprints/billing/models/coupon.py
import datetime
import pytz
import random
import string

from collections import OrderedDict
from os import urandom

from sqlalchemy import or_, and_
from sqlalchemy.sql import text
from sqlalchemy.ext.hybrid import hybrid_property

from buyercall.lib.util_sqlalchemy import ResourceMixin, AwareDateTime
from buyercall.lib.money import cents_to_dollars, dollars_to_cents
from buyercall.extensions import db
from buyercall.blueprints.billing.gateways.stripecom import \
    Coupon as PaymentCoupon


class Coupon(ResourceMixin, db.Model):
    DURATION = OrderedDict([
        ('forever', 'Forever'),
        ('once', 'Once'),
        ('repeating', 'Repeating')
    ])

    __tablename__ = 'coupons'
    id = db.Column(db.Integer, primary_key=True)

    # Coupon details.
    code = db.Column(db.String(32), index=True, unique=True)
    duration = db.Column(db.Enum(*DURATION, name='duration_types'),
                         index=True, nullable=False, server_default='forever')
    amount_off = db.Column(db.Integer())
    percent_off = db.Column(db.Integer())
    currency = db.Column(db.String(8))
    duration_in_months = db.Column(db.Integer())
    max_redemptions = db.Column(db.Integer(), index=True)
    redeem_by = db.Column(AwareDateTime(), index=True)
    times_redeemed = db.Column(db.Integer(), index=True,
                               nullable=False, default=0)
    valid = db.Column(db.Boolean(), nullable=False, server_default='1')

    def __init__(self, **kwargs):
        if self.code:
            self.code = self.code.upper()
        else:
            self.code = Coupon.random_coupon_code()

        # Call Flask-SQLAlchemy's constructor.
        super(Coupon, self).__init__(**kwargs)

    @hybrid_property
    def redeemable(self):
        """
        Return coupons that are still redeemable. Coupons will become invalid
        once they run out on save. We want to explicitly do a date check to
        avoid having to hit Stripe's API to get back potentially valid codes.

        :return: SQLAlchemy query object
        """
        is_redeemable = or_(self.redeem_by.is_(None),
                            self.redeem_by >= datetime.datetime.now(pytz.utc))

        return and_(self.valid, is_redeemable)

    @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_(Coupon.code.ilike(search_query)))

    @classmethod
    def random_coupon_code(cls):
        """
        Create a human readable random coupon code.

        Inspired by:
          http://stackoverflow.com/a/22333563

        :return: str
        """
        random_string = ''.join(
            random.choice('{}{}'.format(string.ascii_lowercase[:10], string.digits)) for _ in range(20)
        )
        long_code = random_string.translate(str.maketrans('0123456789abcdefghij', '234679QWERTYUPADFGHX'))

        short_code = '{0}-{1}-{2}'.format(long_code[0:4], long_code[5:9], long_code[10:14])

        return short_code

    @classmethod
    def expire_old_coupons(cls, compare_datetime=None):
        """
        Invalidate coupons that are past their redeem date.

        :param compare_datetime: Time to compare at
        :type compare_datetime: date
        :return: The result of updating the records
        """
        from buyercall.app import create_app
        # Create a context for the database connection.
        app = create_app()
        db.app = app
        with app.app_context():
            if compare_datetime is None:
                compare_datetime = datetime.datetime.now(pytz.utc)

            condition = Coupon.redeem_by <= compare_datetime
            Coupon.query.filter(condition) \
                .update({Coupon.valid: not Coupon.valid})

            return db.session.commit()

    @classmethod
    def create(cls, params):
        """
        Return whether or not the coupon was created successfully.

        :return: bool
        """
        payment_params = params

        payment_params['code'] = payment_params['code'].upper()

        if payment_params.get('amount_off'):
            payment_params['amount_off'] = \
                dollars_to_cents(payment_params['amount_off'])

        PaymentCoupon.create(**payment_params)

        if 'id' in payment_params:
            payment_params['code'] = payment_params['id']
            del payment_params['id']

        if 'redeem_by' in payment_params:
            if payment_params.get('redeem_by') is not None:
                params['redeem_by'] = payment_params.get('redeem_by').replace(
                    tzinfo=pytz.UTC)

        coupon = Coupon(**payment_params)

        db.session.add(coupon)
        db.session.commit()

        return True

    @classmethod
    def bulk_delete(cls, ids):
        """
        Override the general bulk_delete method because we need to delete them
        one at a time while also deleting them on Stripe.

        :param ids: List of ids to be deleted
        :type ids: list
        :return: int
        """
        delete_count = 0

        for id in ids:
            coupon = Coupon.query.get(id)

            if coupon is None:
                continue

            # Delete on Stripe.
            stripe_response = PaymentCoupon.delete(coupon.code)

            # If successful, delete it locally.
            if stripe_response.get('deleted'):
                coupon.delete()
                delete_count += 1

        return delete_count

    @classmethod
    def find_by_code(cls, code):
        """
        Find a coupon by its code.

        :param code: Coupon code to find
        :type code: str
        :return: Coupon instance
        """
        formatted_code = code.upper()
        coupon = Coupon.query.filter(Coupon.redeemable,
                                     Coupon.code == formatted_code).first()

        return coupon

    def redeem(self):
        """
        Update the redeem stats for this coupon.

        :return: Result of saving the record
        """
        self.times_redeemed += 1

        if self.max_redemptions:
            if self.times_redeemed >= self.max_redemptions:
                self.valid = False

        return db.session.commit()

    def serialize(self):
        """
        Return JSON fields to render the coupon code status.

        :return: dict
        """
        params = {
            'duration': self.duration,
            'duration_in_months': self.duration_in_months,
        }

        if self.amount_off:
            params['amount_off'] = cents_to_dollars(self.amount_off)

        if self.percent_off:
            params['percent_off'] = self.percent_off,

        return params