data-clean/data_clean/api/es_api.py

# -*- coding: utf-8 -*-
# @Time : 2023/8/9 10:41
# @Author : XuJiakai
# @File : es_api
# @Software: PyCharm

from data_clean.exception.fetch_exception import FetchException
from data_clean.utils.async_client import get_aio_elasticsearch

_es = get_aio_elasticsearch()


async def search(index: str, body: dict, doc_type: str = '_doc') -> dict:
    try:
        return await _es.search(index=index, doc_type=doc_type, body=body)
        pass
    except Exception as ex:
        raise FetchException(ex)
    pass


async def get(index: str, id: str, doc_type: str = '_doc') -> dict:
    try:
        return await _es.get(index=index, doc_type=doc_type, id=id)
        pass
    except Exception as ex:
        raise FetchException(ex)
    pass


if __name__ == '__main__':
    pass