概要 Marshmallowは、データのシリアライズとデシリアライズを行うための軽量なライブラリです。このライブラリを使用することで、複雑なORMモデルオブジェクトをPythonのネイティブデータ型に変換することができます。

本記事では、FlaskとSQLAlchemyを使用したプロジェクトでMarshmallowを導入する方法について説明します。

インストール

pip install -U marshmallow-sqlalchemy
pip install -U flask-sqlalchemy
pip install -U flask-marshmallow

初期化

import os

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_session import Session
from flask_migrate import Migrate, MigrateCommand
from flask_jsonrpc import JSONRPC
from flask_marshmallow import Marshmallow

from application.utils import init_blueprint
from application.utils.config import load_config
from application.utils.session import init_session
from application.utils.logger import Log
from application.utils.commands import load_command

# オブジェクトの生成
manager = Manager()
db = SQLAlchemy()
redis = FlaskRedis()
session_store = Session()
migrate = Migrate()
log = Log()
jsonrpc = JSONRPC()
ma = Marshmallow()

def create_app(config_path):
    app = Flask(__name__)
    app.BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

    # 設定の読み込み
    Config = load_config(config_path)
    app.config.from_object(Config)

    # データベースの初期化
    db.init_app(app)
    redis.init_app(app)

    # Marshmallowの初期化
    ma.init_app(app)

    # セッションの初期化
    init_session(app)
    session_store.init_app(app)

    # マイグレーションの初期化
    migrate.init_app(app, db)
    manager.add_command('db', MigrateCommand)

    # ログの初期化
    app.log = log.init_app(app)

    # ブループリントの登録
    init_blueprint(app)

    # JSON-RPCの初期化
    jsonrpc.service_url = "/api"
    jsonrpc.init_app(app)

    # コマンドの登録
    manager.app = app
    load_command(manager)

    return app

使用方法 Marshmallowは、Schemaクラスを使ってデータの形式を変換します。Schemaクラスには、シリアライズ、バリデーション、デシリアライズといった基本的な機能が提供されています。

Schemaを使ったデータのシリアライズ

from marshmallow import Schema, fields
from application.apps.users.models import User, UserProfile

class UserSchema(Schema):
    name = fields.String()
    age = fields.Integer()
    email = fields.Email()
    money = fields.Number()

    class Meta:
        fields = ["name", "age", "money", "email", "info"]
        ordered = True

def index():
    user1 = User(name="xiaoming", password="123456", age=16, email="333@qq.com", money=31.50)
    
    # モデルオブジェクトを辞書に変換
    data1 = UserSchema().dump(user1)
    print(type(data1), data1)

    # モデルオブジェクトをJSON文字列に変換
    data2 = UserSchema().dumps(user1)
    print(type(data2), data2)

    return "ok"

Schemaのフィールドタイプ

Schemaの共通属性

複数のモデルデータのシリアライズ

from marshmallow import Schema, fields
from application.apps.users.models import User, UserProfile

class UserSchema(Schema):
    name = fields.String()
    age = fields.Integer()
    email = fields.Email()
    money = fields.Number()

    class Meta:
        fields = ["name", "age", "money", "email", "info"]
        ordered = True

def index():
    user1 = User(name="xiaoming", password="123456", age=15, email="333@qq.com", money=31.50)
    user2 = User(name="xiaohong", password="123456", age=16, email="333@qq.com", money=31.50)
    user3 = User(name="xiaopang", password="123456", age=17, email="333@qq.com", money=31.50)
    data_list = [user1, user2, user3]
    
    data1 = UserSchema(many=True).dumps(data_list)
    print(type(data1), data1)

    return "ok"

Schemaのネスト使用

from marshmallow import Schema, fields
from application.apps.users.models import User, UserProfile

class UserProfileSchema(Schema):
    education = fields.Integer()
    middle_school = fields.String()

class UserSchema(Schema):
    name = fields.String()
    age = fields.Integer()
    email = fields.Email()
    money = fields.Number()
    info = fields.Nested(UserProfileSchema, only=["middle_school"])

    class Meta:
        fields = ["name", "age", "money", "email", "info"]
        ordered = True

def index():
    user1 = User(name="xiaoming", password="123456", age=15, email="333@qq.com", money=31.50)
    user1.info = UserProfile(education=3, middle_school="qwq学校")

    data = UserSchema().dump(user1)
    print(data)

    data1 = UserSchema().dumps(user1)
    print(data1)

    return "ok"

Schemaを使ったデータのデシリアライズ

from marshmallow import Schema, fields, post_load

class UserSchema(Schema):
    name = fields.String(required=True, error_messages={"required": "必須項目です"})
    sex = fields.String()
    age = fields.Integer(missing=18)
    email = fields.Email(error_messages={'invalid': "有効なメールアドレスを入力してください"})
    mobile = fields.String()

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

def index():
    user_data = {"mobile": "1331345635", "email": "xiaoming@qq.com", "sex": "abc"}
    schema = UserSchema()
    result = schema.load(user_data, partial=True)
    print(result)

    return "ok"

デシリアライズ時のデータの一部変換/無視

from marshmallow import Schema, fields, validate, ValidationError, post_load

class UserSchema(Schema):
    name = fields.String()
    sex = fields.String()
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String(required=True)

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

def index():
    user_data = {"name": "xiaoming", "sex": "abc"}
    schema = UserSchema()
    result = schema.load(user_data, partial=True)
    print(result)

    return "ok"

シリアライズ/デシリアライズ時のフィールドの制御

from marshmallow import Schema, fields, validate, post_load

class UserSchema(Schema):
    name = fields.String()
    sex = fields.Integer(validate=validate.OneOf([0, 1, 2]))
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String()
    password = fields.String(load_only=True)

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

def index():
    user_data = {"name": "xiaoming", "password": "123456", "sex": 1}
    schema = UserSchema()
    result = schema.load(user_data)
    print(result)

    schema = UserSchema(only=["sex", "name", "age", "password"])
    result2 = schema.dump(result)
    print(result2)

    return "ok"

シリアライズ/デシリアライズ時のフックメソッド

from marshmallow import Schema, fields, validate, ValidationError, post_load, post_dump

class UserSchema(Schema):
    name = fields.String()
    sex = fields.Integer(validate=validate.OneOf([0, 1, 2]))
    age = fields.Integer(missing=18)
    email = fields.Email()
    mobile = fields.String()
    password = fields.String(load_only=True)

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

    @post_dump
    def mask_mobile(self, data, **kwargs):
        data["mobile"] = data["mobile"][:3] + "*****" + data["mobile"][-3:]
        return data

def index():
    user_data = {"name": "xiaoming", "password": "123456", "sex": 1, "mobile": "133123454656"}
    schema = UserSchema()
    result = schema.load(user_data)
    print(result)

    schema = UserSchema(only=["sex", "name", "age", "mobile"])
    result2 = schema.dump(result)
    print(result2)

    return "ok"

デシリアライズ時のデータのバリデーション 内蔵バリデーターを使用したバリデーション

from marshmallow import Schema, fields, validate, ValidationError, post_load

class UserSchema(Schema):
    name = fields.String(required=True)
    sex = fields.String(required=True, error_messages={"required": "必須項目です"})
    age = fields.Integer(missing=18, validate=validate.Range(min=18, max=40, error="年齢は18から40の間でなければなりません"))
    email = fields.Email(error_messages={"invalid": "有効なメールアドレスを入力してください"})
    mobile = fields.String(required=True, validate=validate.Regexp("^1[3-9]\d{9}$", error="電話番号の形式が不正です"))

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

def index():
    user_data = {"mobile": "1331345635", "name": "xiaoming", "age": 40, "email": "xiaoming@qq.com", "sex": "abc"}
    schema = UserSchema()
    result = schema.load(user_data)
    result2 = schema.dumps(result)
    print(result)
    print(result2)

    return "ok"

カスタムバリデーションメソッド

from marshmallow import Schema, fields, validate, validates, ValidationError, post_load, validates_schema

class UserSchema(Schema):
    name = fields.String(required=True)
    sex = fields.String(required=True, error_messages={"required": "必須項目です"})
    age = fields.Integer(missing=18, validate=validate.Range(min=18, max=40, error="年齢は18から40の間でなければなりません"))
    email = fields.Email(error_messages={"invalid": "有効なメールアドレスを入力してください"})
    mobile = fields.String(required=True, validate=validate.Regexp("^1[3-9]\d{9}$", error="電話番号の形式が不正です"))
    password = fields.String(required=True, load_only=True)
    password2 = fields.String(required=True, allow_none=True)

    @post_load
    def make_user(self, data, **kwargs):
        return User(**data)

    @validates("name")
    def validate_name(self, value, **kwargs):
        if value == "root":
            raise ValidationError("rootユーザーはスーパーユーザーです。登録できません。")
        return value

    @validates_schema
    def validate_passwords(self, data, **kwargs):
        if data["password"] != data["password2"]:
            raise ValidationError("パスワードと確認用パスワードが一致しません。")
        data.pop("password2")
        return data

def index():
    user_data = {"password": "12345", "password2": "123456", "mobile": "13313345635", "name": "root1", "age": 40, "email": "xiaoming@qq.com", "sex": "abc"}
    schema = UserSchema()
    result = schema.load(user_data)
    print(result)

    return "ok"

モデルスキーマ (ModelSchema) flask-marshmallow 0.12.0以降では、ModelSchemaとTableSchemaは非推奨となり、代わりにSQLAlchemyAutoSchemaとSQLAlchemySchemaが推奨されています。

from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
from application.apps.user.models import User, UserProfile, db

class UserSchema(SQLAlchemySchema):
    username = auto_field("name", dump_only=True)
    created_time = auto_field(format="%Y-%m-%d")
    token = fields.String()

    class Meta:
        model = User
        fields = ["username", "created_time", "token"]

def index():
    from datetime import datetime
    user1 = User(
        name="xiaoming",
        password="123456",
        age=16,
        email="333@qq.com",
        money=31.50,
        created_time=datetime.now(),
    )
    user1.token = "abc"

    data1 = UserSchema().dump(user1)
    print(type(data1), data1)

    return "ok"

from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from application.apps.user.models import User, UserProfile, db

class UserSchema(SQLAlchemyAutoSchema):
    token = fields.String()

    class Meta:
        model = User
        include_fk = False
        include_relationships = False
        fields = ["name", "created_time", "info", "token"]
        sql_session = db.session

def index():
    from datetime import datetime
    user1 = User(
        name="xiaoming",
        password="123456",
        age=16,
        email="333@qq.com",
        money=31.50,
        created_time=datetime.now(),
        info=UserProfile(position="助教")
    )
    user1.token = "abcccccc"

    data1 = UserSchema().dump(user1)
    print(type(data1), data1)

    return "ok"