leguadmin/xbackend
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

ck 连接池

Browse Source
This commit is contained in:
wuaho 2021-07-27 16:03:33 +08:00
parent db2262ec0b
commit 773e1fac82
2 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
db/ckdb.py
View File

@ -5,15 +5,22 @@ import pandas as pd
class CKDrive: class CKDrive:
client: Client = None ClientPool = dict()
@classmethod
async def _execute(cls, *args, **kwargs):
key, client = cls.ClientPool.popitem()
res = await client.execute(*args, **kwargs)
CKDrive.ClientPool[key] = client
return res
async def execute(self, sql) -> dict: async def execute(self, sql) -> dict:
data, columns = await self.client.execute(sql, with_column_types=True, columnar=True) data, columns = await self._execute(sql, with_column_types=True, columnar=True)
df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)}) df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)})
return df.T.to_dict() return df.T.to_dict()
async def query_dataframe(self, sql): async def query_dataframe(self, sql):
data, columns = await self.client.execute(sql, with_column_types=True, columnar=True) data, columns = await self._execute(sql, with_column_types=True, columnar=True)
df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)}) df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)})
return df return df

6
db/ckdb_utils.py
View File

@ -4,8 +4,10 @@ from core.config import settings
from .ckdb import CKDrive from .ckdb import CKDrive
async def connect_to_ck(): async def connect_to_ck(pool_size=100):
CKDrive.client = Client(**settings.CK_CONFIG) for i in range(pool_size):
client = Client(**settings.CK_CONFIG)
CKDrive.ClientPool[i] = client
async def close_ck_connection(): async def close_ck_connection():