20 Apr 2017

Chapter 4: Flask: Structurizing Large App's

Structuring Apps in Flask

In this chapter we will learn how to structurize a flask app, structurizing is a very important step as it makes the code more manageable, easier to navigate and debug, and allows multiple developers to work on simultaneously on the app.

tutorial-series
 |
 +--file manager.py
 |
 +--file requirements.txt
 |    
 +--dir src
 |  |
 |  +--file __init__.py
 |  +--dir utils
 |  |   +-- file __init__.py
 |  |   +-- file schemas.py
 |  |   +-- file factory.py
 |  |   +-- file api.py
 |  |   \-- file models.py
 |  |
 |  +--dir users
 |  |   +-- file __init__.py
 |  |   +-- file schemas.py
 |  |   +-- file models.py
 |  |   \-- file views.py
 |  |
 |  +--dir blog 
 |  |   +-- file __init__.py
 |  |   +-- file schemas.py
 |  |   +-- file models.py
 |  |   \-- file views.py
 |  |
 |  +--file config.py
 |  |

We created a folder named tutorial series inside which we created two files manager.py and requirements.txt and one python package src which will contain our codebase. Inside src there are two three python packages utils, users and blog.

Note: A package is a collection of Python modules: while a module is a single Python file, a package is a directory of Python modules containing an additional init.py file, to distinguish a package from a directory that just happens to contain a bunch of Python scripts.

You can find the structure here: Structure

Lets take a look at all the files one by one.

1. Utils Folder

""" utils/models.py"""

import re
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.declarative import declared_attr

db = SQLAlchemy()


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__)

    __mapper_args__ = {'always_refresh': True}

    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):
    """Provides a string representible form for objects."""

    __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)



""" utils/schemas.py"""

from flask_marshmallow import Marshmallow
from marshmallow_sqlalchemy import ModelSchema, ModelSchemaOpts
from .models import db


class FlaskMarshmallowFactory(Marshmallow):

    def __init__(self,  *args, **kwargs):
        super(FlaskMarshmallowFactory, self).__init__(*args, **kwargs)


class BaseOpts(ModelSchemaOpts):
    def __init__(self, meta):
        if not hasattr(meta, 'sql_session'):
            meta.sqla_session = db.session
        super(BaseOpts, self).__init__(meta)


class BaseSchema(ModelSchema):
    OPTIONS_CLASS = BaseOpts

ma = FlaskMarshmallowFactory()



""" utils/factory.py """
from flask import Flask


def create_app(package_name, config, extensions=None):
    app = Flask(package_name)
    app.config.from_object(config)
    config.init_app(app)
    
    if extensions:
        for extension in extensions:

            extension.init_app(app)

    return app

""" utils/api.py"""
from flask_restful import Api

api = Api()

Our utils folder contains four files models.py , factory.py, api.py and schemas.py. models.py contains base classes for our models and initialization of SQLAlchemy. schemas.py contains base class for our schemas and initialization of Marshmallow. api.py contains initialization of flask_restful, We will learn about creating base classes for our API’s in next to next chapter.

run the following command to install the flask_restful package. pip install flask_restful

factory.py contains a function which initializes our app and bind all the extensions like db, ma with it

User Folder

"""user/models.py"""

from sqlalchemy import UniqueConstraint
from sqlalchemy.ext.hybrid import hybrid_property
from src import db, BaseMixin, ReprMixin


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.user_profile.first_name) + (' {}'.format(self.user_profile.last_name) if
                                               self.user_profile.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')


""" user/schemas.py"""

from src import ma, BaseSchema
from .models import User, UserProfile


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

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


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

    id = ma.Integer(load=True, partial=True)
    first_name = ma.String()
    last_name = ma.String()
    user_id = ma.Integer(load_only=True, partial=True)


"""user/views.py"""

from flask_restful import Resource
from flask import make_response, jsonify, request
from src import api, db
from .models import User
from .schemas import UserSchema


class UserResource(Resource):

    model = User
    schema = UserSchema

    def get(self, slug):
        user = self.model.query.get(slug)
        if not user:
            return make_response(jsonify({'error': 'Resource not found'}), 404)
        return make_response(jsonify(self.schema().dump(user).data), 200)

    def patch(self, slug):
        user = self.model.query.get(slug)
        if not user:
            return make_response(jsonify({'error': 'Resource not found'}), 404)
        user, errors = self.schema().load(request.json, instance=user)
        if errors:
            return make_response(jsonify(errors), 400)
        db.session.commit()
        return make_response(jsonify(self.schema().dump(user).data), 200)

    def delete(self, slug):
        user = self.model.query.get(slug)
        if not user:
            return make_response(jsonify({'error': 'Resource not found'}), 404)
        db.session.delete(user)
        db.session.commit()
        return make_response(jsonify({}), 204)


class UserListResource(Resource):

    model = User
    schema = UserSchema

    def get(self):
        users = self.model.query.limit(20).all()
        if not users:
            return make_response(jsonify({'error': 'Resource not found'}), 404)
        return make_response(jsonify(self.schema().dump(users, many=True).data), 200)

    def post(self):
        users, errors = self.schema().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(self.schema().dump(users, many=True).data), 201)

api.add_resource(UserResource, '/user/<slug>', endpoint='user')
api.add_resource(UserListResource, '/user', endpoint='users')

user/models.py contains our User Models. user/schemas.py contains our User Schema’s. user/api.py We have created class based views for our api’s the code is pretty straight forward.

  1. Our UserResource contains three methods get, put and delete and requires a slug to passed in the url which is the id of the User Model.
  2. Our UserListRessource contains two methods get and post, get methods return multiple users from the database and post method is used to create one or many new Users.

Similarly our posts package contains 4 files views, models, schemas and init containing the respective code for Post.

src/__ init __.py

"""src/__init__.py"""

from .config import configs
from .utils import db, BaseMixin, ReprMixin, ma, BaseSchema, create_app, api

from .blog import models
from .users import models
from .blog import schemas, views
from .users import schemas, views

In our src/__ init __.py file we import everything from our internal packages.

"""src/config.py"""
import os
from datetime import timedelta

basedir = os.path.abspath(os.path.dirname(__file__))


class BaseConfig:
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    SQLALCHEMY_RECORD_QUERIES = False

    MARSHMALLOW_STRICT = True
    MARSHMALLOW_DATEFORMAT = 'rfc'

    SECRET_KEY = 'SECRETKEY@123'

    @staticmethod
    def init_app(app):
        pass


class DevConfig(BaseConfig):
    DEBUG = True

    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URI') or \
                              'sqlite:///{}'.format(os.path.join(basedir, 'dev.db'))


class TestConfig(BaseConfig):
    TESTING = True

    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URI') or \
                              'sqlite:///{}'.format(os.path.join(basedir, 'test.db'))


class ProdConfig(BaseConfig):
    SQLALCHEMY_DATABASE_URI = os.environ.get('PROD_DATABASE_URI') or \
                              'sqlite:///{}'.format(os.path.join(basedir, 'why-is-prod-here.db'))


configs = {
    'dev': DevConfig,
    'testing': TestConfig,
    'prod': ProdConfig,
    'default': DevConfig
}

In our src/config.py we create different configuration for different environments which can be selected by setting an environment variable like

export PYTH_SRVR=dev

All the configuration required are stored in this file. All the required options for all the extensions used are stores in here.

src/manage.py

""" src/manage.py"""

import os

import urllib.parse as up
from flask_script import Manager
from flask import url_for

from src import db, ma, create_app, configs, api

config = os.environ.get('PYTH_SRVR')

config = configs.get(config, 'default')

extensions = [db, ma, api]

app = create_app(__name__, config, extensions=extensions)

manager = Manager(app)


@manager.shell
def _shell_context():
    return dict(
        app=app,
        db=db,
        ma=ma,
        config=config
        )


@manager.command
def list_routes():
    output = []
    for rule in app.url_map.iter_rules():
        options = {}
        for arg in rule.arguments:
            options[arg] = "[{0}]".format(arg)
        methods = ','.join(rule.methods)
        url = url_for(rule.endpoint, **options)
        line = up.unquote("{:50s} {:20s} {}".format(rule.endpoint, methods, url))
        output.append(line)

    for line in sorted(output):
        print(line)

if __name__ == "__main__":
    manager.run()

A lot of things are happening in this file lets go by them one by one.

Note: pip install flask_script We require a plugin Flask Script for this file.

  1. from src import db, ma, create_app, configs, api importing all the extensions and our function create_app to create our app.
  2. config = os.environ.get('PYTH_SRVR') getting the value of environment variable ‘PYTH_SRVR’ and storing it in config.
  3. config = configs.get(config, 'default') setting the config class from our config.py according to the value for our environment variable or setting it to ‘default’.
  4. app = create_app(__name__, config, extensions=extensions) calling our function with configuration and extension which will create our app.

This is going to be the basic structure that we will follow throught the next chapters . The code can be found over here Github Repo

Written with StackEdit.

Tags:
Stats:
0 comments