Flask-SQLAlchemy詳解

Flask-SQLAlchemy

SQLAlchemy是一個關系型數據庫框架，它提供了高層的 ORM 和底層的原生數據庫的操作。flask-sqlalchemy 是一個簡化了 SQLAlchemy 操作的flask擴展。

文檔地址：http://docs.jinkan.org/docs/flask-sqlalchemy

1、安裝及設置

# 安裝 flask-sqlalchemy
pip install flask-sqlalchemy
# 如果連接的是 mysql 數據庫，需要安裝 mysqldb
pip install flask-mysqldb

2、數據庫連接設置

• 在 Flask-SQLAlchemy 中，數據庫使用URL指定，而且程序使用的數據庫必須保存到Flask配置對象的 SQLALCHEMY_DATABASE_URI 鍵中
• 其他數據庫見文檔：https://docs.sqlalchemy.org/en/13/core/engines.html

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test'    
# oracle://scott:tiger@127.0.0.1:1521/test
# mysql://scott:tiger@localhost/mydatabase
# postgresql://scott:tiger@localhost/mydatabase
# sqlite:////absolute/path/to/foo.db   注意開頭四個斜杠

# 動態追蹤修改設置，如未設置只會提示警告 

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True 

#查詢時會顯示原始SQL語句 
app.config['SQLALCHEMY_ECHO'] = True

• 其他配置

名字	備注
SQLALCHEMY_DATABASE_URI	用於連接的數據庫 URI 。例如:sqlite:////tmp/test.dbmysql://username:password@server/db
SQLALCHEMY_BINDS	一個映射 binds 到連接 URI 的字典。
SQLALCHEMY_ECHO	如果設置為Ture， SQLAlchemy 會記錄所有 發給 stderr 的語句，這對調試有用。(打印sql語句)
SQLALCHEMY_RECORD_QUERIES	可以用於顯式地禁用或啟用查詢記錄。查詢記錄 在調試或測試模式自動啟用。更多信息見get_debug_queries()。
SQLALCHEMY_NATIVE_UNICODE	可以用於顯式禁用原生 unicode 支持。當使用 不合適的指定無編碼的數據庫默認值時，這對於 一些數據庫適配器是必須的（比如 Ubuntu 上 某些版本的 PostgreSQL ）。
SQLALCHEMY_POOL_SIZE	數據庫連接池的大小。默認是引擎默認值（通常 是 5 ）
SQLALCHEMY_POOL_TIMEOUT	設定連接池的連接超時時間。默認是 10 。
SQLALCHEMY_POOL_RECYCLE	多少秒后自動回收連接。這對 MySQL 是必要的， 它默認移除閑置多於 8 小時的連接。注意如果 使用了 MySQL ， Flask-SQLALchemy 自動設定 這個值為 2 小時。

3、常用的SQLAlchemy字段類型

類型名	python中類型	說明
Integer	int	普通整數，一般是32位
SmallInteger	int	取值范圍小的整數，一般是16位
BigInteger	int或long	不限制精度的整數
Float	float	浮點數
Numeric	decimal.Decimal	普通整數，一般是32位
String	str	變長字符串
Text	str	變長字符串，對較長或不限長度的字符串做了優化
Unicode	unicode	變長Unicode字符串
UnicodeText	unicode	變長Unicode字符串，對較長或不限長度的字符串做了優化
Boolean	bool	布爾值
Date	datetime.date	時間
Time	datetime.datetime	日期和時間
LargeBinary	str	二進制文件

4、常用的SQLAlchemy列選項

選項名	說明
primary_key	如果為True，代表表的主鍵
unique	如果為True，代表這列不允許出現重復的值
index	如果為True，為這列創建索引，提高查詢效率
nullable	如果為True，允許有空值，如果為False，不允許有空值
default	為這列定義默認值

5、常用的SQLAlchemy關系選項

選項名	說明
backref	在關系的另一模型中添加反向引用
primary join	明確指定兩個模型之間使用的聯結條件
uselist	如果為False，不使用列表，而使用標量值
order_by	指定關系中記錄的排序方式
secondary	指定多對多關系中關系表的名字
secondary join	在SQLAlchemy中無法自行決定時，指定多對多關系中的二級聯結條件

6、數據庫基本操作

• 在Flask-SQLAlchemy中，插入、修改、刪除操作，均由數據庫會話管理。

  • 會話用 db.session 表示。在准備把數據寫入數據庫前，要先將數據添加到會話中然后調用 commit() 方法提交會話。

• 在 Flask-SQLAlchemy 中，查詢操作是通過 query 對象操作數據。

  • 最基本的查詢是返回表中所有數據，可以通過過濾器進行更精確的數據庫查詢。

6.1在視圖函數中定義模型類(定義之前需在數據庫創建test數據庫)

from flask import Flask
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)

#設置連接數據庫的URL
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test'

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

#查詢時會顯示原始SQL語句
app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

class Role(db.Model):
    # 定義表名
    __tablename__ = 'roles'
    # 定義列對象
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    us = db.relationship('User', backref='role')

    #repr()方法顯示一個可讀字符串
    def __repr__(self):
        return 'Role:%s'% self.name

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

    def __repr__(self):
        return 'User:%s'%self.name
if __name__ == '__main__':
    app.run(debug=True)

6.2模型之間的關聯

一對多

class Role(db.Model):
    ...
    #關鍵代碼
    us = db.relationship('User', backref='role', lazy='dynamic')
    ...

class User(db.Model):
    ...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))

• 其中realtionship描述了Role和User的關系。在此文中，第一個參數為對應參照的類"User"
• 第二個參數backref為類User申明新屬性的方法
• 第三個參數lazy決定了什么時候SQLALchemy從數據庫中加載數據
  • 如果設置為子查詢方式(subquery)，則會在加載完Role對象后，就立即加載與其關聯的對象，這樣會讓總查詢數量減少，但如果返回的條目數量很多，就會比較慢
    • 設置為 subquery 的話，role.users 返回所有數據列表
  • 另外,也可以設置為動態方式(dynamic)，這樣關聯對象會在被使用的時候再進行加載，並且在返回前進行過濾，如果返回的對象數很多，或者未來會變得很多，那最好采用這種方式
    • 設置為 dynamic 的話，role.users 返回查詢對象，並沒有做真正的查詢，可以利用查詢對象做其他邏輯，比如：先排序再返回結果

多對多

registrations = db.Table('registrations',  
    db.Column('student_id', db.Integer, db.ForeignKey('students.id')),  
    db.Column('course_id', db.Integer, db.ForeignKey('courses.id'))  
)  
class Course(db.Model):
    ...
class Student(db.Model):
    ...
    courses = db.relationship('Course',secondary=registrations,  
                                    backref='students',  
                                    lazy='dynamic')

6.3 常用的SQLAlchemy查詢過濾器

過濾器	說明
filter()	把過濾器添加到原查詢上，返回一個新查詢
filter_by()	把等值過濾器添加到原查詢上，返回一個新查詢
limit	使用指定的值限定原查詢返回的結果
offset()	偏移原查詢返回的結果，返回一個新查詢
order_by()	根據指定條件對原查詢結果進行排序，返回一個新查詢
group_by()	根據指定條件對原查詢結果進行分組，返回一個新查詢

6.4 常用的SQLAlchemy查詢執行器

方法	說明
all()	以列表形式返回查詢的所有結果
first()	返回查詢的第一個結果，如果未查到，返回None
first_or_404()	返回查詢的第一個結果，如果未查到，返回404
get()	返回指定主鍵對應的行，如不存在，返回None
get_or_404()	返回指定主鍵對應的行，如不存在，返回404
count()	返回查詢結果的數量
paginate()	返回一個Paginate對象，它包含指定范圍內的結果

6.5實例測試

# 創建表：
db.create_all()

# 刪除表
db.drop_all()

# 插入一條數據
ro1 = Role(name='admin')
db.session.add(ro1)
db.session.commit()

#再次插入一條數據
ro2 = Role(name='user')
db.session.add(ro2)
db.session.commit()

# 一次插入多條數據
us1 = User(name='wang',email='[e-mail]',password='123456',role_id=ro1.id)
us2 = User(name='zhang',email='[e-mail]',password='201512',role_id=ro2.id)
us3 = User(name='chen',email='[e-mail]',password='987654',role_id=ro2.id)
db.session.add_all([us1,us2,us3])
db.session.commit()

# 查詢:filter_by精確查詢
# 返回名字等於wang的所有人
User.query.filter_by(name='wang').all()

# first()返回查詢到的第一個對象
User.query.first()

# filter模糊查詢，返回名字結尾字符為g的所有數據。
User.query.filter(User.name.endswith('g')).all()

# get():參數為主鍵，如果主鍵不存在沒有返回內容
User.query.get()

# 邏輯非，返回名字不等於wang的所有數據
User.query.filter(User.name!='wang').all()

# not_ 相當於取反
from sqlalchemy import not_
User.query.filter(not_(User.name=='chen')).all()

#邏輯與，需要導入and，返回and()條件滿足的所有數據
from sqlalchemy import and_
User.query.filter(and_(User.name!='wang',User.email.endswith('163.com'))).all()

# 邏輯或，需要導入or_
from sqlalchemy import or_
User.query.filter(or_(User.name!='wang',User.email.endswith('163.com'))).all()

查詢數據后刪除
user = User.query.first()
db.session.delete(user)
db.session.commit()
User.query.all()


更新數據
user = User.query.first()
user.name = 'dong'
db.session.commit()
User.query.first()

# 關聯查詢示例：
# 角色和用戶的關系是一對多的關系，一個角色可以有多個用戶，一個用戶只能屬於一個角色。

## 查詢角色的所有用戶
#查詢roles表id為1的角色
ro1 = Role.query.get(1)
#查詢該角色的所有用戶
ro1.us.all()

##查詢用戶所屬角色
#查詢users表id為3的用戶
us1 = User.query.get(3)
#查詢用戶屬於什么角色
us1.role

6.6數據顯示&表單添加&表單驗證&刪除數據

（1）數據顯示

定義路由函數，並將 Author 和 Book 的所有結果傳到模板

@app.route('/',methods=['GET','POST'])
def index():
    author = Author.query.all()
    book = Book.query.all()
    return render_template('index.html',author=author,book=book)

模版關鍵代碼

<ul>
    {% for x in author %}
    <li>{{ x }}</li>
    {% endfor %}
</ul>
<hr>
<ul>
    {% for x in book %}
    <li>{{ x }}</li>
    {% endfor %}
</ul>

（2）表單添加

定義表單類

from flask_wtf import FlaskForm
from wtforms.validators import DataRequired
from wtforms import StringField,SubmitField
#創建表單類，用來添加信息
class Append(FlaskForm):
    au_info = StringField(validators=[DataRequired()])
    bk_info = StringField(validators=[DataRequired()])
    submit = SubmitField(u'添加')

傳入至模版中

#創建表單對象
@app.route('/',methods=['GET','POST'])
def index():
    author = Author.query.all()
    book = Book.query.all()
    form = Append()
    return render_template('index.html',author=author,book=book,form=form)

模板中代碼

<form method="post">
    {{ form.csrf_token }}
    <p>作者：{{ form.au_info }}</p>
    <p>書名：{{ form.bk_info }}</p>
    <p>{{ form.submit }}</p>
</form>

（3）表單驗證

@app.route('/', methods=['get', 'post'])
def index():
    append_form = Append()

    if request.method == 'POST':
        if append_form.validate_on_submit():
            author_name = append_form.au_info.data
            book_name = append_form.bk_info.data
            # 判斷數據是否存在
            author = Author.query.filter_by(name=author_name).first()
            if not author:
                try:
                    # 先添加作者
                    author = Author(name=author_name)
                    db.session.add(author)
                    db.session.commit()
                    # 再添加書籍
                    book = Book(name=book_name, author_id=author.id)
                    db.session.add(book)
                    db.session.commit()
                except Exception as e:
                    db.session.rollback()
                    print e
                    flash("數據添加錯誤")
            else:
                book_names = [book.name for book in author.books]
                if book_name in book_names:
                    flash('該作者已存在相同的書名')
                else:
                    try:
                        book = Book(name=book_name, author_id=author.id)
                        db.session.add(book)
                        db.session.commit()
                    except Exception as e:
                        db.session.rollback()
                        print e
                        flash('數據添加錯誤')
        else:
            flash('數據輸入有問題')

    authors = Author.query.all()
    books = Book.query.all()
    return render_template('test2.html', authors=authors, books=books, append_form=append_form)

重新編寫html文件展示列表書籍

<h2>書籍列表</h2>
<ul>
{% for author in authors %}
    <li>
        {{ author.name }}
        <ul>
            {% for book in author.books %}
                <li>{{ book.name }}
            {% else %}
                    <li>無書籍</li>
            {% endfor %}
        </ul>

    </li>
{% endfor %}
</ul>

（4）刪除數據

定義刪除author和book的路由

# 刪除作者
@app.route('/delete_author/<int:author_id>')
def delete_author(author_id):
    author = Author.query.get(author_id)
    if not author:
        flash('數據不存在')
    else:
        try:
            Book.query.filter_by(author_id=author_id).delete()
            db.session.delete(author)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            print e
            flash('操作數據庫失敗')

    return redirect(url_for('index'))


# 刪除書籍
@app.route('/delete_book/<int:book_id>')
def delete_book(book_id):
    book = Book.query.get(book_id)
    if not book:
        flash('數據不存在')
    else:
        try:
            db.session.delete(book)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            print e
            flash('操作數據庫失敗')

    return redirect(url_for('index'))

在模版中添加刪除的 a 標簽鏈接

<h2>書籍列表</h2>
<ul>
{% for author in authors %}
    <li>
        {{ author.name }} <a href="/delete_author/{{ author.id }}">刪除</a>
        <ul>
            {% for book in author.books %}
                <li>{{ book.name }} <a href="/delete_book/{{ book.id }}">刪除</a></li>
            {% else %}
                    <li>無書籍</li>
            {% endfor %}
        </ul>

    </li>
{% endfor %}
</ul>

7、數據庫遷移

• 在開發過程中，需要修改數據庫模型，而且還要在修改之后更新數據庫。最直接的方式就是刪除舊表，但這樣會丟失數據。
• 更好的解決辦法是使用數據庫遷移框架，它可以追蹤數據庫模式的變化，然后把變動應用到數據庫中。
• 在Flask中可以使用Flask-Migrate擴展，來實現數據遷移。並且集成到Flask-Script中，所有操作通過命令就能完成。
• 為了導出數據庫遷移命令，Flask-Migrate提供了一個MigrateCommand類，可以附加到flask-script的manager對象上。

# 安裝Flask-Migrate。
pip install flask-migrate

# 代碼如下
#coding=utf-8
from flask import Flask

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate,MigrateCommand
from flask_script import Shell,Manager

app = Flask(__name__)
manager = Manager(app)

app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:mysql@127.0.0.1:3306/test'
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)

#第一個參數是Flask的實例，第二個參數是Sqlalchemy數據庫實例
migrate = Migrate(app,db) 

#manager是Flask-Script的實例，這條語句在flask-Script中添加一個db命令
manager.add_command('db',MigrateCommand)

#定義模型Role
class Role(db.Model):
    # 定義表名
    __tablename__ = 'roles'
    # 定義列對象
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    user = db.relationship('User', backref='role')

    #repr()方法顯示一個可讀字符串，
    def __repr__(self):
        return 'Role:'.format(self.name)

#定義用戶
class User(db.Model):
    __talbe__ = '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:'.format(self.username)


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

創建遷移倉庫

#這個命令會創建migrations文件夾，所有遷移文件都放在里面。
python database.py db init

創建遷移腳本

• 自動創建遷移腳本有兩個函數
  • upgrade()：函數把遷移中的改動應用到數據庫中。
  • downgrade()：函數則將改動刪除。
• 自動創建的遷移腳本會根據模型定義和數據庫當前狀態的差異，生成upgrade()和downgrade()函數的內容。
• 對比不一定完全正確，有可能會遺漏一些細節，需要進行檢查

python database.py db migrate -m 'initial migration'

更新數據庫

python database.py db upgrade

返回以前的版本

可以根據history命令找到版本號,然后傳給downgrade命令:

python app.py db history

輸出格式：<base> ->  版本號 (head), initial migration

回滾到指定版本

python app.py db downgrade 版本號

實際操作順序:

• 1.python 文件 db init
• 2.python 文件 db migrate -m"版本名(注釋)"
• 3.python 文件 db upgrade 然后觀察表結構
• 4.根據需求修改模型
• 5.python 文件 db migrate -m"新版本名(注釋)"
• 6.python 文件 db upgrade 然后觀察表結構
• 7.若返回版本,則利用 python 文件 db history查看版本號
• 8.python 文件 db downgrade(upgrade) 版本號

8、信號機制

Flask信號機制

• Flask信號(signals, or event hooking)允許特定的發送端通知訂閱者發生了什么（既然知道發生了什么，那我們可以根據自己業務需求實現自己的邏輯）。
• Flask提供了一些信號（核心信號）且其它的擴展提供更多的信號。
• 信號依賴於Blinker庫。

pip install blinker

flask內置信號列表：http://docs.jinkan.org/docs/flask/api.html#id17

template_rendered = _signals.signal('template-rendered')
request_started = _signals.signal('request-started')
request_finished = _signals.signal('request-finished')
request_tearing_down = _signals.signal('request-tearing-down')
got_request_exception = _signals.signal('got-request-exception')
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')
appcontext_pushed = _signals.signal('appcontext-pushed')
appcontext_popped = _signals.signal('appcontext-popped')
message_flashed = _signals.signal('message-flashed')

信號應用場景

Flask-User 這個擴展中定義了名為 user_logged_in 的信號，當用戶成功登入之后，這個信號會被發送。我們可以訂閱該信號去追蹤登錄次數和登錄IP：

from flask import request
from flask_user.signals import user_logged_in

@user_logged_in.connect_via(app)
def track_logins(sender, user, **extra):
    user.login_count += 1
    user.last_login_ip = request.remote_addr
    db.session.add(user)
    db.session.commit()

Flask-SQLAlchemy 信號支持

在 Flask-SQLAlchemy 模塊中，0.10 版本開始支持信號，可以連接到信號來獲取到底發生什么了的通知。存在於下面兩個信號：

• models_committed
  • 這個信號在修改的模型提交到數據庫時發出。發送者是發送修改的應用，模型 和 操作描述符 以 (model, operation) 形式作為元組，這樣的元組列表傳遞給接受者的 changes 參數。
  • 該模型是發送到數據庫的模型實例，當一個模型已經插入，操作是 'insert' ，而已刪除是 'delete' ，如果更新了任何列，會是 'update' 。
• before_models_committed
  • 除了剛好在提交發送前發生，與 models_committed 完全相同。

from flask_sqlalchemy import models_committed

# 給 models_committed 信號添加一個訂閱者，即為當前 app
@models_committed.connect_via(app)
def models_committed(a, changes):
    print(a, changes)

9、常見關系模板代碼

以下羅列了使用關系型數據庫中常見關系定義模板代碼

一對多

• 示例場景：
  • 用戶與其發布的帖子(用戶表與帖子表)
  • 角色與所屬於該角色的用戶(角色表與多用戶表)
• 示例代碼

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

class User(db.Model):
    """用戶表"""
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, index=True)

多對多

• 示例場景
  • 講師與其上課的班級(講師表與班級表)
  • 用戶與其收藏的新聞(用戶表與新聞表)
  • 學生與其選修的課程(學生表與選修課程表)
• 示例代碼

tb_student_course = db.Table('tb_student_course',
                             db.Column('student_id', db.Integer, db.ForeignKey('students.id')),
                             db.Column('course_id', db.Integer, db.ForeignKey('courses.id'))
                             )

class Student(db.Model):
    __tablename__ = "students"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

    courses = db.relationship('Course', secondary=tb_student_course,
                              backref=db.backref('students', lazy='dynamic'),
                              lazy='dynamic')

class Course(db.Model):
    __tablename__ = "courses"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

自關聯一對多

• 示例場景
  • 評論與該評論的子評論(評論表)
  • 參考網易新聞
• 示例代碼

class Comment(db.Model):
    """評論"""
    __tablename__ = "comments"

    id = db.Column(db.Integer, primary_key=True)
    # 評論內容
    content = db.Column(db.Text, nullable=False)
    # 父評論id
    parent_id = db.Column(db.Integer, db.ForeignKey("comments.id"))
    # 父評論(也是評論模型)
    parent = db.relationship("Comment", remote_side=[id],
                             backref=db.backref('childs', lazy='dynamic'))

# 測試代碼
if __name__ == '__main__':
    db.drop_all()
    db.create_all()

    com1 = Comment(content='我是主評論1')
    com2 = Comment(content='我是主評論2')
    com11 = Comment(content='我是回復主評論1的子評論1')
    com11.parent = com1
    com12 = Comment(content='我是回復主評論1的子評論2')
    com12.parent = com1

    db.session.add_all([com1, com2, com11, com12])
    db.session.commit()
    app.run(debug=True)

自關聯多對多

• 示例場景

  • 用戶關注其他用戶(用戶表，中間表)

• 示例代碼

tb_user_follows = db.Table(
    "tb_user_follows",
    db.Column('follower_id', db.Integer, db.ForeignKey('info_user.id'), primary_key=True),  # 粉絲id
    db.Column('followed_id', db.Integer, db.ForeignKey('info_user.id'), primary_key=True)  # 被關注人的id
)

class User(db.Model):
    """用戶表"""
    __tablename__ = "info_user"

    id = db.Column(db.Integer, primary_key=True)  
    name = db.Column(db.String(32), unique=True, nullable=False)

    # 用戶所有的粉絲，添加了反向引用followed，代表用戶都關注了哪些人
    followers = db.relationship('User',
                                secondary=tb_user_follows,
                                primaryjoin=id == tb_user_follows.c.followed_id,
                                secondaryjoin=id == tb_user_follows.c.follower_id,
                                backref=db.backref('followed', lazy='dynamic'),
                                lazy='dynamic')