$ pip install flask-sqlalchemy

app.config["SQLALCHEMY_DATABASE_URI"] = DATABASE_URI
app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True/False  # 每次請(qǐng)求結(jié)束后都會(huì)自動(dòng)提交數(shù)據(jù)庫(kù)中的變動(dòng).
 
app.config[""] = 
app.config[""] = 
app.config[""] = 
app.config[""] = 
 
DATABASE_URI :
    mysql : mysql://username:password@hostname/database
 
    pgsql : postgresql://username:password@hostname/database
 
    sqlite(linux)  : sqlite:////absolute/path/to/database
 
    sqlite(windows) : sqlite:///c:/absolute/path/to/database

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
base_dir = os.path.abspath(os.path.dirname(__file__))
 
app = Flask(__name__)
 
app.config["SQLALCHEMY_DATABASE_URI"] = 'sqlite:///' + os.path.join(base_dir, 'data.sqlite')
app.config["SQLALCHEMY_COMMIT_ON_TEARDOWN"] = True
 
db = SQLAlchemy(app)

db.Model    # 創(chuàng)建模型,
db.Column   # 創(chuàng)建模型屬性.

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):
        """非必須, 用于在調(diào)試或測(cè)試時(shí), 返回一個(gè)具有可讀性的字符串表示模型."""
        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)
 
    def __repr__(self):          
        """非必須, 用于在調(diào)試或測(cè)試時(shí), 返回一個(gè)具有可讀性的字符串表示模型."""
        return '<Role %r>' % self.username

class Role(db.Model):
    # ...
    users = db.relationship('User', backref='role')
 
class User(db.Model):
    # ...
    role_id = db.Column(db.Integer, db.ForeignKey('roles.id'))  # 外鍵關(guān)系.
 
 
###############
db.ForeignKey('roles.id') : 外鍵關(guān)系,
 
Role.users = db.relationship('User', backref='role') : 代表 外鍵關(guān)系的 面向?qū)ο笠暯? 對(duì)于一個(gè) Role 類的實(shí)例, 
其 users 屬性將返回與角色相關(guān)聯(lián)的用戶組成的列表. 
    db.relationship() 第一個(gè)參數(shù)表示這個(gè)關(guān)系的另一端是哪個(gè)模型. 
    backref 參數(shù), 向 User 模型添加了一個(gè) role 數(shù)據(jù)屬性, 從而定義反向關(guān)系.  這一屬性可替代 role_id 訪問 Role 模型, 
    此時(shí)獲取的是模型對(duì)象, 而不是外鍵的值.

registrations = db.Table("registrations",
                         db.Column("student_id", db.Integer, db.ForeignKey("students.id")),
                         db.Column("class_id", db.Integer, db.ForeignKey("classes.id"))
                         )
 
class Student(db.Model):
    __tablename__ = "students"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)
    classes = db.relationship("Class",
                              secondary=registrations,
                              backref=db.backref("students", lazy="dynamic"),
                              lazy="dynamic")
 
class Class(db.Model):
    __tablename__ = "classes"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)

class Follow(db.Model):
    __tablename__ = "follows"
    follower_id = db.Column(db.Integer, db.ForeignKey("users.id"), primary_key=True)
    followed_id = db.Column(db.Integer, db.ForeignKey("users.id"), primary_key=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow)
 
# SQLAlchemy 不能直接使用這個(gè)關(guān)聯(lián)表, 因?yàn)槿绻@個(gè)做程序就無(wú)法訪問其中的自定義字段. 相反的, 要把這個(gè)多對(duì)多關(guān)系的左右
兩側(cè)拆分成兩個(gè)基本的一對(duì)多關(guān)系, 而且要定義成標(biāo)準(zhǔn)的關(guān)系。

class User(UserMixin, db.Model):
    # ...
    followd = db.relationship("Follow",
                              foreign_keys=[Follow.follower_id],
                              backref=db.backref("follower", lazy="joined"),
                              lazy="dynamic",
                              cascade="all, delete-orphan")
    followrs = db.relationship("Follow",
                              foreign_keys=[Follow.followed_id],
                              backref=db.backref("followed", lazy="joined"),
                              lazy="dynamic",
                              cascade="all, delete-orphan")
# 這段代碼中, followed 和 follower 關(guān)系都定義為 單獨(dú)的 一對(duì)多關(guān)系. 
# 注意: 為了消除外鍵歧義, 定義關(guān)系是必須使用可選參數(shù) foreign_keys 指定的外鍵. 而且 db.backref() 參數(shù)并不是指定這兩個(gè)
關(guān)系之間的引用關(guān)系, 而是回引 Follow 模型. 回引中的 lazy="joined" , 該模式可以實(shí)現(xiàn)立即從連接查詢中加載相關(guān)對(duì)象.
# 這兩個(gè)關(guān)系中, user 一側(cè)設(shè)定的 lazy 參數(shù)作用不一樣. lazy 參數(shù)都在 "一" 這一側(cè)設(shè)定, 返回的結(jié)果是 "多" 這一側(cè)中的記錄. 
dynamic 參數(shù), 返回的是查詢對(duì)象.
# cascade 參數(shù)配置在父對(duì)象上執(zhí)行的操作相關(guān)對(duì)象的影響. 比如, 層疊對(duì)象可設(shè)定為: 將用戶添加到數(shù)據(jù)庫(kù)會(huì)話后, 要自定把所有
關(guān)系的對(duì)象都添加到會(huì)話中. 刪除對(duì)象時(shí), 默認(rèn)的層疊行為是把對(duì)象聯(lián)結(jié)的所有相關(guān)對(duì)象的外鍵設(shè)為空值. 但在關(guān)聯(lián)表中, 刪除記錄
后正確的行為是把執(zhí)行該記錄的實(shí)體也刪除, 因?yàn)檫@樣才能有效銷毀聯(lián)結(jié). 這就是 層疊選項(xiàng)值 delete-orphan 的作用. 設(shè)為 all, 
delete-orphan 的意思是啟動(dòng)所有默認(rèn)層疊選項(xiàng), 并且還要?jiǎng)h除孤兒記錄。

db.create_all()

$ python myflask.py shell
> from myflask import db
> db.create_all()

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

$ pip install flask-migrate

from flask_migrate import Migrate, MigrateCommand
 
# ...
migrate = Migrate(app, db)
manager.add_command('db', MigrateCommand)

b. 創(chuàng)建數(shù)據(jù)路遷移腳本.
    $ python myflask.py db revision     # 手動(dòng)創(chuàng)建 Alemic 遷移
    創(chuàng)建的遷移只是一個(gè)骨架, upgrade() 和 downgrade() 函數(shù)都是空的. 開發(fā)者需要使用 Alembic 提供的 Operations 對(duì)象
    指令實(shí)現(xiàn)具體操作.
 
    $ python myflask.py db migrate -m COMMONT    # 自動(dòng)創(chuàng)建遷移.
    自動(dòng)創(chuàng)建的遷移會(huì)根據(jù)模型定義和數(shù)據(jù)庫(kù)當(dāng)前的狀態(tài)之間的差異生成 upgrade() 和 downgrade() 函數(shù)的內(nèi)容.
 
    ** 自動(dòng)創(chuàng)建的遷移不一定總是正確的, 有可能漏掉一些細(xì)節(jié), 自動(dòng)生成遷移腳本后一定要進(jìn)行檢查.
 
 
c. 更新數(shù)據(jù)庫(kù)
    $ python myflask.py db upgrade     # 將遷移應(yīng)用到數(shù)據(jù)庫(kù)中.

>> from myflask import db, User, Role
>> db.create_all()
>> admin_role = Role(name="Admin")
>> mod_role = Role(name="Moderator")
>> user_role = Role(name="User")
>> user_john = User(username="john", role=admin_role)
>> user_susan = User(username="susan", role=mod_role)
>> user_david = User(username="david", role=user_role)
>> admin_role.name
'Admin'
>> admin_role.id
None
---------
>> db.session.add_all([admin_role, mod_role, user_role, user_john, user_susan, user_david])  # 把對(duì)象添加到會(huì)話中.
>> db.session.commit()      # 把對(duì)象寫入數(shù)據(jù)庫(kù), 使用 commit() 提交會(huì)話.

>> admin_role = "Administrator"
>> db.session.add(admin_role)
>> db.session.commit()

>> db.session.delete(mod_role)
>> db.session.commit()

>> Role.query.all()
  [<Role u'Admin'>, <Role u'Moderator'>, <Role u'User'>]
>> User.query.all()
  [<Role u'john'>, <Role u'susan'>, <Role u'david'>]

>> User.query.filter_by(role=user_role).all()   # 以列表形式,返回所有結(jié)果,
>> User.query.filter_by(role=user_role).first() # 返回結(jié)果中的第一個(gè).

# WHERE age<20
users = User.query.filter(User.age<20)
# WHERE name LIKE 'J%' AND age<20
users = User.query.filter(User.name.startswith('J'), User.age<20)

>> db.session.add(ONE)
>> db.session.commit()

>> db.session.add_all([LIST_OF_MEMBER])
>> db.session.commit()

>> db.session.delete(mod_role)
>> db.session.commit()

>> db.session.rollback()

@app.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
        else:
            session["known"] = True
        session["name"] = form.name.data
        form.name.data = ""             # why empty it ?
        return redirect(url_for("index"))
    return render_template("index.html", current_time=datetime.utcnow(), form=form, name=session.get("name"), 
    known=session.get("known"))

paginate(頁(yè)數(shù)[,per_page=20, error_out=True])
    頁(yè)數(shù) : 唯一必須指定的參數(shù),
    per_page : 指定每頁(yè)現(xiàn)實(shí)的記錄數(shù)量, 默認(rèn) 20.
    error_out : True 如果請(qǐng)求的頁(yè)數(shù)超出了返回, 返回 404 錯(cuò)誤; False 頁(yè)數(shù)超出范圍時(shí)返回一個(gè),空列表.

@main.route("/", methods=["GET", "POST"])
def index():
    # ...
    page = request.args.get('page', 1, type=int)    # 渲染的頁(yè)數(shù), 默認(rèn)第一頁(yè), type=int 保證參數(shù)無(wú)法轉(zhuǎn)換成整數(shù)時(shí), 
    返回默認(rèn)值.
    pagination = Post.query.order_by(Post.timestamp.desc()).paginate(page, per_page=current_app.config
    ["FLASKY_POSTS_PER_PAGE"], error_out=False)
    posts = pagination.items
    return render_template('index.html', form=form, posts=posts,pagination=pagination)

{% macro pagination_widget(pagination,endpoint) %}
<ul class="pagination">
    <li {% if not pagination.has_prev %} class="disabled" {% endif %}>
        <a href="{% if pagination.has_prev %}{{url_for(endpoint, page=paginatin.page - 1, **kwargs)}}{% else %}
        #{% endif %}">
            &laquo;
        </a>
    </li>
    {% for p in pagination,.iter_pages() %}
        {% if p %}
            {% if p == pagination.page %}
            <li class="active">
                <a href="{{ url_for(endpoint, page=p, **kwargs) }}" rel="external nofollow" >{{p}}</a>
            </li>
            {% else %}
            <li>
                <a href="{{ url_for(endpoint, page = p, **kwargs) }}" rel="external nofollow" >{{p}}</a>
            </li>
            {% endif %}
        {% else %}
        <li class="disabled"><a href="#" rel="external nofollow" >&hellip;</a> </li>
        {% endif %}
    {% endfor %}
    <li {% if not pagination.has_next %} class="disabled" {% endif%}>
        <a href="{% if paginatin.has_next %}{{ url_for(endpoint, page=pagination.page+1, **kwargs) }}{% else %}
        #{% endif %}">
            &raquo;
        </a>
    </li>
</ul>
{% endmacro %}

{% extends "base.html" %}
{% import "_macros.html" as macros %}
...
<div class="pagination">
    {{ macro.pagination_widget(pagination, ".index")}}
</div>

from markdown import markdown
import bleach
class Post(db.Model):
    # ...
    body = db.Colume(db.Text)
    body_html = db.Column(db.Text)
    # ...
    @staticmethod
    def on_changeed_body(target, value, oldvalue, initiator):
        allowed_tags = ["a", "abbr", "acronym", "b", "blockquote", "code", "em",
                        "i", "li", "ol", "pre", "strong", "ul", "h1", "h2","h3","h4","p"]
        target.body_html = bleach.linkify(bleach.clean(markdown(value, output_format="html"), tags=allowed_tags, 
        strip=True))
db.event.listen(Post.body, "set", Post.on_changeed_body) 
# on_changed_body 函數(shù)注冊(cè)在 body 字段上, 是 SQLIAlchemy "set" 事件的監(jiān)聽程序, 
# 這意味著只要這個(gè)類實(shí)例的 body 字段設(shè)了新值, 函數(shù)就會(huì)自動(dòng)被調(diào)用. 
# on_changed_body 函數(shù)把 body 字段中的文本渲染成 HTML 格式, 
# 結(jié)果保存在 body_html 中, 自動(dòng)高效的完成 Markdown 文本到 HTML 的轉(zhuǎn)換

# orm.reconstructor
from sqlalchemy import orm
class MyMappedClass(object):
    def __init__(self, data):
        self.data = data
        # we need stuff on all instances, but not in the database.
        self.stuff = []
    @orm.reconstructor
    def init_on_load(self):
        self.stuff = []
# InstanceEvents.load()
from sqlalchemy import event
## standard decorator style
@event.listens_for(SomeClass, 'load')
def receive_load(target, context):
    "listen for the 'load' event"
    # ... (event handling logic) ...

class AwsRegions(db.Model):
    name=db.Column(db.String(64))
    ...
    @property
    def zabbix_api(self):
        return ZabbixObj(zabbix_url)
    @zabbix_api.setter
    def zabbix_api(self):
        raise ValueError("zabbix can not be setted!")