Flask 基礎知識與實作 part 3

Author: 毛毛, Alicia

Outline

  • Ch6: Databases
  • Ch7: Large Application Structure

Ch6: Databases

Note: Python 2 & Python 3 will be slightly different

Introduction to SQL and NoSQL Database

SQL Databases

  • Relational databases.
    • Store data in tables.
      • Fixed number of columns and various rows.
      • The columns define data attritubes.
      • Each rows is an actual data consist of values for all columns.
    • Ex: a customer table will have name, address, phone columns.

Primary Key & Foreign Key & Relationships

  • Primary key: a special column that holds a unique indentifier for each row.
  • Foreign key: special columns that reference the primary key of another row from the same or another table.
  • Relationships: connections between each table defined by foreign keys and primary keys.
  • In relational databased, tables need to be joined before they can be presented.

NoSQL Databases

  • Use collections instaed of tables and documents instead of records.
  • Document oriented: document oriented are similar to key-value concept.

Key-value store

  • Data is represented as a collection of key-value pairs, like Python dictionary.
{"FirstName": "Alicia", "Address": "Taipei, Taiwan", "Gender": "Female"}

Document store

  • A "document" that encapsulate and encode data in some standard formats or encodings.
  • Encodings include XML, YAML, and JSON, etc.
  • Documents are access via a unique key that represents that document.
  • Compared to key-value store, document store offers ways to group or organize documents.
    • Collections
    • Tags
    • Non-visible metadata
    • Directory hierarchies
<!-- A document encoded in XML -->
  <contact>
    <firstname>Bob</firstname>
    <lastname>Smith</lastname>
    <phone type="Cell">(123) 555-0178</phone>
    <phone type="Work">(890) 555-0133</phone>
    <address>
      <type>Home</type>
      <street1>123 Back St.</street1>
      <city>Boys</city>
      <state>AR</state>
      <zip>32225</zip>
      <country>US</country>
    </address>
  </contact>

Python Database Frameworks

Python has packages for most database engines.

  • Psycopg: the most popular PostgreSQL adapter for the Python programming language.
  • MySQLdb: an interface to the popular MySQL database server for Python
  • sqlite3: a DB-API 2.0 interface for SQLite databases

Note: you have to use SQL syntax

There are also database abtraction layer packages to use.

  • SQLAlchemy: the Python SQL toolkit and Object Relational Mapper
    • Allow us to work at a higher level with regular Python objects instead of database entities such as tables.
    • ORM can be called ODM: provide conversion of low-level database instructions to high-level obejct-oriented operations.

Evaluate a database framework:

  • Ease of use
  • Performance
    • Translate object into the SQL language have an overhead cost. The cost is negligible.
    • Boost the productivity for people who are unfamiliar with SQL syntax.
    • Choose a framework that provides native SQL syntax instructions in case specific operations are needed.
  • Portability
    • Choose a database that can be used on both the development and production platforms.
    • Provide choices for different database engines with the same object-oriented interface.
    • Ex: SQLALchemy ORM supports MySQL, Postgres, SQLite
  • Flask integration
    • Choose a framework that has integration with Flask to simplify configuration and operation.

Flask extension

flask-sqlalchemy

  • A SQLAlchemy wrapper.
  • Simplifies the use of SQLAlchemy inside Flask applications.

SQLAlchemy

  • A relational database framework.
  • Support several database backends.
  • Offers a high-level ORM and low-level access to the database's native SQL functionality.

Installation

  • pip install flask-sqlalchemy

Database Configuration

SQLALCHEMY_DATABASE_URL

  • In Flask-SQLAlchemy, a database is specify as a URL.
Database Engine URL
MySQL mysql://username:password@hostname/database
Postgres postgresql://username:password@hostname/database
SQLite(Unix) sqlite:////absolute/path/to/database
SQLite (Windows) sqlite:///c:/absolute/path/to/database
  • hostname: the server that host the database, which can be localhost or a remote server.
  • database: the name of the database to use.
  • username, password: user credentails for authentication.
  • SQLite databases do not have a server, so the database is the filename of a disk file.
In [ ]:
# hello_db.py

import os

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy
#from flask.ext.sqlalchemy import SQLAlchemy

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

app = Flask(__name__)
manager = Manager(app)
app.config["SQLALCHEMY_DATABASE_URI"] =\
    "sqlite:///" + os.path.join(basedir, "data.sqlite")
app.config["SQLAlCHEMY_COMMIT_ON_TEARDOWN"] = True
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_ECHO"] = True

db = SQLAlchemy(app)

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

SQLALCHEMY_COMMIT_ON_TEARDOWN

  • Set as True to enable automatic commits of database changes.

SQLALCHEMY_TRACK_MODIFICATIONS

  • If se to True, track modifications of objects and meit signals.
  • This requires extra memory and should be disabled if not needed.

SQLALCHEMY_ECHO

  • If set to True SQLAlchemy will log all the SQL statements.
  • Can be useful for debugging and testing.

More information about configuration: http://www.pythondoc.com/flask-sqlalchemy/config.html

Model Definition

  • Refer to the entities of the database, such as tables.
  • The db instance provides a base class for models and helper classes and functions that are used to define their structure.
In [ ]:
# hello_db.py

import os

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy
#from flask.ext.sqlalchemy import SQLAlchemy

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

app = Flask(__name__)
manager = Manager(app)
app.config["SQLALCHEMY_DATABASE_URI"] =\
    "sqlite:///" + os.path.join(basedir, "data.sqlite")
app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_ECHO"] = True

db = SQLAlchemy(app)
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

    def __repr__(self):
        return '<Role %r>' % self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username
    

if __name__ == '__main__':
    manager.run()
  • The __tablename__ class variable defines the name of the table in the database.
    • If not specified, Flask- SQLAlchemy assigns a default table name.
    • Default names do not follow the convention of using plurals for table names.
  • Class variables such as id and username are attributes of the model (columns of the table).
    • They are defined as instances of the db.Column class.
    • The first argument of the db.Column is the type of the column.
    • Flask-SQLAlchemy requires all models to define a primary key column, which is normally named id.

Most common SQLAlchemy column types

Type name Python type Description
Integer int Regular integer (32 bits)
SmallInteger int Short-range integer (16 bits)
BigInteger int or long Unlimited precision integer
Float float Floating-point number
Numeric decimal.Decimal Fixed-point number
String str Variable-length string
Text str variable-length string (optimized for large or unbound length)
Unicode unicode Variable-length Unicode string
UnicodeText unicode Variable-length Unicode string (optimized for large or unbound length)
Boolean bool Boolean value
Date datetime.date Date value
Time datetime.time Time value
DateTime datetime.datetime Date and time value
Interval datetime.timedelta Time interval
Enum str List of string values
PickleType Any Python object Automatic Pickle serialization
LargeBinary str Binary blob

Most Common SQLAlchemy column options

Option name Description
primary_key If set to True, the column is the table’s primary key.
unique If set to True, do not allow duplicate values for this column.
index If set to True, create an index for this column, so that queries are more efficient.
nullable If set to True, allow empty values for this column.
default Define a default value for the column.

Database Operations

CRUD

  • Create
  • Read
  • Update
  • Delete

Creating The Tables

  • We will use Python shell to walk you through the process of working with models.
  • Use db.create_all() function to create a database based on the model classes.
  • The db.create_all() function will not re-create or update a database table if it already exists in the database.
    • We can update the database by removing the old one first and then re-create the database again.
    • A better solution will be presented near the end of the chapter.
(venv) $ python hello_db.py shell 
>>> from hello_db import db
>>> db.create_all()

>>> db.drop_all()
>>> db.create_all()

Before creating database

After creating database "data.sqlite exits now"

Inspecting Existing Tables

>>> db.Model.metadata.tables.keys()
['users', 'roles']

Inserting Rows

  • Creates roles.
  • We do not need to set the id attribute explicitly becasue Flask-SQLAlchemy will manage the primary keys for us.
  • The objects have not been written to the database yet so the id value has not been assigned.
>>> from hello_db import Role

>>> admin_role = Role(name='Admin')
>>> mod_role = Role(name='Moderator')
>>> user_role = Role(name='User')

>>> admin_role.name
'Admin'
>>> admin_role.id
>>>

Writing to Database

  • Use db.session to add objects prepared to be written to the database.
>>> db.session.add(admin_role)
>>> db.session.add(mod_role)
>>> db.session.add(user_role)

# a more concise way
>>> db.session.add_all([admin_role, mod_role, user_role])

Benefits of Using db.session()

  • Write all the objects added to the session atomically.
    • Atomically means the transcation cannot be cut into pieces.
  • If an error occurs, the whole session is discarded.
  • Avoid database inconsistencies due to partial updates.

Discard and reset a session

  • Use rollback() to discard the previous action and reset the session.
>>> db.session.rollback()
>>> db.session.add_all([admin_role, mod_role, user_role])

Write data to the database

  • Use commit() method to commit the session and write objects to the database.
  • If an error occurs while the session is being written, the whole session is discarded.
>>> db.session.commit()

  • The id attributes are now set to the Model.
>>> admin_role.id
1
>>> mod_role.id
2

Querying Rows

  • query is available in each model class.

Common SQLAlchemy query filters

Option Description
filter() Returns a new query that adds an additional filter to the original query
filter_by() Returns a new query that adds an additional equality filter to the original query
limit() Returns a new query that limits the number of results of the original query to the given number
offset() Returns a new query that applies an offset into the list of results of the original query
order_by() Returns a new query that sorts the results of the original query according to the given criteria
group_by() Returns a new query that groups the results of the original query according to the given criteria

Common SQLAlchemy query executors

Option Description
all() Returns all the results of a query as a list
first() Returns the first result of a query, or None if there are no results
first_or_404() Returns the first result of a query, or aborts the request and sends a 404 error as response if there are no results
get() Returns the row that matches the given primary key, or None if no matching row is found
get_or_404() Returns the row that matches the given primary key. If the key is not found it aborts the request and sends a 404 error as response
count() Returns the result count of the query
paginate() Returns a Pagination object that contains the specified range of results

all()

  • Return the results as a list.
>>> roles = Role.query.all()
>>> roles
[<Role 'Admin'>, <Role 'Moderator'>, <Role 'User'>]
>>> roles[0].name
'Admin'
>>> roles[0].id
1

Filters: filter() and filter_by()

  • Use query filters to search for more specific information.
  • filter_by: used for simple queries on the column names using regular argument.
    • Cannot perform operations such as > or <.
  • filer: use == to accomplish the same thing with object overloaded ahead.
  • Multiple query filters can be called in sequence.
  • You can also applied query filters first and then use the query executors.
>>> Role.query.filter_by(name='Admin')
<flask_sqlalchemy.BaseQuery object at 0x103c11510>

>>> str(Role.query.filter_by(name='Admin'))
'SELECT roles.id AS roles_id, roles.name AS roles_name \nFROM roles \nWHERE roles.name = ?'

# filter -> executor
>>> Role.query.filter_by(name='Admin').all()
[<Role 'Admin'>]

# error
>>> Role.query.filter_by(id > 2).all()
Traceback (most recent call last):
  File "<console>", line 1, in <module>
TypeError: unorderable types: builtin_function_or_method() > int()

# db.Role.id overloaded ahead
# filter -> executor
>>> Role.query.filter(Role.id > 2).all()
[<Role 'User'>]

frist()

  • Reutrn the first result of the query.
  • Return None if no search result.
>>> Role.query.filter_by(name='User').first()
<Role 'User'>
>>> Role.query.filter_by(name='Stranger').first()
>>>

Modifying Rows

  • Use add() method to update any changes to the models.
  • Rename 'Admin' role to 'Administrator'.
>>> admin_role.name = 'Administrator'
>>> db.session.add(admin_role)
>>> db.session.commit()

>>> admin_role.name
'Administrator'

Deleting Rows

  • Use delete() method to delete object from the models.
  • Delete 'Moderator' role from Role model.
>>> db.session.delete(mod_role)
>>> db.session.commit()
>>> Role.query.all()
[<Role 'Admininstrator'>, <Role 'User'>]

Insertion, update and deletion are only executed when the session is committed.

You can get the above code from github

Coding Time

  1. Copy and paste code from http://tw.pyladies.com/~maomao/3_flask.slides.html#/1/25 and http://tw.pyladies.com/~maomao/3_flask.slides.html#/1/26.
  2. Open python shell
    (venv) $ python hello_db.py shell
    >>> from hello_db import db, User
    
  3. Add users:
    • User: maomao, yourname
  4. Make sure data exists using query syntax
  5. Update username of maomao to abby
  6. Delete abby
  7. Make sure only yourname exists in User table using query syntax
Command
db.session.delete()
db.session.add()
db.session.add_all()
db.session.commit()
User.query.all()

Relationship

Types of relationships

  • One-to-one: Both tables can have only one record on either side of the relationship. Each primary key value relates to only one (or no) record in the related table.

  • One-to-many: One of the column has duplicate records.

  • Many-to-many: Each record in both tables can relate to any number of records (or no records) in the other table. It requires a third table (association table or linking table) to accommodate the relationship. We'll talk about many-to-many relationship in Chapter 12.

Relationship between users and roles

  • One-to-many relationship

Foreign key

  • The foreign key has to be declared with the ForeignKey class.
  • The role_id column is defined as a foreign key and establishs a relationship.
  • Pass the value from the table that should be referenced as the foreign key to the ForeignKey class.
  • Here, the foreign key is the 'id' values from the 'roles' table.

Relationship

  • Relationships are expressed with the relationship() function.
  • Given an instance of class Role, the users attribute will return the list of users associated with that role.
  • The first argument indicates what model is on the other side of the relationship.
    • If the class is not yet created, you can use strings to refer to the class and declared it later in the file.
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User') # relationship (from Role link to User)

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username
    

db.relationship() can locate the relationship’s foreign key on its own

Run the shell

(venv) $ python hello_db.py shell

>>> from hello_db import db, User, Role
>>> db.drop_all()
>>> db.create_all()
>>> admin_role = Role(name='Admin')
>>> db.session.add(admin_role)
>>> db.session.commit()   # need to commit first so that the id can be created

>>> maomao = User(username="Maomao", role_id=admin_role.id)
>>> db.session.add(maomao)   
>>> db.session.commit()
>>> admin_role.users
[<User 'Maomao'>]
>>> Alicia = User(username="Alicia", role_id=admin_role.id)
>>> db.session.add(Alicia)
>>> db.session.commit()
>>> admin_role.users
[<User 'Maomao'>, <User 'Alicia'>]
>>>
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User')

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    role = db.relationship('Role') # relationship (from User link to Role)

    def __repr__(self):
        return '<User %r>' % self.username
    

Run the shell

(venv) $ python hello_db.py shell

>>> from hello_db import db, User, Role
>>> db.drop_all()
>>> db.create_all()
>>> admin_role = Role(name='Admin')
>>> db.session.add(admin_role)
>>> db.session.commit()

>>> maomao = User(username="Maomao", role_id=admin_role.id)
>>> Alicia = User(username="Alicia", role_id=admin_role.id)
>>> db.session.add_all([maomao, Alicia])
>>> db.session.commit()
>>> admin_role.users
[<User 'Maomao'>, <User 'Alicia'>]
>>> maomao.role
<Role 'Admin'>
>>> Alicia.role
<Role 'Admin'>

backref

  • The backref argument defines the reverse direction of the relationship.
    • It adds a new role attribute to the User model.
      • role is not a real db column.
      • It is a one-to-many relationship, same as users.
  • You don't need to write relationship in both class.
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role')

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username
    

Run the shell

(venv) $ python hello_db.py shell

>>> from hello_db import db, User, Role
>>> db.drop_all()
>>> db.create_all()
>>> admin_role = Role(name='Admin')
>>> db.session.add(admin_role)
>>> db.session.commit()

>>> maomao = User(username="Maomao", role_id=admin_role.id)
>>> Alicia = User(username="Alicia", role_id=admin_role.id)
>>> db.session.add_all([maomao, Alicia])
>>> db.session.commit()
>>> admin_role.users
[<User 'Maomao'>, <User 'Alicia'>]
>>> maomao.role
<Role 'Admin'>
>>> Alicia.role
<Role 'Admin'>

Ambiguous: multiple columns refer to the same table as foreign key

In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)    

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    primary_role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    secondary_role_id =  db.Column(db.Integer, db.ForeignKey('roles.id'))

    primary_role = db.relationship('Role')  # ambiguous
    secondary_role = db.relationship('Role') # ambiguous
    
    def __repr__(self):
        return '<User %r>' % self.username

Fix ambiguity

  • Specify the foreign_keys argumet
    • User.primary_role relationship will use the value present in primary_role_id to identify the row in Role to be loaded
    • User.secondary_role relationship will use the value present in secondary_role_id to identify the row in Role to be loaded
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)    

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    primary_role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    secondary_role_id =  db.Column(db.Integer, db.ForeignKey('roles.id'))

    primary_role = db.relationship('Role', foreign_keys='User.primary_role_id')
    secondary_role = db.relationship('Role', foreign_keys='User.secondary_role_id')     

    def __repr__(self):
        return '<User %r>' % self.username

Run the shell

(venv)$ python hello_db.py shell 

>>> from hello_db import db, User, Role

>>> db.drop_all()
>>> db.create_all()

>>> admin_role = Role(name='Admin')
>>> mod_role = Role(name='Mod')
>>> db.session.add_all([admin_role, mod_role])
>>> db.session.commit()

>>> maomao = User(username="Maomao", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> Alicia = User(username="Alicia", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> db.session.add_all([maomao, Alicia])
>>> db.session.commit()

>>> maomao.primary_role
<Role 'Admin'>
>>> maomao.secondary_role
<Role 'Mod'>
>>>
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True) 
    
    users_primary_role = db.relationship('User', primaryjoin='Role.id==User.primary_role_id')
    users_secondary_role = db.relationship('User', primaryjoin='Role.id==User.secondary_role_id')     

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    primary_role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    secondary_role_id =  db.Column(db.Integer, db.ForeignKey('roles.id'))

    primary_role = db.relationship('Role', foreign_keys='User.primary_role_id')
    secondary_role = db.relationship('Role', foreign_keys='User.secondary_role_id')     

    def __repr__(self):
        return '<User %r>' % self.username

Run the shell

(venv) $ python hello_db.py shell 

>>> from hello_db import db, User, Role

>>> db.drop_all()
>>> db.create_all()

>>> admin_role = Role(name='Admin')
>>> mod_role = Role(name='Mod')
>>> db.session.add_all([admin_role, mod_role])
>>> db.session.commit()

>>> maomao = User(username="Maomao", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> Alicia = User(username="Alicia", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> db.session.add_all([maomao, Alicia])
>>> db.session.commit()

>>> maomao.primary_role
<Role 'Admin'>
>>> Alicia.secondary_role
<Role 'Mod'>
>>> admin_role.users_primary_role
[<User 'Maomao'>, <User 'Alicia'>]
>>> admin_role.users_secondary_role
[]
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True) 
    
    users_primary_role = db.relationship('User', backref='primary_role', primaryjoin='Role.id==User.primary_role_id')
    users_secondary_role = db.relationship('User', backref='secondary_role', primaryjoin='Role.id==User.secondary_role_id')     

    def __repr__(self):
        return '<Role %r>' % self.name
    

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    primary_role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    secondary_role_id =  db.Column(db.Integer, db.ForeignKey('roles.id'))    

    def __repr__(self):
        return '<User %r>' % self.username

Run the shell

(venv) $ python hello_db.py shell 

>>> from hello_db import db, User, Role

>>> db.drop_all()
>>> db.create_all()

>>> admin_role = Role(name='Admin')
>>> mod_role = Role(name='Mod')
>>> db.session.add_all([admin_role, mod_role])
>>> db.session.commit()

>>> maomao = User(username="Maomao", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> Alicia = User(username="Alicia", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> db.session.add_all([maomao, Alicia])
>>> db.session.commit()

>>> maomao.primary_role
<Role 'Admin'>
>>> Alicia.secondary_role
<Role 'Mod'>
>>> admin_role.users_primary_role
[<User 'Maomao'>, <User 'Alicia'>]
>>> admin_role.users_secondary_role
[]

You can get the above code from github

2 ways to create user

Case 1: use foreign key

# add relationship using foreign key
>>> maomao = User(username="Maomao", primary_role_id=admin_role.id, secondary_role_id=mod_role.id)
>>> db.session.add(maomao)     
>>> db.session.commit()

Case 2: use back reference

# add relationship using back reference
>>> abby = User(username="Abby", primary_role=admin_role, secondary_role=mod_role)
>>> db.session.add(abby)
>>> db.session.commit()

Common SQLAlchemy relationship options

Option name Description
backref Add a back reference in the other model in the relationship.
primaryjoin Specify the join condition between the two models explicitly. This is necessary only for ambiguous relationships.
lazy Specify how the related items are to be loaded. Possible values are select (items are loaded on demand the first time they are accessed), immediate (items are loaded when the source object is loaded), joined (items are loaded immediately, but as a join), subquery (items are loaded immediately, but as a subquery), noload (items are never loaded), and dynamic (instead of loading the items the query that can load them is given).
uselist If set to False, use a scalar instead of a list.
order_by Specify the ordering used for the items in the relationship.
secondary Specify the name of the association table to use in many-to-many relationships.
secondaryjoin Specify the secondary join condition for many-to-many relationships when SQLAlchemy cannot determine it on its own.

Some Lazy options

select (default): users = db.relationship('User', backref='role')

Does not query users data directly

>>> first_role = Role.query.first()

select (default): users = db.relationship('User', backref='role')

Query data when accessing users

>>> first_role.users

select (default): users = db.relationship('User', backref='role')

Does not query again after querying

>>> first_role.users

select (default): users = db.relationship('User', backref='role')

Does not query role data directly

>>> first_user = User.query.first()

select (default): users = db.relationship('User', backref='role')

Query data when accessing role

>>> first_user.role

select (default): users = db.relationship('User', backref='role')

Does not query again after querying

>>> first_user.role

joined: users = db.relationship('User', backref=db.backref('role', lazy='joined'), lazy='joined')

Select data one time, query users data directly

>>> first_role = Role.query.first()

joined: users = db.relationship('User', backref=db.backref('role', lazy='joined'), lazy='joined')

Does not query again when accessing users

>>> first_role.users

joined: users = db.relationship('User', backref=db.backref('role', lazy='joined'), lazy='joined')

Select data one time, query role data directly

>>> first_user = User.query.first()

joined: users = db.relationship('User', backref=db.backref('role', lazy='joined'), lazy='joined')

Does not query again when accessing role

>>> first_user.role

dynamic: users = db.relationship('User', backref='role', lazy='dynamic')

  • When setting lazy to dynamic, the query will not be automatically executed.

Have not query yet

>>> first_role = Role.query.first()

dynamic: users = db.relationship('User', backref='role', lazy='dynamic')

  • When setting lazy to dynamic, the query will not be automatically executed.

Query object

>>> first_role.users

dynamic: users = db.relationship('User', backref='role', lazy='dynamic')

  • When setting lazy to dynamic, the query will not be automatically executed.

Use filter and executors

>>> first_role.users.order_by(User.username).all()

>>> first_role.users.count()

Use Database in View functions

  • We can use database operations directly in the view functions.
  • We can record data in the database from the web form.

Greet users

  • When user exists in database, greet them with "Happy to see you again".
  • When new user joins, greet them with "Pleased to meet you" and record user name in database.
In [ ]:
# hello_db.py

import os
from flask import Flask, render_template, session, redirect, url_for
from flask_script import Manager
from flask_bootstrap import Bootstrap
from flask_moment import Moment
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import Required
from flask_sqlalchemy import SQLAlchemy

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

app = Flask(__name__)
app.config['SECRET_KEY'] = 'hard to guess string'
app.config['SQLALCHEMY_DATABASE_URI'] =\
    'sqlite:///' + os.path.join(basedir, 'data.sqlite')
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config["SQLALCHEMY_ECHO"] = True

manager = Manager(app)
bootstrap = Bootstrap(app)
moment = Moment(app)
db = SQLAlchemy(app)
In [ ]:
# hello_db.py

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role', lazy='dynamic')

    def __repr__(self):
        return '<Role %r>' % self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username


class NameForm(FlaskForm):
    name = StringField('What is your name?', validators=[Required()])
    submit = SubmitField('Submit')
In [ ]:
# hello_db.py

@app.errorhandler(404)
def page_not_found(e):
    return render_template('404.html'), 404


@app.errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500


@app.route('/', methods=['GET', 'POST'])
def index():
    form = NameForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.name.data).first()     # query user
        if user is None:
            admin_role = Role.query.filter_by(name='Admin').first()      # query role
            user = User(username=form.name.data, role_id=admin_role.id)  # add user
            db.session.add(user)     # add to database
            session['known'] = False
        else:
            session['known'] = True
        session['name'] = form.name.data
        return redirect(url_for('index'))
    return render_template('index.html', form=form, name=session.get('name'),
                           known=session.get('known', False))
In [ ]:
# hello_db.py

@app.route('/db/users')
def list_users_in_db():
    users = User.query.all()
    return "<br>".join([user.username for user in users])

if __name__ == '__main__':
    manager.run()
In [ ]:
<!-- templates/index.html -->

{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}

{% block title %}Flasky{% endblock %}

{% block page_content %}
<div class="page-header">
    <h1>Hello, {% if name %}{{ name }}{% else %}Stranger{% endif %}!</h1>
    {% if not known %}
    <p>Pleased to meet you!</p>
    {% else %}
    <p>Happy to see you again!</p>
    {% endif %}
</div>
{{ wtf.quick_form(form) }}
{% endblock %}

Run the shell

(venv) $ python hello_db.py shell
>>> from hello_db import db, User, Role
>>> db.drop_all()
>>> db.create_all()

>>> admin_role = Role(name='Admin')
>>> mod_role = Role(name='Mod')
>>> db.session.add_all([admin_role, mod_role])
>>> db.session.commit()

>>> maomao = User(username='Maomao', role_id=admin_role.id)
>>> db.session.add(maomao)
>>> db.session.commit()
(venv) $ python hello_db.py runserver
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

Integration with the Python Shell

  • Use Flask-script's shell command to import objects automatically.
  • The make_shell_context() function registers the app, db, Role and User model.
  • They will be automatically imported into the shell.
  • If make_context is not set, it will return app as default.
In [ ]:
# hello_db.py

#...
from flask_script import Shell

def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)

manager.add_command("shell", Shell(make_context=make_shell_context))

You can get the above code from github

Database Migration

Database migration framwork

  • Keeps track of changes to a database model.
  • Incremental changes can be applied to the database.
    • No need to destory the old database and re-create.
    • No data will be lost.

alembic

  • A database migrations tool for SQLAlchemy written by the author of SQLAlchemy.

Flask extension

flask-migrate

  • A lightweight Alembic wrapper that integrates with Flask-Script
    • Provide all operations through Flask-Script commands.

Installation

$ pip install flask-migrate

flask-migrate configuration

  • Attach MigrateCommand class to Flask-Script's manager object.
  • Here the command is attached using db.
In [ ]:
# hello_db.py

from flask_migrate import Migrate, MigrateCommand
# from flask.ext.migrate import Migrate, MigrateCommand 

# ...

migrate = Migrate(app, db)
manager.add_command('db', MigrateCommand)

Reset environment

  • In order to test the functionality of migration, we need to delete data.sqlite manually.

Create migration repository

  • Create a migration repository before migrations.
  • The repository stores all migration scripts under each version.
$ python hello_db.py db init

Create a migration script

  • A database migration is represented by a migration script.
  • Automatically create migration using migrate command.
    • upgrade() and downgrade() look for differences between the model definitions and the current database.
    • upgrade(): updates changes to the database.
    • downgrade(): removes changes.
  • Manually create migration using revision command.
    • Create empty upgrade() and downgrade() functions that need to be implemented by the developer.

Create automatic migration script

  • The db migrate command creates an automatic migration script.
$ python hello_db.py db migrate -m "Initial migration"

Upgrading the database

  • Once a migration script has been reviewed and accepted, we can update the database using db upgrade command.
  • For the first migration, it is equivalent to db.create_all().
$ python hello_db.py db upgrade

Modify model

  • Add a new column 'email' into User table.
In [ ]:
# hello_db.py

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))
    email = db.Column(db.String(64), unique=True, index=True)

    def __repr__(self):
        return '<User %r>' % self.username

Create migration script again

$ python hello_db.py db migrate -m “Add new column in ‘users’”

Upgrade database again

  • New column 'email' is added into Users table.
    $ python hello_db.py db upgrade
    

You can get the above code from github

Ch7: Large Application Structure

Flask does not impose any specific organizatin for large project

  • The structure of the application is left entirely to the developer.

Basic multiple-file Flask application structure

|-flasky
  |-app/
    |-templates/
    |-static/
    |-main/
      |-__init__.py
      |-errors.py
      |-forms.py
      |-views.py
    |-__init__.py
    |-email.py
    |-models.py
  |-migrations/
  |-tests/
    |-__init__.py
    |-test*.py
  |-venv/
  |-requirements.txt
  |-config.py
  |-manage.py

Four top-level folders

  • app: our Flask application
  • migrations: database migration scripts
  • tests: unit tests
  • venv: Python virtual environment

Special files

  • requirements.txt:
    • Lists the package dependencies so it is easy to regenerate an identical virtual environment on a different computer.
    • Use pip freeze > requirments.txt to generate the file.
    • Use pip install -r requirments.txt to regenerate the packages.
  • config.py: stores the configuration settings.
  • manage.py: launches the application and other application tasks.

requirements.txt

Convert hello.py to our application structure.

config.py

  • Usually, we use different databases during development, testing, and production, so we often need different configuration sets.
  • We can use a hierarchy of configuration classess instead of the simple dictionary-like structure.
  • Some sensitive setting can be imported from environment variables. ex: SECRET_KEY
In [ ]:
# config.py

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

# base class
class Config(object):
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    MAIL_SERVER = 'smtp.googlemail.com'
    MAIL_PORT = 587
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <flasky@example.com>'
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')

    @staticmethod
    def init_app(app):
        pass
In [ ]:
# SQLALCHEMY_DATABASE_URI is assigned different values

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')


class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-test.sqlite')


class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')

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

    'default': DevelopmentConfig
}

Application Factory

  • The single-file version script creates the application in the global scope.
  • When the script is running, the application instance has already been created.
  • No time to make configuration change.
  • Move the initialization into a factory function.
    • The factory function can be explicitly invoked.
    • Gives the script time to set the configuration first.
    • Can create multiple application instances as well.
  • Define the application factory function in the app package constructor (i.e. __init__)

app/

|-app/
    |-templates/
    |-static/
    |-main/
      |-__init__.py
      |-errors.py
      |-forms.py
      |-views.py
    |-__init__.py
    |-email.py
    |-models.py
In [ ]:
# app/__init__.py  (Application package constructor)

from flask import Flask
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()

# application factory
def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    
    # attach routes and custom error pages blueprint here

    return app

Blueprint

  • Single script without Blueprint: app instance need to exist before defining the route.
    • eg: app exists in the global scope, and we use app.route decorator to register the route.
  • Blueprint: an application template.
    • Routes associated with a blueprint are in a dormant state.
    • Routes become part of the applicaiton after the blueprint is registered with an application instance.

More information: https://spacewander.github.io/explore-flask-zh/7-blueprints.html

Organize blueprint structure

  • Blueprints can be defined all in a single file or in a more strucutred way.
  • Here, we put different modules inside a package that holds all the blueprint files.
|-main/
  |-__init__.py
  |-errors.py
  |-forms.py
  |-views.py
  • Tow modules: views.py and errors.py
  • Import these modules to associate them with the blueprint.
  • Modules are imported at the bottom to avoid circular dependencies
    • Because views.py and errors.py need to import main blueprint.
In [ ]:
# app/main/__init__.py  (Blueprint creation)

from flask import Blueprint

main = Blueprint('main', __name__)

# can use from main import views
from . import views, errors

Writing view function inside a blueprint

  • The route decorator comes from the blueprint.
  • Multiple blueprints can define view functions with the same endpoint names.
  • Change the endpoint name of url_for() function.
    • The namesapce of the blueprint: [name of the blueprint].[name of the view function]
    • ex: index -> main.index
    • ex: url_for('index') -> url_for('main.index')
  • Short hand notation url_for('.index'): the blueprint for the current request is used.
    • Redirect within the same blueprint can use the shorter form.
    • Redirect acrss blueprints must use the verbose namespcaed endpoint name.
In [ ]:
# app/main/views.py (Blueprint with application routes)

from flask import render_template, session, redirect, url_for, current_app
from .. import db
from ..models import User
from ..email import send_email
from . import main
from .forms import NameForm


@main.route('/', methods=['GET', 'POST'])
def index():
    form = NameForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.name.data).first()
        if user is None:
            user = User(username=form.name.data)
            db.session.add(user)
            session['known'] = False
            if current_app.config['FLASKY_ADMIN']:
                send_email(current_app.config['FLASKY_ADMIN'], 'New User',
                           'mail/new_user', user=user)
        else:
            session['known'] = True
        session['name'] = form.name.data
        return redirect(url_for('.index'))
    return render_template('index.html',
                           form=form, name=session.get('name'),
                           known=session.get('known', False))

Writing error handlers in a blueprint

  • .errorhandler: only be invoked for errors that originate in the blueprint
  • .app_errorhandler: invoked for errors application-wide.
In [ ]:
# app/main/errors.py (Blueprint with error handlers)

from flask import render_template
from . import main


@main.app_errorhandler(404)
def page_not_found(e):
    return render_template('404.html'), 404


@main.app_errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

Register blueprint

  • The blueprint is registered with the application inside the create_app() factory function.
In [ ]:
# app/__init__.py

from flask import Flask
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_moment import Moment
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)

    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    
    # blueprint registration
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app

Forms module

  • Move forms inside the blueprint as a module.
In [ ]:
# app/main/forms.py

from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import Required


class NameForm(FlaskForm):
    name = StringField('What is your name?', validators=[Required()])
    submit = SubmitField('Submit')

Models

In [ ]:
# app/models.py

from . import db

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    users = db.relationship('User', backref='role', lazy='dynamic')

    def __repr__(self):
        return '<Role %r>' % self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True, index=True)
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username

email.py

In [ ]:
from threading import Thread
from flask import current_app, render_template
from flask_mail import Message

from . import mail


def send_async_email(app, msg):
    with app.app_context():
        mail.send(msg)


def send_email(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + ' ' + subject,
                  sender=app.config['FLASKY_MAIL_SENDER'], recipients=[to])
    msg.body = render_template(template + '.txt', **kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email, args=[app, msg])
    thr.start()
    return thr

Static files and template

  • Move templates and static folders under app folder.
|-app/
    |-templates/
    |-static/
    |-main/
      |-__init__.py
      |-errors.py
      |-forms.py
      |-views.py
    |-__init__.py
    |-email.py
    |-models.py

manage.py

  • A launch script to start the application.
  • The configuration is taken from the environment variable FLASK_CONFIG.
  • Import Flask-Script, Flask-Migrate and define the custom context for Python shell.
  • Add a shebang line so on Unix-based OS the script can be executed as ./manage.py.
In [ ]:
# manage.py

#!/usr/bin/env python
import os
from app import create_app, db
from app.models import User, Role
from flask_script import Manager, Shell
from flask_migrate import Migrate, MigrateCommand

app = create_app(os.getenv('FLASK_CONFIG') or 'default')
manager = Manager(app)
migrate = Migrate(app, db)


def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)
manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command('db', MigrateCommand)


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

tests/

  • We write two simple tests in the test_basics.py here.
  • setUp() and tearDown() run before and after each test.
  • Any method that begins with test_ are executed as tests.
|-tests/
    |-__init__.py
    |-test_basics.py
  • Add an empyt tests/__init__.py file to make the tests folder a package.
  • unittest package will scan all the modules and locate the tests automatically.

Learn more about unittest package: https://docs.python.org/2/library/unittest.html

In [ ]:
# tests/test_basics.py

import unittest
from flask import current_app
from app import create_app, db


class BasicsTestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()

    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
    
    # Ensures that application instance exists
    def test_app_exists(self):
        self.assertFalse(current_app is None)
    
    # Ensures that the application is running testing config
    def test_app_is_testing(self):
        self.assertTrue(current_app.config['TESTING'])

Add custom test command

  • Add a custom test command to the manage.py script.
In [ ]:
# manage.py

@manager.command
def test():
    """Run the unit tests."""
    import unittest
    tests = unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

Database Setup

  • The database URL is taken from an environment variables.
  • If not given, SQLite database is set to default.
In [ ]:
# config.py

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')


class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data-test.sqlite')


class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')

migrations/

  • Run these command to migrate database:
    1. python manage.py db init
    2. python manage.py db migrate -m “Initial migration”
    3. python manage.py db upgrade

Launch the Application

  • python manage.py runserver

You can get the above code from github