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}
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
andmarshmallow_sqlalchemy
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 isload_only
value can beTrue/False
.many
: Shows whether this field is a collection of nested object value can beTrue/False
.load
: Field will be deserialized value can beTrue/False
.required
: Field is required, can’t be empty.partial
: Field is optional.
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 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 """
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()
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.']}
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)