기록/궁금증

[Flask] Flask에 middleware을 넣어보자

그래서그렇지 2022. 8. 23. 14:53

 

규모, 단위가 작은 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)
view raw deco.py hosted with ❤ by GitHub

 

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
view raw auth.py hosted with ❤ by GitHub

 

 

데코레이터로 구현하였을 때 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)
view raw before.py hosted with ❤ by GitHub

 

 

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)
view raw auth.py hosted with ❤ by GitHub

 

 

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
view raw api.py hosted with ❤ by GitHub

 

 

데코레이터와 다르게 직접 함수들에 적용시키는 게 아니라 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)
view raw wsgi.py hosted with ❤ by GitHub

 

 

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)
view raw auth.py hosted with ❤ by GitHub

 

 

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 수정