from sqlalchemy import types as T from sqlalchemy import sql, Index, Column, ForeignKey, create_engine from sqlalchemy.orm import relationship, backref, scoped_session, sessionmaker,\ column_property from sqlalchemy.ext.declarative import declarative_base, declared_attr from sqlalchemy.ext.hybrid import hybrid_property import datetime import decimal from functools import partial from collections import namedtuple __all__ = ["Category", "SingleExpense", "ConstExpense", "CatExpense", "MonthExpense", "Session"] # # DB Setup # engine = create_engine("sqlite:///test.sqlite") engine.echo = True Session = scoped_session(sessionmaker(bind=engine)) # # Global definitions # class Base(object): @declared_attr def __tablename__ (cls): return cls.__name__.lower() id = Column(T.Integer, primary_key=True) query = Session.query_property() @classmethod def get_by (cls, *args, **kwargs): return cls.query.filter_by(*args, **kwargs).first() @classmethod def get (cls, *args, **kwargs): return cls.query.get(*args, **kwargs) Base = declarative_base(cls=Base) ReqColumn = partial(Column, nullable = False) ExpNum = T.Numeric(scale = 2, precision = 10) def to_exp(d): """Converts decimal into expense""" return d.quantize(decimal.Decimal('.01'), rounding = decimal.ROUND_UP) # # Database Entities # class Category (Base): name = ReqColumn(T.Unicode(50), unique = True) parent_id = Column(T.Integer, ForeignKey('category.id')) children = relationship('Category', backref=backref('parent', remote_side="Category.id")) def __repr__ (self): if self.parent: return '' % (self.name, self.parent.name) else: return '' % self.name class Expense (Base): __abstract__ = True description = Column(T.Unicode(50)) expense = ReqColumn(ExpNum) @declared_attr def category_id(cls): return ReqColumn(T.Integer, ForeignKey(Category.id)) @declared_attr def category(cls): return relationship(Category, innerjoin = True) class SingleExpense (Expense): year = ReqColumn(T.Integer) month = ReqColumn(T.SmallInteger) day = ReqColumn(T.SmallInteger) @classmethod def of_month (cls, month, year): comp = sql.and_( cls.month == month, cls.year == year) return cls.query.filter(comp) @property def date (self): return datetime.date(self.year, self.month, self.day) @date.setter def date (self, d): self.year = d.year self.month = d.month self.day = d.day class ConstExpense (Expense): months = ReqColumn(T.SmallInteger) start = ReqColumn(T.Date, index = True) end = ReqColumn(T.Date, index = True) prev_id = Column(T.Integer, ForeignKey('constexpense.id')) prev = relationship('ConstExpense', remote_side = "ConstExpense.id", uselist = False, backref=backref('next', uselist = False)) @property def monthly(self): return to_exp(self.expense / self.months) @classmethod def of_month (cls, month, year): d = datetime.date(year, month, 1) return cls.query.filter(sql.between(d, cls.start, cls.end)) # # Work entities (not stored in DB) # class CatExpense (namedtuple('CatExpense', 'cat expense exps')): __slots__ = () @property def all (self): return self.exps.order_by(SingleExpense.day).all() class MonthExpense (namedtuple('MonthExpense', 'date catexps')): def __init__ (self, *args, **kwargs): self._consts = None super(MonthExpense, self).__init__(*args, **kwargs) @property def consts (self): if self._consts is None: self._consts = ConstExpense.of_month(self.date.month, self.date.year).all() return self._consts @property def constsum (self): s = sum(c.monthly for c in self.consts) return s or 0 @property def sum (self): return self.constsum + sum(x.expense for x in self.catexps) @property def all (self): return SingleExpense.of_month(self.date.month, self.date.year).order_by(SingleExpense.day).all() def __str__ (self): return '' % (self.date, self.sum) # # Extra indizes have to be here # Index('idx_single_date', SingleExpense.year, SingleExpense.month) Index('idx_start_end', ConstExpense.start, ConstExpense.end) if __name__ == "__main__": Base.metadata.create_all(engine)