30 Jan 2017

Chapter 3: Flask: Serialization, Deserialization and Validation

Serialization, Deserialization and Validation

What is Serialization?

Serialization is a process of converting complex python objects into flat structure consisting of only native python datatypes.

class User(object):
""" Complex Object in python """
	name = 'Saurabh'
	age = 24
	
user = User()

""" Serialized object """

{'name': 'Saurabh', 'age': 24}

What is Deserialization?

Deserialization is the inverse process of converting serialized data back into python objects.

In this chapter we will learn how we can convert complex data types, such as objects, to and from native Python datatypes and use HTTP requests to create a crud for our models

Note: we will use a library called marshmallow for this, you need to install marshmallow, flask_marshmallow and marshmallow_sqlalchemy

Schema’s

We will create schema’s for our model which will help us to serialize, deserialize and validate data before storing in our database.

from flask_marshmallow import Marshmallow
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields

ma = Marshmallow(app)


class UserSchema(ModelSchema):

    class Meta:
        model = User
        exclude = ('password',)

    id = fields.Integer(dump_only=True)
    email = fields.Email(require=True)
    mobile_number = fields.String(require=True)
    user_profile = fields.Nested('UserProfileSchema', many=False, load=True)


class UserProfileSchema(ModelSchema):

    class Meta:
        model = UserProfile
        exclude = ('user',)

    id = fields.Integer(load=True, partial=True)
    first_name = fields.String()
    last_name = fields.String()

Here we have created two schema’s for our User and UserProfile models respectively. The schema’s are self explanatory. We have created a class User which inherits from ModelSchema, it contains a metaclass where you can define with which model to bind and which fields to include or exclude.

The field id in user model will be validated as number and serialized or deserialized using integer. The field email in user model will be validated as email and serialized or deserialized using string.

Note: email field will be validated for a email value. Example: ‘email’ or ‘email.com’ will fail but ‘email@example.com’ will pass.

The fields user_profile is a nested field which represents a relation between User and UserProfile.

dump_only: Field will only be serialized and not deserialized, opposite of this is load_only value can be True/False. many: Shows whether this field is a collection of nested object value can be True/False. load: Field will be deserialized value can be True/False. required: Field is required, can’t be empty. partial: Field is optional.

Final Code

import os
import re

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import func, select, UniqueConstraint, Index
from flask_marshmallow import Marshmallow
from marshmallow import fields
from marshmallow_sqlalchemy import ModelSchema


basedir = os.path.abspath(os.path.dirname(__file__))
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///{}'.format(os.path.join(basedir, 'test.db'))
db = SQLAlchemy(app)
ma = Marshmallow(app)


def to_underscore(name):

    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()


class BaseMixin(object):

    @declared_attr
    def __tablename__(self):
        return to_underscore(self.__name__)

    id = db.Column(db.Integer, primary_key=True, index=True)
    created_on = db.Column(db.TIMESTAMP, default=db.func.current_timestamp())
    updated_on = db.Column(db.TIMESTAMP, onupdate=db.func.current_timestamp())


class ReprMixin(object):

    __repr_fields__ = ['id', 'name']

    def __repr__(self):
        fields = {f: getattr(self, f, '<BLANK>') for f in self.__repr_fields__}
        pattern = ['{0}=}'.format(f) for f in self.__repr_fields__]
        pattern = ' '.join(pattern)
        pattern = pattern.format(**fields)
        return '<{} {}>'.format(self.__class__.__name__, pattern)


class User(db.Model, BaseMixin, ReprMixin):
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password = db.Column(db.String(255))
    active = db.Column(db.Boolean(), default=False)
    last_login_at = db.Column(db.DateTime())
    mobile_number = db.Column(db.String(10), unique=True, index=True)

    roles = db.relationship('Role', back_populates='users', secondary='user_role')
    user_profile = db.relationship("UserProfile", back_populates="user",
                                   uselist=False, cascade='all, delete-orphan',
                                   lazy='select')
    comments = db.relationship('Comment', back_populates='commenter', uselist=True,
                               lazy='dynamic')
    ratings = db.relationship('UserRating', back_populates='rater', uselist=True,
                              lazy='dynamic')

    @hybrid_property
    def name(self):
        return '{}'.format(self.first_name) + (' {}'.format(self.last_name) if
                                               self.last_name else '')


class UserProfile(db.Model, BaseMixin, ReprMixin):
    __repr_fields__ = ['id', 'first_name']

    first_name = db.Column(db.String(40), nullable=False)
    last_name = db.Column(db.String(40))
    profile_picture = db.Column(db.Text())
    bio = db.Column(db.Text())
    date_of_birth = db.Column(db.Date)
    gender = db.Column(db.Enum('male', 'female', 'other', name='varchar'))
    marital_status = db.Column(db.Enum('single', 'married', 'divorced', 'widowed', name='varchar'))
    education = db.Column(db.Enum('undergraduate', 'graduate', 'post_graduate', name='varchar'))

    user_id = db.Column(db.Integer, db.ForeignKey('user.id', ondelete='CASCADE'), unique=True, index=True)

    user = db.relationship('User', back_populates="user_profile", single_parent=True)


class Role(db.Model, BaseMixin, ReprMixin):

    name = db.Column(db.String(80), unique=True)
    description = db.Column(db.Text, unique=True)

    users = db.relationship('User', secondary='user_role', back_populates='roles')


class UserRole(db.Model, BaseMixin, ReprMixin):
    __repr_fields__ = ['user_id', 'role_id']

    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    role_id = db.Column(db.Integer, db.ForeignKey('role.id'))

    role = db.relationship('Role', foreign_keys=[role_id])
    user = db.relationship('User', foreign_keys=[user_id])

    UniqueConstraint(role_id, user_id, 'role_user_un')


class Post(db.Model, BaseMixin, ReprMixin):
    __repr_fields__ = ['id', 'slug']

    slug = db.Column(db.String(55), unique=True, nullable=False, index=True)
    title = db.Column(db.String(255), nullable=False, index=True)
    data = db.Column(db.Text, nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey('user.id'), index=True)

    author = db.relationship('User', single_parent=True, foreign_keys=[author_id])
    ratings = db.relationship('UserRating', back_populates='post', uselist=True,
                              lazy='dynamic')
    comments = db.relationship('Comment', back_populates='post', uselist=True,
                               lazy='dynamic')

    @hybrid_property
    def avg_rating(self):
        return self.ratings.with_entities(func.Avg(UserRating.rating)).filter(UserRating.post_id == self.id).scalar()

    @hybrid_property
    def total_comments(self):
        return self.comments.with_entities(func.Count(Comment.id)).filter(Comment.post_id == self.id).scalar()

    @avg_rating.expression
    def avg_rating(cls):
        return select([func.Avg(UserRating.rating)]).where(cls.id == UserRating.post_id).as_scalar()


class Comment(db.Model, BaseMixin, ReprMixin):
    __repr_fields__ = ['id', 'commented_by']

    data = db.Column(db.Text, nullable=False)
    is_moderated = db.Column(db.Boolean(), default=False)

    post_id = db.Column(db.Integer, db.ForeignKey('post.id'), index=True)
    commented_by = db.Column(db.Integer, db.ForeignKey('user.id'), index=True)
    parent_comment_id = db.Column(db.Integer, db.ForeignKey('comment.id'), index=True)

    post = db.relationship('Post', foreign_keys=[post_id], back_populates='comments')
    commenter = db.relationship('User', foreign_keys=[commented_by], back_populates='comments')
    parent_comment = db.relationship('Comment', remote_side='Comment.id')
    children_comment = db.relationship('Comment', remote_side='Comment.parent_comment_id')


class UserRating(db.Model, BaseMixin, ReprMixin):
    __repr_fields__ = ['rating', 'post_id', 'rated_by']

    rating = db.Column(db.SmallInteger, nullable=False)

    rated_by = db.Column(db.Integer, db.ForeignKey('user.id'), index=True)
    post_id = db.Column(db.Integer, db.ForeignKey('post.id'), index=True)

    post = db.relationship('Post', back_populates='ratings', foreign_keys=[post_id])
    rater = db.relationship('User', foreign_keys=[rated_by], back_populates='ratings')

    UniqueConstraint(rated_by, post_id, 'user_post_un')


class UserSchema(ModelSchema):
    class Meta:
        model = User
        exclude = ('password',)

    id = fields.Integer(load=True, partial=True)
    email = fields.Email(require=True)
    mobile_number = fields.String(require=True)
    user_profile = fields.Nested('UserProfileSchema', many=False, load=True)


class UserProfileSchema(ModelSchema):
    class Meta:
        model = UserProfile
        exclude = ('user',)

    id = fields.Integer(dump_only=True)
    first_name = fields.String()
    last_name = fields.String()


@app.route('/')
def hello_world():
    return 'Hello, World!'

Serializing an Object

Serializing a python object can be serialized into a flat structure and then dumped into json to sent back in a http call.

>> user = User.query.first()
>> print(user)
>> <User id=1 name=Linus Torvalds>
>> user_data, errors = UserSchema().dump(user)
>> print(user_schema)
>>{'active': None,
 'comments': [],
 'created_on': None,
 'email': 'email@example.com',
 'id': 1,
 'last_login_at': None,
 'mobile_number': '99999999999',
 'ratings': [],
 'roles': [],
 'updated_on': None,
 'user_profile': {
	'bio': None,
	'created_on': None,
	'date_of_birth': None,
	'education': None,
	'first_name': 'Linus',
	'gender': None,
	'id': 1,
	'last_name': 'Torvalds',
	'marital_status': None,
	'profile_picture': None,
	'updated_on': None
	}
 }

""" A python dict representation of the user object """

Deserialization data to a python object

A flat data can be converted or deserialized into python object and persisted in database.

>> user_data = {'email': 'email2@example.com',
   'mobile_number': '1234567890',
   'user_profile': {
	   'first_name': 'Richard',
	    'last_name': 'Stallman'
	    }
   }
>> user_2, errors = UserSchema().load(user_data)
>> print(user_2)
>> <User id=None name=Richard Stallman>

""" A python object of class User created from a dict by deserializing.
	We can persist this newly created object """

>> db.session.add(user_2)
>> db.session.commit()

Validating using Schema’s

We can user schema to valid the data before adding for updating a resource.

>> user_data = {'email': 'email3.com', 'mobile_number': 11, 'user_profile': {'first_name': 'Aaron', 'last_name': 'Swartz'}}

>> user_3, errors = UserSchema().load(user_data)
>> print(errors)
>> {'email': ['Not a valid email address.'],
 'mobile_number': ['Not a valid string.']}

Creating a CRUD for User Model

We will create a CRUD (Create, Read, Update, Delete) for our user model.

@app.route('/users', methods=['GET', 'POST'])
def users_view():
    if request.method == 'GET':
        users = User.query.all()
        users_data = UserSchema().dump(users, many=True).data
        return make_response(jsonify(users_data), 200)

    else:
        users, errors = UserSchema().load(request.json, many=True)
        if errors:
            return make_response(jsonify(errors), 400)
        else:
            db.session.add_all(users)
            db.session.commit()
            return make_response(jsonify(UserSchema().dump(users, many=True).data), 201)


@app.route('/user/<int:slug>', methods=['GET', 'PUT', 'DELETE'])
def user_view(slug):
    user = User.query.get(slug)
    if not user:
        return make_response(jsonify({'error': 'Resource not found'}), 404)
    if request.method == 'GET':
        return make_response(jsonify(UserSchema().dump(user).data), 200)
    if request.method == 'PUT':
        user, errors = UserSchema().load(request.json, instance=user)
        if errors:
            return make_response(jsonify(errors), 400)
        db.session.commit()
        return make_response(jsonify(UserSchema().dump(user).data), 200)
    if request.method == 'DELETE':
        db.session.delete(user)
        db.session.commit()
        return make_response(jsonify({}), 204)

Requests:

GET: ‘/users’ will return all the users.

POST: ‘/users’ takes array of user in json and creates new users.

GET: ‘/user/1’ will return first user.

PUT: ‘/user/1’ takes user data and will update user with id 1

DELETE: ‘/user/1’ deletes user with id 1.

import requests

url = "http://127.0.0.1:5000/users"

""" Get request to fetch all users"""
response = requests.request("GET", url) 
print(response.text)


payload = [{
   "email": "email3@example.com",
   "mobile_number": "123",
   "user_profile": {
      "first_name": "Arron",
      "last_name": "Swart"
   }
 }]
headers = {
    'content-type': "application/json"
    }


""" POST request to create multiple users"""
response = requests.request("POST", url, data=json.dumps(payload), headers=headers)
print(response.text)



url = "http://127.0.0.1:5000/user/3"
payload = {
    "email": "email3@example.com",
    "mobile_number": "123",
    "user_profile": {
    	"id": 5,
        "first_name": "Arron",
        "last_name": "Swart"
    }
 }

""" PUT request to update existing user """
response = requests.request("PUT", url, data=payload, headers=headers)
print(response.text)


""" DELETE request to delete a user """
response = requests.request("DELETE", url, headers=headers)
print(response.text)

Stats:

0 comments