-
Notifications
You must be signed in to change notification settings - Fork 0
/
Queries.py
60 lines (52 loc) · 1.74 KB
/
Queries.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from datetime import date
from sqlalchemy.sql import asc, cast, func
from sqlalchemy.types import INT
import Model as M
from Database import ensureSession, resetSession, requireSession, flushSession
date_part = func.date_part
def allTransactionMonths():
session = requireSession()
parts = "year", "month"
query = session \
.query(*[cast(date_part(p, M.Transaction.date), INT)
.label(p) for p in parts]) \
.order_by(*[asc(p) for p in parts]) \
.distinct()
return list(query)
def transactionsInDateRange(start, end):
session = requireSession()
query = session \
.query(M.transactionListMapper) \
.filter((M.transactionListMapper.c.date >= start)
& (M.transactionListMapper.c.date < end)) \
.order_by(asc(M.transactionListMapper.c.date))
return list(query)
def transactionsForMonth(year, month):
if month == 12:
end_year, end_month = year+1, 1
else:
end_year, end_month = year, month+1
return transactionsInDateRange(date(year, month, 1),
date(end_year, end_month, 1))
def markTransactionAsInvoiceable(t, invoiceable):
if invoiceable == t.isInvoiceable:
return
session = requireSession()
if invoiceable:
session.add(M.InvoiceableItem(t))
else:
session.delete(
session \
.query(M.InvoiceableItem) \
.filter(M.InvoiceableItem.fitid == t.fitid))
def markTransactionAsPayment(t, person):
person.payments.append(M.PaymentWithStatementEvidence(t))
def invoicePerson(person, invoiceDate, amount=None):
session = requireSession()
i = M.Invoice(invoiceDate, amount or M.Decimal(0))
person.invoices.append(i)
session.flush()
if amount is None:
i.amount = i.computedAmount
session.flush()
return i