import decimal
from billing.lib.exceptions import NoResultFound
from billing.model.defaults import *
from billing.model.address import Address
from billing.model.company import Company
from billing.model.creditcard import CreditCard,PaymentAccount
from billing.model.membership import SiteMembership
from billing.model.meta import Session
class SecretQuestion(Base):
"""A class that represents a secret question for account lookup."""
__tablename__ = "secret_questions"
__table_args__ = (UniqueConstraint("company_id", "question"), {"mysql_engine": "InnoDB"})
query = Session.query_property()
id = Column(Integer, primary_key=True)
question = Column(Unicode(255), unique=True, index=True, nullable=False)
company_id = Column(Integer, ForeignKey(Company.id), index=True, nullable=False)
active = Column(Boolean, default=True, server_default="1", nullable=False, index=True)
created = Column(DateTime, nullable=False, default=datetime.datetime.now)
modified = Column(DateTime, nullable=True, onupdate=datetime.datetime.now)
def __repr__(self):
return self.question
DDL("ALTER TABLE %(fullname)s COMMENT = 'A secret question to use for confirmation. Ie., What city did you grow up in?'").execute_at("after-create", SecretQuestion.__table__)
class Account(Base):
"""A class representing a global account."""
__tablename__ = "accounts"
__table_args__ = (UniqueConstraint("company_id", "username"), {"mysql_engine": "InnoDB"})
query = Session.query_property()
id = Column(mysql.BIGINT(unsigned=True), primary_key=True)
company_id = Column(Integer, ForeignKey(Company.id), nullable=False, index=True)
first_name = Column(Unicode(50), nullable=False, index=True)
last_name = Column(Unicode(50), nullable=False, index=True)
email = Column(Unicode(100), unique=True, nullable=False, index=True)
active = Column(Boolean, nullable=False, server_default="1", default=True, index=True)
username = Column(Unicode(100), unique=True, nullable=False, index=True)
password = Column(Unicode(30), nullable=False, index=True)
secret_question_id = Column(Integer(unsigned=True), ForeignKey(SecretQuestion.id), nullable=True, index=True)
secret_answer = Column(Unicode(255), nullable=True, index=True)
primary_payment_account_id = Column(mysql.BIGINT(unsigned=True), ForeignKey(CreditCard.id, name="primary_payment_account_id", use_alter=True), nullable=True, index=True)
secondary_payment_account_id = Column(mysql.BIGINT(unsigned=True), ForeignKey(CreditCard.id, name="secondary_payment_account_id", use_alter=True), nullable=True, index=True)
created = Column(DateTime, nullable=False, default=datetime.datetime.now)
modified = Column(DateTime, nullable=True, onupdate=datetime.datetime.now)
# relations
question = relation("SecretQuestion", backref=backref("accounts", order_by=id), lazy=True)
memberships = relation("SiteMembership", primaryjoin=id==SiteMembership.account_id, backref=backref("account"), lazy=True)
primary_payment = relation("PaymentAccount", primaryjoin=primary_payment_account_id==PaymentAccount.id, backref=backref("account", foreign_keys=[primary_payment_account_id,]), lazy=True)
secondary_payment = relation("PaymentAccount", primaryjoin=secondary_payment_account_id==PaymentAccount.id, backref=backref("account2", foreign_keys=[secondary_payment_account_id,]), lazy=True)
credit_cards = relation("CreditCard", primaryjoin=id==CreditCard.account_id, lazy=True)
addresses = relation("Address", backref=backref("accounts", order_by=id), lazy=True)
company = relation("Company", backref=backref("company", order_by=id), lazy=True)
@classmethod
def by_username(self, username=None):
"""Lookup a user by their username."""
if username:
try:
return self.query.filter_by(username=username).one()
except NoResultFound, e:
return False
@classmethod
def by_id(self, id=None):
"""Lookup a user by their user id."""
if id:
try:
return self.query.filter_by(id=id).one()
except NoResultFound, e:
return False
@classmethod
def by_email(self, email=None):
"""Lookup a user by their user email."""
if email:
try:
return self.query.filter_by(email=email).one()
except NoResultFound, e:
return False
@property
def full_name(self):
"""Returns their full name for easier display."""
return "%s %s" % (self.first_name, self.last_name)
@property
def transactions(self):
from billing.model.transaction import Transaction
return Transaction.query.filter_by(account_id=self.id).all()
@property
def total_billing(self):
from billing.model.transaction import Transaction
total = decimal.Decimal(0)
t = Transaction.query.filter_by(account_id=self.id).all()
for transaction in t:
total += decimal.Decimal(transaction.amount) if transaction.amount else decimal.Decimal(0)
return total
#return Transaction.query.filter_by(account_id=self.id).values(Transaction.amount).one()
@property
def total_transactions(self):
from billing.model.transaction import Transaction
#return 5
return Transaction.query.filter_by(account_id=self.id).value(func.count(Transaction.id).label("total"))
@property
def last_transaction_id(self):
from billing.model.transaction import Transaction
latest_trans = Transaction.query.filter_by(account_id=self.id).\
filter_by(payment_account_id=self.primary_payment_account_id).\
order_by(Transaction.created.desc()).limit(1).first()
return latest_trans.processor_tid
def __repr__(self):
return self.full_name
Index("ix_account_username_password", Account.username, Account.password)
Index("ix_account_username_password_active", Account.username, Account.password, Account.active)
Index("ix_account_username_active", Account.username, Account.active)
Index("ix_account_username_sqid", Account.username, Account.secret_question_id)
Index("ix_account_username_ccid", Account.username, Account.primary_payment_account_id)
Index("ix_account_id_ccid", Account.id, Account.primary_payment_account_id)
Index("ix_account_id_sqid", Account.id, Account.secret_question_id)
DDL("ALTER TABLE %(fullname)s COMMENT = 'A global account to which we link all products.'").execute_at("after-create", Account.__table__)