accounting/test/test_rf.py
2024-09-22 12:05:37 +08:00

41 lines
1.2 KiB
Python

from appPublic.registerfunction import RegisterFunction
rf = RegisterFunction()
async def get_providers_by_orgid(sor, accounting_orgid):
sql = """select * from organization a, orgtypes b
where a.parentid = ${accounting_orgid}$ and
b.orgtypeid = 'provider' and
a.id = b.orgid """
recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
return recs
rf.register('get_providers_by_orgid', get_providers_by_orgid)
async def get_resellers_by_orgid(sor, accounting_orgid):
sql = """select * from organization a, orgtypes b
where a.parentid=${accounting_orgid}$ and
a.id = b.orgid and
b.orgtypeid = 'reseller'"""
recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
return recs
rf.register('get_resellers_by_orgid', get_resellers_by_orgid)
async def get_customers_by_orgid(sor, accounting_orgid):
sql = """select * from organization a,
(select orgid, count(*) cnt
from orgtypes
where orgtypeid in ('customer', 'personalcustomer', 'agencycustomer')
group by orgid
) b
where a.parentid=${accounting_orgid}$ and
a.id = b.orgid and
b.cnt > 0
"""
recs = await sor.sqlExe(sql, {'accounting_orgid':accounting_orgid})
return recs
rf.register('get_customers_by_orgid', get_customers_by_orgid)