ck 查询

api/api_v1/api.py

@@ -7,6 +7,7 @@ from .endpoints import dashboard
 from .endpoints import report
 from .endpoints import authority
 from .endpoints import table_struct
+from .endpoints import query
 
 api_router = APIRouter()
 
@@ -20,3 +21,5 @@ api_router.include_router(report.router, tags=["报表接口"], prefix='/report'
 api_router.include_router(authority.router, tags=["权限管理接口"], prefix='/authority')
 
 api_router.include_router(table_struct.router, tags=["表结构"], prefix='/table_struct')
+
+api_router.include_router(query.router, tags=["ck"], prefix='/ck')

api/api_v1/endpoints/query.py

@@ -1,11 +1,12 @@
 import json
 
-from aioredis import Redis
+import aioch
+import pandas as pd
 from fastapi import APIRouter, Depends, Request
 import crud, schemas
 
 from api import deps
-from db.redisdb import get_redis_pool
+from db.ckdb import get_ck_db
 
 router = APIRouter()
 
@@ -13,11 +14,24 @@ router = APIRouter()
 @router.post("/sql")
 async def query_sql(
         request: Request,
-        data_in: schemas.GetTable,
-        rdb: Redis = Depends(get_redis_pool),
+        data_in: schemas.Sql,
+        ckdb: aioch.Client = Depends(get_ck_db),
         current_user: schemas.UserDB = Depends(deps.get_current_user)
 ) -> schemas.Msg:
     """原 sql 查询 """
-    data = await rdb.get(f'{data_in.game}_{data_in.name}')
-    return schemas.Msg(code=0, msg='ok', data=json.loads(data))
+    data, columns = await ckdb.execute(data_in.sql, with_column_types=True, columnar=True)
+    df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)})
+    return schemas.Msg(code=0, msg='ok', data=df.to_dict())
 
+
+@router.post("/query")
+async def query(
+        request: Request,
+        data_in: schemas.CkQuery,
+        ckdb: aioch.Client = Depends(get_ck_db),
+        current_user: schemas.UserDB = Depends(deps.get_current_user)
+) -> schemas.Msg:
+    """ json解析 sql 查询"""
+    # data, columns = await ckdb.execute(data_in.sql, with_column_types=True, columnar=True)
+    # df = pd.DataFrame({col[0]: d for d, col in zip(data, columns)})
+    return schemas.Msg(code=0, msg='ok', data=data_in)

api/api_v1/endpoints/table_struct.py

@@ -10,14 +10,14 @@ from db.redisdb import get_redis_pool
 router = APIRouter()
 
 
-@router.post("/event")
-async def read_event(
+@router.post("/")
+async def read_table_struct(
         request: Request,
         data_in: schemas.GetTable,
         rdb: Redis = Depends(get_redis_pool),
         current_user: schemas.UserDB = Depends(deps.get_current_user)
 ) -> schemas.Msg:
-    """获取时间表结构"""
+    """获取表结构"""
     data = await rdb.get(f'{data_in.game}_{data_in.name}')
     return schemas.Msg(code=0, msg='ok', data=json.loads(data))
 

ck_test.py

@@ -7,9 +7,9 @@ from core.config import settings
 
 
 async def exec_progress():
-    client = Client(**settings.CK_CONFIG)
+    client = Client('119.29.176.224')
 
-    progress = await client.execute_with_progress('LONG AND COMPLICATED QUERY')
+    progress = await client.execute_with_progress('show databases')
     timeout = 20
     started_at = datetime.now()
 
@@ -26,10 +26,10 @@ async def exec_progress():
 
 
 async def exec_no_progress():
-    client = Client('localhost')
-    rv = await client.execute('LONG AND COMPLICATED QUERY')
+    client = Client(**settings.CK_CONFIG)
+    rv = await client.execute('show databases')
     print(rv)
 
 
 loop = asyncio.get_event_loop()
-loop.run_until_complete(asyncio.wait([exec_progress(), exec_no_progress()]))
+loop.run_until_complete(asyncio.wait([exec_no_progress()]))

db/ckdb.py

@@ -1,12 +1,12 @@
-from aioredis import create_redis_pool, Redis
+from aioch import Client
 
 
-class RedisBase:
-    client: Redis = None
+class CKBase:
+    client: Client = None
 
 
-rdb = RedisBase()
+ckdb = CKBase()
 
 
-def get_redis_pool() -> Redis:
-    return rdb.client
+def get_ck_db() -> Client:
+    return ckdb.client

db/ckdb_utils.py

@@ -1,14 +1,12 @@
-import aioredis
-from aioredis import create_redis_pool, Redis
+from aioch import Client
 
 from core.config import settings
-from .redisdb import rdb
+from .ckdb import ckdb
 
 
-async def connect_to_redis():
-    rdb.client = await create_redis_pool(**settings.REDIS_CONF)
+async def connect_to_ck():
+    ckdb.client = Client(**settings.CK_CONFIG)
 
 
-async def close_redis_connection():
-    rdb.client.close()
-    await rdb.client.wait_closed()
+async def close_ck_connection():
+    await ckdb.client.disconnect()

main.py

@@ -8,6 +8,7 @@ from starlette.middleware.authentication import AuthenticationMiddleware
 from fastapi_authz import CasbinMiddleware
 
 from db import connect_to_mongo, close_mongo_connection, get_database
+from db.ckdb_utils import connect_to_ck, close_ck_connection
 from db.redisdb_utils import connect_to_redis, close_redis_connection
 from utils import *
 from api.api_v1.api import api_router
@@ -19,9 +20,11 @@ app.include_router(api_router, prefix=settings.API_V1_STR)
 
 app.add_event_handler("startup", connect_to_mongo)
 app.add_event_handler("startup", connect_to_redis)
+app.add_event_handler("startup", connect_to_ck)
 
 app.add_event_handler("shutdown", close_mongo_connection)
 app.add_event_handler("shutdown", close_redis_connection)
+app.add_event_handler("shutdown", close_ck_connection)
 
 
 class CurrentUser(BaseUser):

schemas/__init__.py

@@ -6,4 +6,6 @@ from .space import *
 from .dashboard import *
 from .report import *
 from .authotity import *
-from .table_struct import *
+from .table_struct import *
+
+from .sql import *

schemas/sql.py

@@ -0,0 +1,11 @@
+from typing import List
+
+from pydantic import BaseModel
+
+
+class Sql(BaseModel):
+    sql: str
+
+
+class CkQuery(BaseModel):
+    report_id: List[str]