https://blog.csdn.net/Likianta/article/details/90123678 https://blog.csdn.net/u012089823/article/details/82258738 https://elasticsearch-py.readthedocs.io/en/master/async.html https://ddeevv.com/docs/elasticsearch/7.9.1/async.html https://elasticsearch-py.readthedocs.io/en/7.9.1/async.html

    安装：

    1. python3 -m pip install elasticsearch[async]

    Fastapi 使用ES示例：

    1. from elasticsearch import AsyncElasticsearch
    2. from fastapi import APIRouter, Query, Path
    3. from fastapi.encoders import jsonable_encoder
    4. from logzero import logger
    5. from pydantic import BaseModel, Field
    7. router = APIRouter()
    8. # amis_action_data es
    9. ES_ONLINE_HOST = "10.00.00.00"
    10. ES_QA_HOST = "10.00.00.00"
    11. ES_AUTH = ('user', 'test')
    12. ES_PORT = 8200
    13. TUIJIAN_INDEX = "test"
    15. # # 本地es
    16. # LOCAL_ES_HOST = "127.0.0.1"
    17. # LOCAL_ES_PORT = 9200
    19. @app.on_event("startup")
    20. async def startup_event():
    21.     app.state.es = AsyncElasticsearch(
    22.         hosts=[ES_QA_HOST], 
    23.         http_auth=ES_AUTH, 
    24.         port=ES_PORT
    25.         )
    27. # 确保AsyncElasticsearch实例被垃圾回收之前close掉，此方法一旦应用程序被关闭时调用
    28. @router.on_event("shutdown")
    29. async def app_shutdown():
    30.     await es.close()
    33. @router.get("/search_es_by_item_id", tags=["QA环境"], summary="通过item_id获取推荐库物料")
    34. async def search_es_by_item_id(
    35.     item_id: str=Query(..., 
    36.                        title="查询物料id", 
    37.                        example="MEETING_9d5c50287"
    38.                                               )):
    39.     body = {
    40.         "query": {
    41.             "match": {
    42.                 "item_id": item_id
    43.             }
    44.         }
    45.     }
    46.     queryData = await app.state.es.search(index=TUIJIAN_INDEX,
    47.                                 doc_type="item_content",
    48.                                 scroll='10m',
    49.                                 size=1,
    50.                                 body=body)
    51.     logger.info(f"queryData: {queryData}")
    52.     return queryData
    55. class UpdateModel(BaseModel):
    56.     item_id: str = Field(..., title="物料id", example="MEETING_9d5c502872d34200")
    57.     update_data: str = Field(..., title="待更新字段", example="title=\"侧丝222\"")
    59. @router.post("/update_field_by_item_id", tags=["QA环境"], summary="根据匹配条件修改某个字段")
    60. async def update_fields(data: UpdateModel):
    61.     result = jsonable_encoder(data)
    62.     body = {
    63.         "query": {
    64.             "match": {
    65.                 "item_id": result.get('item_id')
    66.             }
    67.         },
    68.         'script': {
    69.             "source": f"ctx._source.{result.get('update_data')}"
    70.         }
    71.     }
    72.     update_data = await app.state.es.update_by_query(index=TUIJIAN_INDEX, doc_type="item_content", scroll="10m", body=body)
    73.     if update_data["updated"] == 1:
    74.         return {
    75.             "status_code": 200,
    76.             "msg": "更新成功",
    77.         }
    78.     return {
    79.         "status_code": 422,
    80.         "msg": "更新失败",
    81.     }
    83. @router.delete("/delete_field_by_item_id", tags=["QA环境"], summary="删除某个字段")
    84. async def delete_fields(
    85.     item_id: str=Query(..., title="物料id", example="MEETING_9d5c502872d"),
    86.     delete_data: str=Query(..., title="待删除的字段", example="title")):  
    87.     body = {
    88.         "query": {
    89.             "match": {
    90.                 "item_id": item_id
    91.             }
    92.         },
    93.         "script": {
    94.             "source": f"ctx._source.remove(\"{delete_data}\")"
    95.         }
    96.     }
    97.     delete_data = await app.state.es.update_by_query(index=TUIJIAN_INDEX, doc_type="item_content", scroll="10m", body=body)
    98.     if delete_data["updated"] == 1:
    99.         return {
    100.             "status_code": 200,
    101.             "msg": "删除成功",
    102.         }
    103.     return {
    104.         "status_code": 422,
    105.         "msg": "删除失败",
    106.     }