data-clean/data_clean/api/hbase_api.py

# -*- coding: utf-8 -*-
# @Time : 2023/8/14 17:54
# @Author : XuJiakai
# @File : hbase_api
# @Software: PyCharm
from data_clean.api.http_api import HttpSessionReuse

from data_clean.env.environment_switch import project_env
from data_clean.exception.fetch_exception import FetchException

_hosts = project_env.get_val('winhc_open_api.eci_data.host')
http_session_reuse = HttpSessionReuse(20)


async def get(table_name: str, rowkey: str):
    result = await http_session_reuse.get(_hosts + '/hbase/get/' + table_name + '/' + rowkey)
    if not result['success']:
        raise FetchException(200, result, '内容异常')
    return result['data']
    pass


async def bulk_get(table_name, rowkey: list, query_key: list = None):
    if query_key is None:
        query_key = []
    rowkey = list(set(rowkey))
    result = await http_session_reuse.post(_hosts + '/hbase/bulk-get?tableName=' + table_name, data={
        "query_key": query_key,
        "rowkey": rowkey
    })
    if not result['success']:
        raise FetchException(200, result, '内容异常')
    r = []
    result = result['data']
    for k in result:
        result[k]['ROWKEY'] = k
        r.append(result[k])
    return r
    pass


if __name__ == '__main__':
    import asyncio

    asyncio.run(get("ng_rt_company", "bc702f0f5202342a9c1c75fbf9be9aff"))
    # asyncio.run(bulk_get("ng_rt_company", ["a33f3cd172f8f9bd61d5f3cd84a4ffd9", "bc702f0f5202342a9c1c75fbf9be9aff"]))
    pass