This commit is contained in:
yumoqing 2020-05-06 16:44:12 +08:00
parent 2271bc4e15
commit 34421ed2d3

View File

@ -2,6 +2,7 @@
import asyncio import asyncio
from functools import wraps from functools import wraps
import codecs import codecs
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from appPublic.myImport import myImport from appPublic.myImport import myImport
@ -166,17 +167,21 @@ class ConnectionPool(object):
@SingletonDecorator @SingletonDecorator
class DBPools: class DBPools:
def __init__(self,databases={},max_connect=10,loop=None): def __init__(self,databases={},max_connect=100,loop=None):
if loop is None: if loop is None:
loop = asyncio.get_event_loop() loop = asyncio.get_event_loop()
self.loop = loop self.loop = loop
self.max_connect = max_connect
self.sema = asyncio.Semaphore(max_connect)
self._cpools = {} self._cpools = {}
self.databases = databases self.databases = databases
self.meta = {}
def addDatabase(self,name,desc): def addDatabase(self,name,desc):
self.databases[name] = desc self.databases[name] = desc
async def getSqlor(self,name): async def getSqlor(self,name):
await self.sema.acquire()
desc = self.databases.get(name) desc = self.databases.get(name)
sor = sqlorFactory(desc) sor = sqlorFactory(desc)
sor.name = name sor.name = name
@ -186,6 +191,7 @@ class DBPools:
async def freeSqlor(self,sor): async def freeSqlor(self,sor):
await self._releaseConn(sor.name,sor.conn,sor.cur) await self._releaseConn(sor.name,sor.conn,sor.cur)
self.sema.release()
@asynccontextmanager @asynccontextmanager
async def sqlorContext(self,name): async def sqlorContext(self,name):
@ -193,14 +199,15 @@ class DBPools:
try: try:
yield sqlor yield sqlor
except: except:
if sqlor.dataChanged: if if sqlor and sqlor.dataChanged:
sqlor.rollback() sqlor.rollback()
finally: finally:
if sqlor.dataChanged: if sif sqlor and qlor.dataChanged:
sqlor.commit() sqlor.commit()
await self.freeSqlor(sqlor) await self.freeSqlor(sqlor)
async def _aquireConn(self,dbname): async def _aquireConn(self,dbname):
"""
p = self._cpools.get(dbname) p = self._cpools.get(dbname)
if p == None: if p == None:
p = ConnectionPool(self.databases.get(dbname),self.loop) p = ConnectionPool(self.databases.get(dbname),self.loop)
@ -212,12 +219,26 @@ class DBPools:
else: else:
cur = conn.cursor() cur = conn.cursor()
return self.isAsyncDriver(dbname),conn,cur return self.isAsyncDriver(dbname),conn,cur
"""
dbdesc = self.databases.get(dbname)
driver = myImport(dbdesc['driver'])
conn = None
cur = None
if self.isAsyncDriver(dbname):
conn = await driver.connect(**dbdesc['kwargs'])
cur = await conn.cursor()
return True,conn,cur
else:
conn = driver.connect(**dbdesc.kwargs)
cur = conn.cursor()
return False,conn,cur
def isAsyncDriver(self,dbname): def isAsyncDriver(self,dbname):
ret = self.databases[dbname].get('async_mode',False) ret = self.databases[dbname].get('async_mode',False)
return ret return ret
async def _releaseConn(self,dbname,conn,cur): async def _releaseConn(self,dbname,conn,cur):
"""
if self.isAsyncDriver(dbname): if self.isAsyncDriver(dbname):
await cur.close() await cur.close()
else: else:
@ -230,6 +251,16 @@ class DBPools:
if p == None: if p == None:
raise Exception('database (%s) not connected'%dbname) raise Exception('database (%s) not connected'%dbname)
await p.release(conn) await p.release(conn)
"""
if self.isAsyncDriver(dbname):
await cur.close()
else:
try:
cur.fetchall()
except:
pass
cur.close()
conn.close()
async def useOrGetSor(self,dbname,**kw): async def useOrGetSor(self,dbname,**kw):
commit = False commit = False
@ -343,6 +374,19 @@ class DBPools:
await self.freeSqlor(sor) await self.freeSqlor(sor)
return wrap_func return wrap_func
def setMeta(self,key,n,v):
if not self.meta.get(key):
self.meta[key] = {}
self.meta[key][n] = v
def getMeta(self,key,n=None):
if not self.meta.get(key):
self.meta[key] = {}
if n is None:
return self.meta[key]
return self.meta[key].get(n,None)
async def getTables(self,dbname,**kw): async def getTables(self,dbname,**kw):
@self.inSqlor @self.inSqlor
async def _getTables(dbname,NS,**kw): async def _getTables(dbname,NS,**kw):
@ -354,32 +398,52 @@ class DBPools:
async def getTableFields(self,dbname,tblname,**kw): async def getTableFields(self,dbname,tblname,**kw):
@self.inSqlor @self.inSqlor
async def _getTableFields(dbname,NS,tblname,**kw): async def _getTableFields(dbname,NS,tblname,**kw):
key = '%s:%s' % (dbname,tblname)
fields = self.getMeta(key, 'fields')
if fields:
return fields
sor = kw['sor'] sor = kw['sor']
ret = await sor.fields(tblname) ret = await sor.fields(tblname)
self.setMeta(key,'fields',ret)
return ret return ret
return await _getTableFields(dbname,{},tblname,**kw) return await _getTableFields(dbname,{},tblname,**kw)
async def getTablePrimaryKey(self,dbname,tblname,**kw): async def getTablePrimaryKey(self,dbname,tblname,**kw):
@self.inSqlor @self.inSqlor
async def _getTablePrimaryKey(dbname,NS,tblname,**kw): async def _getTablePrimaryKey(dbname,NS,tblname,**kw):
key = '%s:%s' % (dbname,tblname)
pri = self.getMeta(key,'primarykey')
if pri:
return pri
sor = kw['sor'] sor = kw['sor']
ret = await sor.primary(tblname) ret = await sor.primary(tblname)
self.setMeta(key,'primarykey',ret)
return ret return ret
return await _getTablePrimaryKey(dbname,{},tblname,**kw) return await _getTablePrimaryKey(dbname,{},tblname,**kw)
async def getTableIndexes(self,dbname,tblname,**kw): async def getTableIndexes(self,dbname,tblname,**kw):
@self.inSqlor @self.inSqlor
async def _getTablePrimaryKey(dbname,NS,tblname,**kw): async def _getTablePrimaryKey(dbname,NS,tblname,**kw):
key = '%s:%s' % (dbname,tblname)
idxes = self.getMeta(key,'indexes')
if idxes:
return idxes
sor = kw['sor'] sor = kw['sor']
ret = await sor.indexes(tblname) ret = await sor.indexes(tblname)
self.setMeta(key,'indexes',ret)
return ret return ret
return await _getTablePrimaryKey(dbname,{},tblname,**kw) return await _getTablePrimaryKey(dbname,{},tblname,**kw)
async def getTableForignKeys(self,dbname,tblname,**kw): async def getTableForignKeys(self,dbname,tblname,**kw):
@self.inSqlor @self.inSqlor
async def _getTableForignKeys(dbname,NS,tblname,**kw): async def _getTableForignKeys(dbname,NS,tblname,**kw):
key = '%s:%s' % (dbname,tblname)
forignkeys = self.getMeta(key,'forignkeys')
if forignkeys:
return forignkeys
sor = kw['sor'] sor = kw['sor']
ret = await sor.fkeys(tblname) ret = await sor.fkeys(tblname)
self.setMeta(key,'forignkeys',forignkeys)
return ret return ret
return await _getTableForignKeys(dbname,{},tblname,**kw) return await _getTableForignKeys(dbname,{},tblname,**kw)