Flask でフォーム情報を受け取る方法 (jsonまたxmlでも可)
概要
前回 json or xml どちらでも受け取れるような仕組みを考えてみました
今回はさらに form でも受け取れるような API に変更します
環境
- macOS 13.4
- Python 3.11.3
- Flask 2.3.2
サンプルコード
レスポンスの変換に関するコードも含まれています
ポイントは Convert クラスの is_form と from_form です
特定のキーが含まれているかでフォームかどうかを判断するようにしています
flask で入れ子なフォームデータを受け取ると Immutablemultidict がなぜか入れ子にならずそのままキー名として使われるようです
リクエスト情報を Converter.to_dict するとどんなデータでも dict に変換してくれるイメージです
import copy
import json
import traceback
from json.decoder import JSONDecodeError
from xml.parsers.expat import ExpatError
import dicttoxml
import xmltodict
from flask import Flask, Response, jsonify, request
app = Flask(__name__)
class Profile:
def __init__(self) -> None:
self.langs = ["ruby", "python", "swift"]
def to_dict(self):
return self.langs
def set_item_name(_: str) -> str:
return "lang"
class User:
def __init__(self, name: str = "default_user", **_) -> None:
self.name = name
self.age = 10
self.profile = Profile()
def to_dict(self):
return {"name": self.name, "age": self.age, "profile": self.profile.to_dict()}
def convert_obj(func):
def wrapper(*args, content_dict: dict, **kwargs):
try:
user = User(**content_dict["xml"])
except KeyError:
raise ValueError("Root attribute must start with 'xml'.")
return func(*args, user=user, **kwargs)
return wrapper
class Converter:
def __init__(self, body: bytes, form: dict) -> None:
self.body = body
self.form = form
def to_dict(self) -> dict:
if self._is_xml():
return self._from_xml()
elif self._is_json():
return self._from_json()
elif self._is_form():
return self._from_form()
return {}
def _is_xml(self) -> bool:
try:
self._from_xml()
except ExpatError:
return False
return True
def _is_json(self) -> bool:
try:
self._from_json()
except JSONDecodeError:
return False
return True
def _is_form(self) -> bool:
try:
self._from_form()
except KeyError:
return False
return True
def _from_xml(self) -> dict:
return xmltodict.parse(self.body)
def _from_json(self) -> dict:
return json.loads(self.body)
def _from_form(self) -> dict:
return {"xml": {"name": self.form["xml[name]"]}}
def convert_dict(func):
def wrapper(*args, **kwargs):
body = copy.copy(request.data)
form = copy.copy(request.form)
data = Converter(body, form).to_dict()
if not data:
raise ValueError("You must be specified request body xml, json or form.")
return func(*args, content_dict=data, **kwargs)
return wrapper
@app.route("/", methods=["GET", "POST"])
@convert_dict
@convert_obj
def parse_xml(user: User):
return jsonify(user.to_dict())
@app.after_request
def after_request(response: Response):
accept_header = request.headers.get("Accept")
if accept_header == "application/xml":
xml = dicttoxml.dicttoxml(
response.get_json(),
attr_type=True,
root=True,
custom_root="orgRoot",
item_func=set_item_name,
)
response.headers["Content-Type"] = "applicatoin/xml"
response.data = xml
elif accept_header == "application/json":
pass
return response
@app.errorhandler(ValueError)
def handle_bad_request(e):
return traceback.format_exception_only(e), 400
動作確認
curl localhost:5000 -XPOST -d '{"xml": {"name":"hawksnowlog"}}' -H 'content-type: application/json'
curl localhost:5000 -XPOST -d '<xml><name>hawksnowlog</name></xml>' -H 'content-type: application/xml'
curl localhost:5000 -XPOST -d "xml[name]=hawksnowlog" -H 'content-type: application/x-www-form-urlencoded'
結果はすべて同じになります
