Photo by Ray Hennessy / Unsplash

Background

marshmallow 是一個用來將 ORM Object 與 Python 原生的 object 互相轉換的 Library. (object -> dict, objects -> list, string -> dict, string -> list) 這個套件其實可以幫助你提升開API的品質. 這篇文章主要就介紹&記錄基本 marshmallow 的使用方法.

Installation

$ pip3 install -U marshmallow

marshmallow 的基本元素

簡單介紹一下在 marshmallow 中常見的元素

schema

要實現數據物件之間的轉換, 那就必須要一個中間的載體. 在 marshmallow 中稱為 Schema. 因此我們常會在定義 model 時順便會定義 schema.

field

通常我們會用多個 marshmallow field 來組成一個 marshmallow schema. 基本上在跟你創立 ORM 的 Model 想法很像. marshmallow 的 field 有以下幾種常見的類型:

  • Str, Int, Bool, Float
  • DateTime, LocalDateTim
  • Email
  • Nested

Validators

marshmallow 內建的驗證器, 可用來驗證 field 是不是符合格式.

marshmallow 的名詞介紹

Serializing

marshmallow 提供 dump() or dumps() 的 funnction.

  • dump(): obj -> dict
  • dumps(): obj -> string

Deserializing

相對於 dump() function, marshmallow 也有提供 load() & loads() funtion 來提供數據轉換. 在 Marshmallow 中, dict -> object 的方法需要自己實做, 在該方法前面加上一個decoration:post_load 即可.

Example

接下來會利用 Flask 來實做一個簡易的 Restful api. (裡面會包含 marshmallow)

Install

$ mkdir marshmallow
$ cd marshmallow
$ virtualenv .venv
$ source .venv/bin/activate
$ vim requirements.txt

requirements.txt

aniso8601==4.0.1
Click==7.0
Flask==1.0.2
flask-marshmallow==0.9.0
Flask-RESTful==0.3.7
Flask-SQLAlchemy==2.3.2
itsdangerous==1.1.0
Jinja2==2.10
MarkupSafe==1.1.0
marshmallow==2.17.0
pytz==2018.7
six==1.12.0
SQLAlchemy==1.2.15
Werkzeug==0.14.1
$ pip3 install -r requirements.txt

Initialization

$ mkdir src tests src/models
$ touch README.md run.py config.py tests/__init__.py src/__init__.py src/models/__init__.py

start

設定&安裝完基本的東西之後 接下來來完成程式的部分

config.py

class Config(object):
    SQLALCHEMY_TRACK_MODIFICATIONS = True

class ProductionConfig(Config):
    DEBUG = False
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///../flask.db'

class DevelopmentConfig(Config):
    DEVELOPMENT = True
    DEBUG = True
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///../flask.db'

class TestingConfig(Config):
    TESTING = True
    DEBUG = True
    DOMAIN = 'http://127.0.0.1:5000'
    SQLALCHEMY_DATABASE_URI = 'sqlite:///../flask.db'

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,

    'default': DevelopmentConfig
}

rum.py

from src import app, db
from config import config
from src.models import users
import os

@app.cli.command()
def renewdb():
    """Remove the database."""
    click.echo('Renew the db')
    db.reflect()
    db.drop_all()
    db.create_all()

@app.cli.command()
def test():
    """Remove the database."""
    with app.app_context():
        db.reflect()
        db.drop_all()
        db.create_all()
        db.session.commit()

    os.system('pytest')

if __name__ == '__main__':
    app.run()

create model

src/model/users.py

from flask import Flask
from marshmallow import Schema, fields, pre_load, validate
from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy
from src import db, ma

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(250), nullable=False, unique=True)
    password = db.Column(db.String(250), nullable=False)
    created_date = db.Column(db.TIMESTAMP, server_default=db.func.current_timestamp(), nullable=False)
    status = db.Column(db.Integer)

    def __init__(self, email, password):
        self.email = email
        self.password = password
        self.status = 0

class UserSchema(ma.Schema):
    id = fields.Integer(dump_only=True)
    email = fields.Email(required=True)
    password = fields.String(required=True, validate=validate.Length(6))
    created_date = fields.DateTime()
    status = fields.Integer()

create api

src/__init__.py

import os
from config import config
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow

app = Flask(__name__)
app.config.from_object(config['default'])
db = SQLAlchemy(app)
ma = Marshmallow()

from src import user

src/user.py

from flask import jsonify, redirect, request
from flask_restful import Resource, Api
from src import app
from src import db
from src.models.users import User, UserSchema

api = Api(app)
users_schema = UserSchema(many=True)
user_schema = UserSchema()

class RUser(Resource):
    # endpoint to show all users
    def get(self):
        users = User.query.all()
        users = users_schema.dump(users).data
        return jsonify({'err': 0, 'rows': users})

    def post(self):
        json_data = request.get_json(force=True)
        if not json_data:
            return jsonify({'err': -1, 'err_msg': 'No input data provided'}), 400

        # Validate and deserialize input
        data, errors = user_schema.load(json_data)
        if errors:
            return jsonify({"err": -1, "err_msg": errors}), 422

        user = User(data['email'], data['password'])
        db.session.add(user)
        db.session.commit()

        return jsonify({'err': 0, 'err_msg': 'Sucessful'})
    
    def put(self):
        json_data = request.get_json(force=True)
        if not json_data:
            return jsonify({'err': -1, 'err_msg': 'No input data provided'}), 400

        # Validate and deserialize input
        data, errors = user_schema.load(json_data)
        if errors:
            return jsonify({"err": -1, "err_msg": errors}), 422

        print(data)
        user = User.query.filter_by(id=json_data['id']).first()

        if not user:
            return jsonify({"err": -1, "err_msg": "user is not find"}), 422

        user.email = data['email']
        user.password = data['password']

        db.session.commit()
        return jsonify({'err': 0, 'err_msg': 'Sucessful'})


api.add_resource(RUser, '/user')

create unit test

tests/conftest.py

import pytest
from src import app, db

@pytest.fixture(scope='session')
def app_client():
    print("app_client")
    return app.test_client()

tests/test_user.py

import json
import pytest

def test_post(app_client):

    data = {
        'email': '[email protected]',
        'password': 'paulpassword',
    }

    response = app_client.post('/user', data=json.dumps(data))
    #print(json.loads(response.get_data(as_text=True)))
    res = json.loads(response.get_data(as_text=True))
    assert res['err'] == 0


def test_get(app_client):

    data = {
        'id': 0,
    }

    response = app_client.get('/user', query_string=data)
    #print(response.get_data(as_text=True))
    res = json.loads(response.get_data(as_text=True))
    assert res['err'] == 0


def test_put(app_client):

    data = {
        'id': 1,
        'email': '[email protected]',
        'password': 'paulpassword2'
    }

    response = app_client.put('/user', data=json.dumps(data))

    print(response.get_data(as_text=True))
    res = json.loads(response.get_data(as_text=True))
    assert res['err'] == 0

run

程式都完成之後, 整體的目錄架構如下:

$ tree
.
├── README.md
├── config.py
├── flask.db
├── requirements.txt
├── run.py
├── src
│   ├── __init__.py
│   ├── models
│   │   ├── __init__.py
│   │   └── users.py
│   └── user.py
└── tests
    ├── __init__.py
    ├── conftest.py
    └── test_user.py

執行Test看一下相關的結果.

$ export FLASK_APP=./run.py
$ flask test
===================== test session starts ===========================
platform darwin -- Python 3.7.0, pytest-4.0.2, py-1.7.0, pluggy-0.8.0
rootdir: /Users/taiker/workspace/marshmallow, inifile:
collected 3 items

tests/test_user.py ...                                                                                                                                          [100%]

==================== 3 passed in 0.13 seconds ========================

如果你想看詳細的結果可以查看 sqlite db (flask.db) 或是 assert Flase 在 test_user.py 中.

Reference