sqlor/test/test.py

61 lines
1.4 KiB
Python
Raw Normal View History

2019-05-14 10:47:28 +08:00
import asyncio
from sqlor.dbpools import DBPools
2019-05-23 16:28:36 +08:00
from sqlor.records import Records
2019-05-14 10:47:28 +08:00
dbs={
"aiocfae":{
"driver":"aiomysql",
"async_mode":True,
"coding":"utf8",
"dbname":"cfae",
"kwargs":{
"user":"test",
"db":"cfae",
"password":"test123",
"host":"localhost"
}
},
"cfae":{
"driver":"mysql.connector",
"coding":"utf8",
"dbname":"cfae",
"kwargs":{
"user":"test",
"db":"cfae",
"password":"test123",
"host":"localhost"
}
}
}
loop = asyncio.get_event_loop()
pool = DBPools(dbs,loop=loop)
2019-05-14 11:20:28 +08:00
async def testfunc1():
2019-05-14 10:47:28 +08:00
@pool.runSQL
def sql(db,ns,callback):
return {
"sql_string":"select * from product",
}
2019-05-23 16:28:36 +08:00
recs = Records()
x = await sql('cfae',{},recs.add)
print('--------------%s---------------' % x)
for r in recs:
print(type(r),r)
2019-05-14 10:47:28 +08:00
2019-05-14 11:20:28 +08:00
async def testfunc2():
@pool.runSQLResultFields
2019-05-23 16:28:36 +08:00
def sql(db,ns,callback=None):
2019-05-14 11:20:28 +08:00
return {
"sql_string":"select * from product",
}
2019-05-23 16:28:36 +08:00
recs = Records()
x = await sql('cfae',{},callback=recs.add)
print('-------%s------' % x)
for i in recs._records:
print("--",i)
2019-05-14 11:20:28 +08:00
loop.run_until_complete(testfunc1())
loop.run_until_complete(testfunc2())