Flask SQLAlchemy管理數(shù)據(jù)庫

關(guān)系型數(shù)據(jù)庫框架，支持多種數(shù)據(jù)庫，提供高層ORM, 也支持執(zhí)行原生SQL

$ pip install flask-sqlalchemy
$ pip install pymysql

數(shù)據(jù)庫連接URL

Mysql

mysql+pymysql://username:password@server/db

創(chuàng)建數(shù)據(jù)庫 mysql控制臺
mysql> create database school4 default character set = utf8;

from flask import Flask
from flask_script import Manager
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 設(shè)置數(shù)據(jù)庫連接字符串
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:Vff123456@127.0.0.1/school4?charset=UTF8MB4'
# 不跟蹤修改，不設(shè)置會有警告
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 創(chuàng)建數(shù)據(jù)庫連接
db = SQLAlchemy(app)

import application.models
import application.views

manager = Manager(app)

定義模型

from application import db
import datetime

# 用戶表
class Users(db.Model):
    __tablename__ = 'users'
    # 主鍵
    id = db.Column(db.Integer, primary_key=True)
    # 用戶名 唯一索引
    username = db.Column(db.String(128), unique=True, nullable=False)
    # 密碼 必填字段
    password = db.Column(db.String(512), nullable=False)
    # 姓名 創(chuàng)建索引，加快查詢
    fullname = db.Column(db.String(128), index=True, nullable=False)
    # 狀態(tài) (1: 生效 0: 禁用)
    status = db.Column(db.SmallInteger, default=1, nullable=False)
    # 創(chuàng)建時間 默認當前時間
    created_time = db.Column(db.DateTime, nullable=False, default=datetime.datetime.utcnow, index=True)

    def __repr__(self):
        return 'username=%s' % username    

根據(jù)model創(chuàng)建表

$ python manage.py shell
>>> from application import db
>>> db.drop_all()
>>> db.create_all()

插入數(shù)據(jù)

$ python manage.py shell
>>> user = Users(username='lisa', password='asdasdfad', fullname='李霞')
>>> db.session.add(user) # 添加到數(shù)據(jù)庫會話中
>>> db.session.commit() # 提交數(shù)據(jù)
>>> db.session.rollback() # 數(shù)據(jù)庫回滾會有效么？試下放在commit前面呢？

查詢數(shù)據(jù)

>>> Users.query.all()  # 查詢?nèi)?>>> Users.query.filter_by(username='lisa') # 返回一個新查詢
>>> Users.query.filter_by(username='lisa').all() # 返回查詢結(jié)果
>>> Users.query.filter(Users.username=='lisa').first_or_404()
>>> str(Users.query.filter(Users.username=='lisa3')) # 查看sql語句
>>> db.session.execute('select * from users where id = 1').first() # 原生sql語句

修改行

user = Users.query.get(1)
user.password = '123412343'
db.session.add(user)
db.session.commit()

刪除行

db.session.delete(user)
db.session.commit()

關(guān)系

一對多關(guān)系

# 學生表
class Student(db.Model):
    __tablename__ = 'student'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), unique=True, nullable=False)
    # 建立指向班級的外鍵
    class_id = db.Column(db.Integer, db.ForeignKey('class.id'))

    def __repr__(self):
        return 'student=%s' % self.name  

# 班級表
class Class(db.Model):
    __tablename__ = 'class'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), unique=True, nullable=False)
    # 引用學生
    students = db.relationship('Student', backref='sclass')

一對多關(guān)系測試

>>> from application import db
>>> db.create_all()
>>> from application.models import *
>>> student1 = Student(name='lisa')
>>> db.session.add(student1)
>>> db.session.commit()
>>> c1 = Class(name='python1904')
>>> db.session.add(c1)
>>> db.session.commit()
>>> student1.sclass = c1

一對一關(guān)系

一對一關(guān)系可以用前面介紹的一對多關(guān)系 表示，但調(diào)用db.relationship() 時要把uselist設(shè)為False

student = db.relationship('Student', backref='sclass', uselist=False)

多對多關(guān)系建模

class StudentLesson(db.Model):
    __tablename__ = 'student_lesson'
    student_id = db.Column(db.Integer, db.ForeignKey('student.id'), primary_key=True)
    lesson_id = db.Column(db.Integer, db.ForeignKey('lesson.id'), primary_key=True)
    score = db.Column(db.Integer, nullable=True)

class Student(db.Model):
    __tablename__ = 'student'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), unique=True, nullable=False)
    lessons = db.relationship('StudentLesson', foreign_keys=[StudentLesson.student_id], backref=db.backref('student', lazy='joined'), lazy='dynamic', cascade='all, delete-orphan') `

    def __repr__(self):
        return 'student=%s' % self.name    

class Lesson(db.Model):
    __tablename__ = 'lesson'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), unique=True, nullable=False)
    students = db.relationship('StudentLesson', foreign_keys=[StudentLesson.lesson_id], backref=db.backref('lesson', lazy='joined'), lazy='dynamic', cascade='all, delete-orphan')

    def __repr__(self):
        return 'lesson=%s' % self.name    

多對多關(guān)系測試

>>> from application.models import *
>>> from application import db
>>> db.create_all()
>>> student1 = Student(name='carmack')
>>> student2 = Student(name='lisa')
>>> db.session.add_all([student1, student2])
>>> db.session.commit()
>>> lesson1 = Lesson(name='數(shù)學')
>>> lesson2 = Lesson(name='語文')
>>> db.session.add_all([lesson1, lesson2])
>>> db.session.commit()
>>> student_lesson = StudentLesson(student=student1, lesson=lesson1)
>>> db.session.add(student_lesson)
>>> db.session.commit()

常用的SQLAlchemy關(guān)系選項

• backref
  在關(guān)系的另一個模型中添加反向引用
• primaryjoin
  明確指定兩個模型之間使用的聯(lián)結(jié)條件。只在模棱兩可的關(guān)系中需要指定
• lazy
  指定如何加載相關(guān)記錄?？蛇x值有 select(首次訪問時按需加載)、immediate(源對象加 載后就加載)、joined(加載記錄，但使用聯(lián)結(jié))、subquery(立即加載，但使用子查詢)， noload(永不加載)和 dynamic(不加載記錄，但提供加載記錄的查詢)
• uselist
  設(shè)為False，不使用列表

使用Flask-Migrate實現(xiàn)數(shù)據(jù)庫遷移

安裝migrate

pip install flask-migrate

配置使用migrate

初始化，使用app和db進行migrate對象的初始化

from flask_migrate import Migrate

#綁定app和數(shù)據(jù)庫
migrate = Migrate(app, db)

在Manager()對象上添加遷移指令

from flask_migrate import Migrate, MigrateCommand

app = Flask(__name__)

manage = Manager(app=app)

manage.add_command(‘db’, MigrateCommand)

操作：

python manage.py db init 初始化出migrations的文件，只調(diào)用一次

python manage.py db migrate -m ‘comment’ 生成遷移文件

python manage.py db upgrade 執(zhí)行遷移文件中的升級

python manage.py db downgrade 執(zhí)行遷移文件中的降級

python manage.py db —help 幫助文檔