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
""" 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/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.
Similarly our posts package contains 4 files views, models, schemas and init containing the respective code for Post.
"""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"""
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.
from src import db, ma, create_app, configs, api
importing all the extensions and our function create_app to create our app.config = os.environ.get('PYTH_SRVR')
getting the value of environment variable ‘PYTH_SRVR’ and storing it in config.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’.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.