如何从同一 FastAPI 应用程序中的不同 API 端点调用 API 端点？

Question

（我确实在 SO 上发现了以下问题，但它对我没有帮助：是否可以让一个 api 调用另一个 api，让它们都在同一个应用程序中？）

我正在使用 Fastapi 制作一个具有以下文件夹结构的应用程序

main.py

是应用程序的入口点

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.v1 import lines, upload
from app.core.config import settings

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=0.1,
    openapi_url=f'{settings.API_V1_STR}/openapi.json',
    root_path=settings.ROOT_PATH
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.BACKEND_CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(upload.router, prefix=settings.API_V1_STR)
app.include_router(lines.router, prefix=settings.API_V1_STR)

在

lines.py

中，我有2个GET端点：

• /one-random-line

  --> 从

  .txt

  文件中返回随机行

• /one-random-line-backwards

  --> 应该返回

  /one-random-line

的输出

由于第二个 GET 端点的输出应该是第一个 GET 端点输出的反向字符串，因此我尝试执行here

提到的以下步骤

代码：

import random

from fastapi import APIRouter, Request
from starlette.responses import RedirectResponse

router = APIRouter(
    prefix="/get-info",
    tags=["Get Information"],
    responses={
        200: {'description': 'Success'},
        400: {'description': 'Bad Request'},
        403: {'description': 'Forbidden'},
        500: {'description': 'Internal Server Error'}
    }
)


@router.get('/one-random-line')
def get_one_random_line(request: Request):
    lines = open('netflix_list.txt').read().splitlines()
    if request.headers.get('accept') in ['application/json', 'application/xml']:
        random_line = random.choice(lines)
    else:
        random_line = 'This is an example'
    return {'line': random_line}


@router.get('/one-random-line-backwards')
def get_one_random_line_backwards():
    url = router.url_path_for('get_one_random_line')
    response = RedirectResponse(url=url)
    return {'message': response[::-1]}

当我这样做时，我收到以下错误：

TypeError: 'RedirectResponse' object is not subscriptable

当我将第二个 GET 端点的

return

更改为

return {'message': response}

时，我得到以下输出

我犯了什么错误？

示例：

如果

/one-random-line

端点的输出是'Maverick'，那么

/one-random-line-backwards

的输出应该是'kcirevam'

Answer 1

您可以直接从代码中调用任何端点作为函数调用，您不必处理

RedirectResponse()

或任何东西。下面是一个示例，展示了它的外观和按原样运行：

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/one-random-line")
async def get_one_random_line(request: Request):
    # implement your own logic here, this will only return a static line
    return {"line": "This is an example"}


@app.get("/one-random-line-backwards")
async def get_one_random_line_backwards(request: Request):
    # You don't have to do fancy http stuff, just call your endpoint:
    one_line = await get_one_random_line(request)
    return {"line": one_line["line"][::-1]}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

使用

curl

我们得到以下结果：

% curl localhost:8000/one-random-line          
{"line":"This is an example"}%     
% curl localhost:8000/one-random-line-backwards
{"line":"elpmaxe na si sihT"}%  

Answer 2

重构您的代码，将公共部分作为您调用的函数 - 通常将其放在控制器外部的模块中。

# this function could live as LineService.get_random_line for example
# its responsibility is to fetch a random line from a file
def get_random_line(path="netflix_list.txt"):
    lines = open(path).read().splitlines()
    return random.choice(lines)


# this function encodes the rule that "if the accepted response is json or xml
# we do the random value, otherwise we return a default value"
def get_random_or_default_line_for_accept_value(accept, path="netflix_list.txt", default_value="This is an example"):
    if accept not in ("application/json", "application/xml"):
        return default_value

    return get_random_line(path=path)


@router.get('/one-random-line')
def get_one_random_line(request: Request):
    return {
        "line": get_random_or_default_line_for_accept_value(
            accept=request.headers.get('accept'),
        ),
    }


@router.get('/one-random-line-backwards')
def get_one_random_line_backwards(request: Request):
    return {
        "line": get_random_or_default_line_for_accept_value(
            accept=request.headers.get('accept'),
        )[::-1],
    }

Answer 3

实际上，可以从该 API 的另一个端点调用该 API 端点。您只需要从另一个线程调用它即可。因此，为调用创建一个函数，并从端点内部使用类似的东西：

thread = threading.Thread(target=request_foo, args=(arg1, arg2))
thread.start()

只知道这是一种不好的做法。最好在外部文件中创建共享代码并从您想要的任何端点使用它。