[Flask] Flask에 middleware을 넣어보자
규모, 단위가 작은 api 개발 시 주로 Flask를 사용한다.
규모가 조금 더 커지면 Express를 사용하는데, 사용하다 보면 middleware의 편리함을 많이 느끼게 된다.
그러면 middleware을 Flask에 넣는 게 가능한지 테스트해보려 한다.
Flask
Flask는 파이썬의 경량 웹 프레임 워크이다.
잘 알려진 Django와의 차이점은 프레임워크에서 기본적으로 지원하는 기능이 적다는 것이고,
기능이 적은 만큼 튜닝 자유도 및 메모리 사용량이 적다는 장점이 있다.
요즘 클라우드에서 작은 단위의 api를 작성하는 일이 많아, 유용하게 사용하고 있다.
Express MiddleWare
express middleware란 아래의 공식문서 정의처럼 요청-응답 사이에서 실행되는 함수이다.
미들웨어 함수는 요청 오브젝트(req), 응답 오브젝트 (res), 그리고 애플리케이션의 요청-응답 주기 중 그다음의 미들웨어 함수 대한 액세스 권한을 갖는 함수입니다. 그다음의 미들웨어 함수는 일반적으로 next라는 이름의 변수로 표시됩니다
주로 사용하는 곳이 요청 전 처리 혹은 요청 후 처리에 주로 사용하면서 사용하면서 편함을 느끼고 있다.
직관적으로 응답 객체 및 요청 객체를 수정 가능하고 요청-응답 주기를 직접 조절이 가능해 Flask에서도 가능할지 궁금해졌다.
Flask MiddleWare
Flask에서 미들웨어를 작성하려고 하였을 때 가장 먼저 떠오른 방법은 데코레이터를 이용하는 방법이었다.
해당 방법으로 구현 후 인터넷에서 조사를 해본 후 두 가지의 방법을 더 찾아내었다.
아래 예제 코드에서는 cors 및 jwt검증을 하는 기능을 middleware으로 작성했다.
1. Decorator
deco.py
from flask import Flask, request, Response | |
import functools | |
from auth import health_cors, jwt_deco | |
app = Flask('App') | |
def abstract_request(func): | |
@functools.wraps(func) | |
def wrapper(*args, **kwargs): | |
return func(request) | |
return wrapper | |
@app.route('/', methods=['GET', 'POST']) | |
@abstract_request | |
@health_cors | |
@jwt_deco | |
def ping(): | |
return Response(response="pong", status=200) | |
if __name__ == "__main__": | |
app.run('127.0.0.1', '5000', debug=True) |
auth.py
import functools | |
from flask import Response, Request | |
import os | |
import jwt | |
def health_cors(func): | |
@functools.wraps(func) | |
def wrapper(req: Request): | |
if(req.method == "GET"): | |
return Response(status=200, response="pong") | |
elif(req.method == 'OPTIONS'): | |
headers = { | |
'Access-Control-Allow-Origin': '*', | |
'Access-Control-Allow-Methods': ["GET","POST"], | |
'Access-Control-Allow-Headers': ['Content-Type', 'Authorization'], | |
'Access-Control-Max-Age': '3600' | |
} | |
return Response(status=204, response="", headers=headers) | |
return func(req) | |
return wrapper | |
def jwt_deco(func): | |
headers = { | |
'Access-Control-Allow-Origin': '*' | |
} | |
@functools.wraps(func) | |
def valid(req: Request): | |
if("Authorization" not in req.headers): | |
return Response(status=401, response="Authorization is not exist", headers=headers) | |
jwt_header = req.headers["Authorization"] | |
try: | |
token_type, token = jwt_header.split(" ") | |
if(token_type.lower() != "bearer"): | |
raise ValueError | |
secret = 'thsissecret' | |
jwt.decode(token, secret, algorithms=["HS256"]) | |
except Exception as e: | |
return Response(status=400, response="Authorization Failed", headers=headers) | |
return func(req) | |
return valid |
데코레이터로 구현하였을 때 request를 직접 전달할 수 있도록 앞에 현재 request객체를 반환하는 함수를 추가해 놓았다.
잘 작동하였다. 데코레이터 순서대로 진행되므로 흐름도 직관적으로 알 수 있었다.
다만 지금은 하나의 함수만 데코레이터를 하는 상황인데
함수가 많아질 경우 데코레이터를 등록하는 귀찮음 및 Human error으로 누락될 경우 찾기 힘들어 보였다.
함수가 많아지면 각각 데코레이터를 등록해야 한다는 것이다. 자동적으로 등록하는 방법이 있으면 좋겠다.
2. flask.blueprint.before_res_fun
Flask에서 지원하는 blueprint에서 before_res_fun이라는 함수를 찾았다.
워낙 작은 api들 작성에 Flask을 이용하다 보니 blueprint는 잘 알지 못했는데, 지식이 늘었다.
before.py
from flask import Flask | |
from api import make_api | |
app = Flask('App') | |
api = make_api() | |
app.register_blueprint(api, url_prefix='/test') | |
if __name__ == "__main__": | |
app.run('127.0.0.1', '5000', debug=True) |
auth.py
from flask import Response, request | |
import jwt | |
def health_cors(): | |
if(request.method == "GET"): | |
return Response(status=200, response="pong") | |
elif(request.method == 'OPTIONS'): | |
headers = { | |
'Access-Control-Allow-Origin': '*', | |
'Access-Control-Allow-Methods': ["GET","POST"], | |
'Access-Control-Allow-Headers': ['Content-Type', 'Authorization'], | |
'Access-Control-Max-Age': '3600' | |
} | |
return Response(status=204, response="", headers=headers) | |
def jwt_deco(): | |
headers = { | |
'Access-Control-Allow-Origin': '*' | |
} | |
if("Authorization" not in request.headers): | |
return Response(status=401, response="Authorization is not exist", headers=headers) | |
jwt_header = request.headers["Authorization"] | |
try: | |
token_type, token = jwt_header.split(" ") | |
if(token_type.lower() != "bearer"): | |
raise ValueError | |
secret = 'thsissecret' | |
jwt.decode(token, secret, algorithms=["HS256"]) | |
except Exception as e: | |
return Response(status=400, response="Authorization Failed", headers=headers) |
api.py
from flask import Blueprint, Response | |
from auth import health_cors, jwt_deco | |
def make_api(): | |
api = Blueprint("test", __name__) | |
api.before_request(health_cors) | |
api.before_request(jwt_deco) | |
@api.route('/foo', methods=['GET', 'POST']) | |
def ping(): | |
return Response(response="pong", status=200) | |
return |
데코레이터와 다르게 직접 함수들에 적용시키는 게 아니라 Blueprint객체에 미리 등록을 하는 방식이었다.
request객체의 경우 상당히 헷갈렸는데, before_request에 함수를 등록할 때 인자를 넘겨줄 방법을 찾지 못해
함수 실행부에서 현재 context에서 request를 찾아서 사용했다.
3. WSGI
마지막 방법은 wsgi을 이용하는 방식이다.
wsgi는 파이썬 스크립트가 웹서버와 통신하기 위한 인터페이스이다.
웹서버에서 wsgi를 통해 웹 애플리케이션(flask)으로 요청을 전달할 때
environ과 start_reponse 두 개의 인자를 건네어서 요청을 처리한다.
environ에는 http 요청에 대한 정보들이 담겨있는 dictonary 타입의 테이터이다.
start_reponse에는 응답에 관한 명세를 넣도록 되어있는 콜백 함수다.
flask에서 wsgi를 이용해 middleware을 작성하려면 http 요청에 대한 정보가 있는 environ을 이용하면 된다.
# environ 데이터
[
'wsgi.version', 'wsgi.url_scheme', 'wsgi.input', 'wsgi.errors',
'wsgi.multithread', 'wsgi.multiprocess', 'wsgi.run_once',
'werkzeug.server.shutdown', 'werkzeug.socket', 'SERVER_SOFTWARE',
'REQUEST_METHOD', 'SCRIPT_NAME', 'PATH_INFO', 'QUERY_STRING',
'REQUEST_URI', 'RAW_URI', 'REMOTE_ADDR', 'REMOTE_PORT', 'SERVER_NAME',
'SERVER_PORT', 'SERVER_PROTOCOL', 'HTTP_HOST', 'HTTP_CONNECTION',
'CONTENT_LENGTH', 'HTTP_SEC_CH_UA_MOBILE', 'HTTP_USER_AGENT',
'HTTP_SEC_CH_UA', 'HTTP_SEC_CH_UA_PLATFORM', 'CONTENT_TYPE',
'HTTP_ACCEPT', 'HTTP_SEC_FETCH_SITE', 'HTTP_SEC_FETCH_MODE',
'HTTP_SEC_FETCH_DEST', 'HTTP_ACCEPT_ENCODING', 'HTTP_ACCEPT_LANGUAGE',
'werkzeug.request'
]
wsgi.py
from flask import Flask, request, Response | |
from auth import corsMiddleware, jwtMiddleware | |
from modify import ModifyMethod, ModifyPostBody | |
app = Flask('App') | |
app.wsgi_app = corsMiddleware(app.wsgi_app) | |
app.wsgi_app = jwtMiddleware(app.wsgi_app) | |
app.wsgi_app = ModifyMethod(app.wsgi_app) | |
app.wsgi_app = ModifyPostBody(app.wsgi_app) | |
@app.route('/', methods=['GET', 'POST']) | |
def hello(): | |
msg = request.get_data().decode("utf-8") + "ok" | |
return Response(response=msg, status=200) | |
if __name__ == "__main__": | |
app.run(host='localhost', port=5050, debug=True) |
auth.py
from werkzeug.wrappers import Request, Response, ResponseStream | |
import jwt | |
class corsMiddleware(): | |
def __init__(self, app): | |
self.app = app | |
def __call__(self, environ, start_response): | |
request = Request(environ) | |
if(request.method == "GET"): | |
res = Response(status=200, response="pong", mimetype= "text/plain") | |
return res(environ, start_response) | |
elif(request.method == 'OPTIONS'): | |
headers = { | |
'Access-Control-Allow-Origin': '*', | |
'Access-Control-Allow-Methods': ["GET","POST"], | |
'Access-Control-Allow-Headers': ['Content-Type', 'Authorization'], | |
'Access-Control-Max-Age': '3600' | |
} | |
res = Response(status=204, response="", headers=headers, mimetype= "text/plain") | |
return res(environ, start_response) | |
return self.app(environ, start_response) | |
class jwtMiddleware(): | |
def __init__(self, app): | |
self.app = app | |
def __call__(self, environ, start_response): | |
request = Request(environ) | |
headers = { | |
'Access-Control-Allow-Origin': '*' | |
} | |
if("Authorization" not in request.headers): | |
res = Response(status=401, response="Authorization is not exist", headers=headers) | |
return res(environ, start_response) | |
jwt_header = request.headers["Authorization"] | |
try: | |
token_type, token = jwt_header.split(" ") | |
if(token_type.lower() != "bearer"): | |
raise ValueError | |
secret = 'thsissecret' | |
jwt.decode(token, secret, algorithms=["HS256"]) | |
except Exception as e: | |
res = Response(status=400, response="Authorization Failed", headers=headers) | |
return res(environ, start_response) | |
return self.app(environ, start_response) |
modify.py
from werkzeug.wrappers import Request, Response | |
import json | |
import io | |
class ModifyPostBody: | |
def __init__(self, app) -> None: | |
self.app = app | |
def __call__(self, environ: dict, start_response): | |
# request = environ["werkzeug.request"] | |
request = Request(environ) | |
if(request.method == "POST"): | |
# 기존의 body 읽어오기 | |
content_length = int(environ["CONTENT_LENGTH"]) | |
row_body: io.BufferedReader = environ['wsgi.input'] | |
body = row_body.read(content_length).decode("utf-8") | |
# 새로운 body 작성 | |
new_body = json.dumps({ | |
"old_body": body, | |
"new_body": { | |
"hello": "world!" | |
} | |
}).encode("utf-8") | |
# 새로 작성한 body를 environ에 다시 넣기 | |
# CONTENT_LENGTH를 수정하지 않는다면 제대로 body가 들어가지 않음 | |
new_body_length = len(new_body) | |
environ["CONTENT_LENGTH"] = new_body_length | |
environ['wsgi.input'] = io.BytesIO(new_body) | |
return self.app(environ, start_response) | |
class ModifyMethod: | |
def __init__(self, app) -> None: | |
self.app = app | |
def __call__(self, environ: dict, start_response): | |
method = environ["REQUEST_METHOD"] | |
if(method == "PUT"): | |
environ["REQUEST_METHOD"] = "POST" | |
elif(method == "PATCH"): | |
environ["REQUEST_METHOD"] = "GET" | |
return self.app(environ, start_response) |
아직 요청이 라우팅 되기 전에 처리하는 방법밖에 찾지 못했다.
라우팅 되고 응답 전에 처리가 가능한 방법이 있으면 추후에 추가하려고 한다.
정리
위의 내용은 요청이 처리되기 전 단계까지만 테스트해본 내용이다.
일단 middleware처럼 작동하는 방법은 찾은 것 같다.
참고 주소
https://spoqa.github.io/2012/05/07/about-flask-request.html
Werkzeug와 Flask의 Request
Werkzeug와 Flask가 HTTP 요청을 어떻게 추상화 하는지를 살펴봅니다.
spoqa.github.io
GitHub 주소
https://github.com/mannamman/namthplaygroundblog/tree/main/question/flask_middleware
# 2022.09.30 WSGI 수정