aiohttp-在每个API调用请求之前 [英] aiohttp - before request for each API call
本文介绍了aiohttp-在每个API调用请求之前的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!
问题描述
当我使用Flask时,每个API调用在处理之前都会经过验证:
app = connexion.App(__name__, specification_dir='./swagger/', swagger_json=True, swagger_ui=True, server='tornado')
app.app.json_encoder = encoder.JSONEncoder
app.add_api('swagger.yaml', arguments={'title': 'ABCD API'})
# add CORS support
CORS(app.app)
@app.app.before_request
def before_request_func():
app_id = request.headers.get("X-AppId")
token = request.headers.get("X-Token")
user, success = security.Security().authorize(token)
if not success:
status_code = 401
response = {
'code': status_code,
'message': 'Unauthorized user'
}
return jsonify(response), status_code
g.user = user
当我将其更改为AioHttp,我的身份验证我s设置不正确:
options = {'swagger_path': 'swagger/', "swagger_ui": True}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})
app = web.Application(middlewares=[auth_through_token])
async def auth_through_token(app: web.Application, handler: Any) -> Callable:
@web.middleware
async def middleware_handler(request: web.Request) -> web.Response:
headers = request.headers
x_auth_token = headers.get("X-Token")
app_id = headers.get("X-AppId")
user, success = security.Security().authorize(x_auth_token)
if not success:
return web.json_response(status=401, data={
"error": {
"message": ("Not authorized. Reason: {}"
)
}
})
response = await handler(request)
return response
return middleware_handler
我的请求没有被重定向
有人可以帮助我为每个API设置我的before_request身份验证吗?
谢谢。
Could anyone please help me to set up, my before_request authentication for every API? Thanks.
推荐答案
首先,您必须将 middleware_handler 从中移出auth_through_token 。
然后,
引用您的代码:
Then, Quote your code:
options = {'swagger_path': 'swagger/', "swagger_ui": True}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})
app = web.Application(middlewares=[auth_through_token])
您必须删除最后一行,并将第一行更改为:
You have to remove the last line and change the first line to:
options = {'swagger_path': 'swagger/', "swagger_ui": True, 'middlewares': [middleware_handler]}
所以最终代码应如下所示:
So finally the code should look like:
options = {'swagger_path': 'swagger/', "swagger_ui": True, 'middlewares': [middleware_handler]}
app = connexion.AioHttpApp(__name__, specification_dir='swagger/', options=options)
app.add_api('swagger.yaml', arguments={'title': ' ABCD API'})
@web.middleware
async def middleware_handler(request: web.Request, handler: Any) -> web.Response:
headers = request.headers
x_auth_token = headers.get("X-Token")
app_id = headers.get("X-AppId")
user, success = security.Security().authorize(x_auth_token)
if not success:
return web.json_response(status=401, data={
"error": {
"message": ("Not authorized. Reason: {}"
)
}
})
response = await handler(request)
return response
这篇关于aiohttp-在每个API调用请求之前的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!
查看全文