您现在的位置：程式碼之家 >> 編程語言 > >> 更多程式語言 >> 更多程式設計知識

Flask-SQLAlchemy 數據庫操作全攻略：從模型定義到遷移管理實戰

編輯：更多程式設計知識

數據持久化是 Web 應用的核心。本文詳細講解 Flask-SQLAlchemy ORM 的配置、模型定義、數據庫創建、遷移管理、關聯關係建立及實戰查詢技巧，助你構建健壯的數據層。

環境與配置

採用 SQLite 作為輕量級數據庫，配置文件 config.py：

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

SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'app.db')
SQLALCHEMY_MIGRATE_REPO = os.path.join(basedir, 'db_repository')

SQLALCHEMY_DATABASE_URI：數據庫連接 URI，SQLite 單文件存儲
SQLALCHEMY_MIGRATE_REPO：遷移腳本存放目錄

初始化擴展（app/__init__.py）：

from flask import Flask
from flask.ext.sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config.from_object('config')
db = SQLAlchemy(app)
from app import views, models

數據模型定義

app/models.py 定義 User 模型：

from app import db

ROLE_USER = 0
ROLE_ADMIN = 1

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    nickname = db.Column(db.String(64), index=True, unique=True)
    email = db.Column(db.String(120), index=True, unique=True)
    role = db.Column(db.SmallInteger, default=ROLE_USER)

    def __repr__(self):
        return '' % (self.nickname)

db.Column 定義字段類型、約束（主鍵、唯一、索引）
__repr__ 便於調試輸出

數據庫創建與遷移腳本

使用 SQLAlchemy-migrate 管理版本，編寫通用腳本避免手動操作。

創建數據庫 `db_create.py`

#!flask/bin/python
from migrate.versioning import api
from config import SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO
from app import db
import os.path

db.create_all()
if not os.path.exists(SQLALCHEMY_MIGRATE_REPO):
    api.create(SQLALCHEMY_MIGRATE_REPO, 'database repository')
    api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
else:
    api.version_control(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, api.version(SQLALCHEMY_MIGRATE_REPO))

運行 ./db_create.py 生成 app.db 與 db_repository。

遷移腳本 `db_migrate.py`

#!flask/bin/python
import imp
from migrate.versioning import api
from app import db
from config import SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO

migration = SQLALCHEMY_MIGRATE_REPO + '/versions/%03d_migration.py' % (api.db_version(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO) + 1)
tmp_module = imp.new_module('old_model')
old_model = api.create_model(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
exec old_model in tmp_module.__dict__
script = api.make_update_script_for_model(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, tmp_module.meta, db.metadata)
open(migration, "wt").write(script)
api.upgrade(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
print 'New migration saved as ' + migration
print 'Current database version: ' + str(api.db_version(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO))

運行 ./db_migrate.py 自動對比模型與數據庫差異，生成遷移腳本並升級。輸出示例：

New migration saved as db_repository/versions/001_migration.py
Current database version: 1

升級與回滾操作

升級 db_upgrade.py：

api.upgrade(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)

回滾 db_downgrade.py：

v = api.db_version(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO)
api.downgrade(SQLALCHEMY_DATABASE_URI, SQLALCHEMY_MIGRATE_REPO, v - 1)

部署新版本時，僅需複製遷移腳本至生產服務器並運行升級腳本，實現零停機遷移。

關聯關係：一對多建模

添加 Post 模型，建立「用戶-微博」一對多關聯：

class User(db.Model):
    ...
    posts = db.relationship('Post', backref='author', lazy='dynamic')

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.String(140))
    timestamp = db.Column(db.DateTime)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))

    def __repr__(self):
        return '' % (self.body)

user_id + db.ForeignKey：外鍵約束
db.relationship：反向引用，user.posts 獲取該用戶所有微博，post.author 獲取作者對象
lazy='dynamic'：返回查詢對象而非列表，支援鏈式過濾

再次運行 ./db_migrate.py 記錄結構變更（版本 2）。

Python Shell 實戰驗證

啟動交互環境（Linux/Mac: flask/bin/python，Windows: flask/Scripts/python）：

>> from app import db, models
>>> u = models.User(nickname='john', email='john@email.com')
>>> db.session.add(u)
>>> db.session.commit()
>>> u = models.User(nickname='susan', email='susan@email.com')
>>> db.session.add(u)
>>> db.session.commit()

會話機制：多次修改累積在 session 中，commit() 原子提交，錯誤時 rollback() 回滾，保證一致性。

查詢操作示例

>> users = models.User.query.all()
[, ]
>>> for u in users: print u.id, u.nickname
1 john
2 susan

>>> u = models.User.query.get(1)


>>> import datetime
>>> u = models.User.query.get(1)
>>> p = models.Post(body='my first post!', timestamp=datetime.datetime.utcnow(), author=u)
>>> db.session.add(p)
>>> db.session.commit()

注意：直接賦值 author=u，ORM 自動填充 user_id 外鍵。

關聯查詢演示

>> u = models.User.query.get(1)
>>> posts = u.posts.all()
[]
>>> for p in posts: print p.id, p.author.nickname, p.body
1 john my first post!

>>> u = models.User.query.get(2)
>>> u.posts.all()
[]

>>> models.User.query.order_by('nickname desc').all()
[, ]