小编Nap*_*rty的帖子

tl;dr

1. 以下哪个选项是正确的工作流程fastapi？
2. 如何以编程方式测试调用是否真正被阻止（而不是从浏览器手动测试）？是否有对or 的压力测试扩展？uvicornfastapi

我在服务器中有许多端点fastapi（uvicorn目前正在使用），它们对常规同步 Python 代码有长时间的阻塞调用。尽管有文档（https://fastapi.tiangolo.com/async/），我仍然不清楚我是否应该单独使用def或async def混合我的功能。

据我了解，我有三个选择，假设：

def some_long_running_sync_function():
  ...

选项 1 - 始终def仅用于端点

@app.get("route/to/endpoint")
def endpoint_1:
  some_long_running_sync_function()


@app.post("route/to/another/endpoint")
def endpoint_2:
  ...

选项 2 - 始终async def仅使用并在执行器中运行阻塞同步代码

import asyncio


@app.get("route/to/endpoint")
async def endpoint_1:
  loop = asyncio.get_event_loop()
  await loop.run_in_executor(None, some_long_running_sync_function)


@app.post("route/to/another/endpoint")
async def endpoint_2:
  ...

选项 3 - 混合搭配def并async def基于底层调用

import asyncio


@app.get("route/to/endpoint")
def endpoint_1: …

我想以这样的方式从DiGraphinnetworkx中删除节点，即保留所有子节点并将它们的边重新映射（或删除旧边+创建新边）到其父节点的父节点。节点将根据节点属性（例如属于哪个组）被删除。

例如，我希望以编程方式删除具有属性的节点Group B，并将所有直接子节点重新映射到原始父节点（在多个祖父母的情况下，我想映射到所有祖父母）。

我可以想到一种粗略的方法，通过迭代DiGraph，查看其直接后继者和前任者，在它们之间创建关系，然后删除节点，但也许还有更优雅的方法吗？

for node in DiG.nodes(data=True):
  if node[1].get('node_group') == "Group B":
    pre = DiG.predecessors(node[0])
    suc = DiG.successors(node[0])
    for p in pre:
      for s in suc:
        DiG.add_edge(s, p) 
    DiG.remove_node(node[0])