析言GBI对于数据在本地最佳实践

from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_util_sse import models as util_models
import asyncio
import json

class RunDataAnalysis:
  def __init__(self) -> None:
    self.endpoint = None
    self._client = None
    self._api_info = self._create_api_info()
    self._runtime = util_models.RuntimeOptions(read_timeout=1000 * 100)
    self._init_app()

  def _init_app(self):
    endpoint = 'dataanalysisgbi.cn-beijing.aliyuncs.com'
    access_key_id = 'accessKeyId'
    access_key_secret = 'accessKeySecret'
    assert endpoint is not None and access_key_id is not None and access_key_secret is not None

    self._client = self._create_client(access_key_id, access_key_secret, endpoint)

  def _create_client(
    self,
    access_key_id: str,
    access_key_secret: str,
    endpoint: str,
  ) -> OpenApiClient:
    config = open_api_models.Config(
      access_key_id=access_key_id,
      access_key_secret=access_key_secret
    )
    config.endpoint = endpoint if endpoint is not None else 'dataanalysisgbi.cn-beijing.aliyuncs.com'
    return OpenApiClient(config)

  def _create_api_info(self) -> open_api_models.Params:
    """
    API 相关
    @param path: params
    @return: OpenApi.Params
    """
    params = open_api_models.Params(
      # 接口名称
      action='RunSqlGeneration',
      # 接口版本
      version='2024-08-23',
      # 接口协议
      protocol='HTTPS',
      # 接口 HTTP 方法
      method='POST',
      auth_type='AK',
      style='RPC',
      # 接口 PATH,
      pathname='/gbi/runSqlGeneration',
      # 接口请求体内容格式,
      req_body_type='json',
      # 接口响应体内容格式,
      body_type='sse'
    )
    return params

  async def do_sse_query(self, query: str):
    assert self._client is not None
    assert isinstance(query, str), '"query" is mandatory and should be str'

    body = {
      'specificationType': 'STANDARD_MIX',
      'query': query,
    }
    request = open_api_models.OpenApiRequest(
        body=body,
        query={
          'workspaceId': 'workspaceId'
        }

    )
    sse_receiver = self._client.call_sse_api_async(params=self._api_info, request=request, runtime=self._runtime)
    return sse_receiver

async def query():
    run_data_analysis = RunDataAnalysis()
    frame_count = 0
    async for res in await run_data_analysis.do_sse_query('最畅销的3个商品'):
        try:
            data = json.loads(res.get('event').data)
            print(data)
        except json.JSONDecodeError:
            print('------json.JSONDecodeError--------')
            print(res.get('headers'))
            print(res.get('event').data)
            print('------json.JSONDecodeError-end--------')
            continue
    print('------end--------')

if __name__ == '__main__':
    asyncio.run(query())

import mysql.connector

# 数据库连接配置
config = {
    'user': 'USER',
    'password': 'PASSWORD',
    'host': 'URL',
    'database': 'DATABASE',
    'raise_on_warnings': True
}

# 连接到数据库
cnx = mysql.connector.connect(**config)

# 创建游标对象
cursor = cnx.cursor()

# 执行SQL查询
query = '''select c.`name`, p.product_name, o.* from orders o
			left join customers c on c.customer_id=o.customer_id
			left join products p on p.product_id=o.product_id
		'''
cursor.execute(query)

# 获取查询结果
results = cursor.fetchall()

# 打印结果
for row in results:
    print(row)

# 关闭游标和连接
cursor.close()
cnx.close()

from alibabacloud_tea_openapi_sse.client import Client as OpenApiClient
from alibabacloud_tea_openapi_sse import models as open_api_models
from alibabacloud_tea_util_sse import models as util_models
import asyncio
import json

class RunDataAnalysis:
  def __init__(self) -> None:
    self.endpoint = None
    self._client = None
    self._api_info = self._create_api_info()
    self._runtime = util_models.RuntimeOptions(read_timeout=1000 * 100)
    self._init_app()

  def _init_app(self):
    endpoint = 'dataanalysisgbi.cn-beijing.aliyuncs.com'
    access_key_id = '******'
    access_key_secret = '******'
    assert endpoint is not None and access_key_id is not None and access_key_secret is not None

    self._client = self._create_client(access_key_id, access_key_secret, endpoint)

  def _create_client(
    self,
    access_key_id: str,
    access_key_secret: str,
    endpoint: str,
  ) -> OpenApiClient:
    config = open_api_models.Config(
      access_key_id=access_key_id,
      access_key_secret=access_key_secret
    )
    config.endpoint = endpoint if endpoint is not None else 'dataanalysisgbi.cn-beijing.aliyuncs.com'
    return OpenApiClient(config)

  def _create_api_info(self) -> open_api_models.Params:
    """
    API 相关
    @param path: params
    @return: OpenApi.Params
    """
    params = open_api_models.Params(
      # 接口名称
      action='RunDataResultAnalysis',
      # 接口版本
      version='2024-08-23',
      # 接口协议
      protocol='HTTPS',
      # 接口 HTTP 方法
      method='POST',
      auth_type='AK',
      style='RPC',
      # 接口 PATH,
      pathname='/gbi/runDataResultAnalysis',
      # 接口请求体内容格式,
      req_body_type='json',
      # 接口响应体内容格式,
      body_type='sse'
    )
    return params

  async def do_sse_query(self):
    assert self._client is not None

    self._api_info.pathname='/gbi/runDataResultAnalysis'

    sql_data_data_0 = {
        'month': '10',
        'amount': '111'
    }
    sql_data_data_1 = {
        'month': '12',
        'amount': '100'
    }
    sql_data = {
        'data': [
            sql_data_data_0,
            sql_data_data_1
        ],
        'column': [
            'month',
            'amount'
        ]
    }

    body = {
      'sqlData': sql_data,
      'requestId': 'xxxA8903-776C-5CBA-A6E6-E5xxxx',
      'workspaceId': 'ws_SKuxxxxzWcfx'
    }
    request = open_api_models.OpenApiRequest(
        body=body,
        query={
          'workspaceId': 'ws_SKuqhbsS5zWcfzI2'
        }

    )
    sse_receiver = self._client.call_sse_api_async(params=self._api_info, request=request, runtime=self._runtime)
    return sse_receiver

async def query():
    run_data_analysis = RunDataAnalysis()
    frame_count = 0
    async for res in await run_data_analysis.do_sse_query():
        try:
            data = json.loads(res.get('event').data)
            print(data)
        except Exception as e:
          print(f"Error: {e}")
    print('------end--------')

if __name__ == '__main__':
    asyncio.run(query())