在Python中动态创建端点

Question

我是测试开发人员，我正在尝试创建基本的HTTP服务器模拟应用程序，可以使用一个"主端点"来生成端点。如。/generate_endpoint。

我将提供url和正文(可能稍后响应代码)到/generate_endpoint，当我调用我创建的endpoint时，它会给我"body"响应。

它必须在不重新启动服务器的情况下工作，因为我想多次使用不同的url和正文。

下面是我尝试的代码。

如果不可能动态创建端点，那么也许你可以给我建议-因为我想创建Mock来测试MyApp和基本工作流程是这样的:

1. 检查订单是否存在(MyApp)
2. MyApp连接到externalApi并检查订单是否存在(我想模拟)
3. MyApp根据externalApi
中给定的值返回值

，但有多个响应(和多个端点)，这可能会发生，我想有测试用例为他们所有，所以我不需要外部应用程序为我的测试。

这是我尝试的:

from fastapi import HTTPException
router = APIRouter()
endpoints = {}

def generate_route(url: str, body: dict):
async def route():
return body
router.get(path=url)(route)
endpoints[url] = body

@router.post("/generate_endpoint")
async def generate_endpoint(endpoint_data: dict):
endpoint_url = endpoint_data.get("url")
endpoint_body = endpoint_data.get("body")
if not endpoint_url or not endpoint_body:
raise HTTPException(status_code=400, detail="url and body required")
generate_route(endpoint_url, endpoint_body)
return {"message": f"route added for url {endpoint_url}"}

或

from flask_restful import Api, Resource, reqparse
app = Flask(__name__)
api = Api(app)
class GenerateEndpoint(Resource):
def post(self):
parser = reqparse.RequestParser()
parser.add_argument("url", type=str)
parser.add_argument("response", type=str)
args = parser.parse_args()
def response():
return args["response"]
api.add_resource(response, args["url"])
return {"status": "success", "url": args["url"]}, 201
api.add_resource(GenerateEndpoint, "/generate_endpoints")
if __name__ == "__main__":
app.run(debug=True)

然后用{"url": "/someurl", "body": "something"}之类的东西测试generate_endpoints然后我期望当我调用GET 127.0.0.1:5000/someurl时，我将有"某事";响应

Answer 1

您可以使用这个变体:

import uvicorn
from fastapi import FastAPI, APIRouter, HTTPException

router = APIRouter()
endpoints = {}

@router.get("/{url:path}")
def get_response(url: str):
if url in endpoints:
return endpoints[url]
raise HTTPException(status_code=404, detail="Not Found")

def generate_route(url: str, body: dict):
endpoints[url] = body

@router.post("/generate_endpoint")
async def generate_endpoint(endpoint_data: dict):
endpoint_url = endpoint_data.get("url")
endpoint_body = endpoint_data.get("body")
if not endpoint_url or not endpoint_body:
raise HTTPException(status_code=400, detail="url and body required")
generate_route(endpoint_url, endpoint_body)
return {"message": f"route added for url {endpoint_url}"}

app = FastAPI()
app.include_router(router)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=5000)