flask_sqlalchemy

簡介：

flask_sqlalchemy 是 Flask Web 框架的一個擴展。

使 Web 服務器以 (ORM: Object Relational Mapping) 對象關系映射的方式與數據庫交互。

一、安裝 flask_sqlalchemy

shell > pip install Flask-SQLAlchemy

shell > ipython

import flask_sqlalchemy

print(flask_sqlalchemy.__version__)
2.1

# 如果沒有報錯，那就是安裝成功了。

二、配置 flask_sqlalchemy 連接數據庫

# app.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

# app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://tmp:9898@192.168.1.40/tmp_db?charset=utf8"
# app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

db = SQLAlchemy(app)


@app.route("/")
def index():
    return "Hello World"

if __name__ == "__main__":
    db.create_all()
    app.run(debug=True)

# 上面注釋掉的兩行代碼使用的數據庫類型為 Mysql，連接數據的驅動是 PyMySQL (pip install PyMySQL)
# 如果是 Python 2.x，那么驅動可能是 MySQLdb，而不是 PyMySQL

# 啟動這個 APP 的時候，會實例化一個對象 db，並且將該 app 注冊到了 db 中，然后 db.create_all() 代表生成數據表 (雖然這里沒有創建表)。

三、創建表

# app.py

class User(db.Model):
    """user 用戶表"""

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

    def __init__(self):
        self.name = self.address = ""

    def __repr__(self):
        return "<Name: %r>" % self.name
        
# 重啟服務器，然后去數據庫中查看，就會多了一張名為 user 的表，且有三個字段 id, name, address

# 所以說，ORM 中一張表就對應一個 Python class ！

四、增、查、更、刪

# ipython

from app import db, User

1、增

u = User()
u.name = "小王"
u.address = "BJ"

db.session.add(u)
db.session.commit()
db.session.close()

# 實例化一個對象，然后給該對象的屬性賦值，最后將該對象添加到數據庫會話中，提交事務，關閉會話即可！

# 這樣就新增了一條數據到數據庫中了。

2、查

print(User.query.all())

# 查找 user 表中所有數據，注意：返回的是一個對象的列表

print(User.query.first())

# 查詢 user 表中第一條數據，毫不意外也是個對象

print(User.query.filter_by(id=1).first())
print(User.query.filter_by(id=1, name="小王").first())

# 查詢 id 等於 1 的數據，只顯示第一條。多個條件也是可以的。但格式只能是 key=value 的方式，多條件關系為 and。

print(User.query.filter(User.id == 1).first())
print(User.query.filter(User.id == 1, User.name == "小王").first())

# 先注意一下 filter 與 filter_by 參數寫法上的區別。
# 另外再注意一下：filter 是 不 支 持 x and x 或者 x or x 這樣的操作的，雖然這樣寫不報錯...

# filter 支持的操作是 > < = != 這些，當然還有上面 filter_by 的那種關系 x, y -> x and y。

# 那要用這種 and、or 怎么辦 ？

from sqlalchemy import and_, or_

print(User.query.filter(and_(User.id == 1, User.address == "BJ")).first())
print(User.query.filter(or_(User.id == 1, User.address == "SH")).first())

# 對，就這么搞

print(User.query.filter_by(id=1).value("name"))
print(list(User.query.filter_by(id=2).values("name", "address")))

# 輸出匹配數據的指定字段，默認是 select * form xxx，現在是 select name, address from xxx。
# 又要注意了：.value 只能用於找到的對象是一個時，例如找到多個對象，則 .value 只返回第一個對象的值。

from sqlalchemy import desc

print(User.query.order_by(User.id).all())
print(User.query.order_by(desc(User.id)).all())

# 以 User.id 升序、降序排列

print(User.query.limit(10).all()) == "limit 10" == "limit 0, 10"
print(User.query.limit(10).offset(5).all()) == "limit 5, 10"
print(User.query.slice(0, 10).all()) == "limit 0, 10"

# 這不是 limit 么，從 offset 取，取 limit 條

# 腦經急轉彎：什么時候 0 等於 1 ？
# 答案：limit 的時候！( limit 0, 10 其實是從第一條開始取十條 ) PS：還有 list index !

3、更

*
u = User.query.filter_by(id=1).first()
u.name = "天王"

db.session.add(u)
db.session.commit()
db.session.close()

**
User.query.filter_by(id=2).update({"name": "地虎", "address": "SH"})

db.session.commit()
db.session.close()

# 兩種更新方法都在這里啦！

4、刪

u = User.query.filter_by(id=1).first()

db.session.delete(u)
db.session.commit()
db.session.close()

# 完事！

五、一對多

# app.py

class User(db.Model):
    """user 用戶表"""

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

    def __init__(self):
        self.name = self.address = ""

    def __repr__(self):
        return "<Name: %r>" % self.name


class Article(db.Model):
    """article 文章表"""
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(64), nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    
    def __init__(self):
        self.title = ""
        
    def __repr__(self):
        return "<Title: %r>" % self.title

# 創建了兩張表 user、article，且 article 表的 author_id 字段通過外鍵關聯到了 user 表的 id 字段！

# 好了，關系已經建立，標准的一對多(一個作者對應多篇文章)

# ipython

from app import db, User, Article

print(User.query.all())
[<Name: '小王'>, <Name: '大王'>]

print(Article.query.all())
[<Title: 'flask 學習筆記'>, <Title: '敢問路在何方'>, <Title: 'python 從入門到精通'>]

# 通過上面學到的技能，新增了兩個用戶、三篇文章

# 需求1：看看 "敢問路在何方" 的作者是誰？

article = Article.query.filter_by(title="敢問路在何方").first()
user = User.query.filter_by(id=at.author_id).first()

print(user.name)

# 需求2：看看 "小王" 寫了哪些文章？

uid = User.query.filter_by(name="小王").first()
article_list = Article.query.filter_by(author_id=uid).all()

for obj in article_list:
    print(obj.title)

# 有沒有更方便的方法呢 ？有

# app

class Article(db.Model):
    """article 文章表"""

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(64), nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)

    user = db.relationship("User", backref="articles")

    def __init__(self):
        self.title = ""

    def __repr__(self):
        return "<Title: %r>" % self.title

# 只需給 Article 類加一個屬性即可。注意：user 並不是一個存在的字段，而是一個屬性！
# 該屬性關聯到了 User 類，backref 指定反向引用的名稱為 articles，即 User 表訪問這個屬性就可以找到該用戶的所有文章！

from app import db, User, Article

# 需求1：

article = Article.query.filter_by(title="敢問路在何方").first()

print(article.user.name)

# 需求2:

user = User.query.filter_by(name="小王").first()

for obj in user.articles:
    print(obj.title)
    
# 方便不？只查詢一張表即可！

六、多對多

# app.py

article_tag = db.Table(
    "article_tag",
    db.Column("article_id", db.Integer, db.ForeignKey("article.id"), primary_key=True),
    db.Column("tag_id", db.Integer,db.ForeignKey("tag.id"), primary_key=True)
)

# 多對多關系需要建立一個中間表，你看建表的方式跟原來的還不一樣！
# 表名，字段，類型 需要跟外鍵關聯的字段類型一致，外鍵 關聯到哪張表的哪個字段，主鍵

class Article(db.Model):
    """article 文章表"""

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(64), nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)

    user = db.relationship("User", backref="articles")
    tags = db.relationship("Tag", secondary="article_tag", backref="articles")

    def __init__(self):
        self.title = ""

    def __repr__(self):
        return "<Title: %r>" % self.title

# 又加了一個屬性 tags，多了一個參數 secondary 關鍵字指向中間表

class Tag(db.Model):
    """tag 標簽表"""

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

    def __init__(self):
        self.name = ""

    def __repr__(self):
        return "<Name: %r>" % self.name
        
# 多了一張 tag 表，這就是張普通表，用來存儲標簽信息，有個隱藏屬性 articles 哦！

# 多對多：一篇文章有多個標簽，一個標簽又能標記給多篇文章

# ipython

from app import db, Article, Tag

t = Tag()
t.name = "python"

an = Article()
an.title = "python 數據分析"
an.author_id = 1
an.tags.append(t) or an.tags.append(Tag.query.filter_by(id=1).first())

db.session.add(t)
db.session.add(an)
db.session.commit()
db.session.close()

# 這就新加了一個標簽、一篇文章，且給這篇文章加了一個標簽

an = Article.query.filter_by(title="python 數據分析").first()

for obj in an.tags:
    print(obj.name)

# 這樣就查到了這篇文章的所有標簽

tn = Tag.query.filter_by(id=1).first()

for obj in tn.articles:
    print(obj.title)
    
# 這樣就查到了這個標簽的所有文章

# End

# 好了，這就是 flask_sqlalchemy 的連接方式、基本增、刪、改、查、一對多、多對多的一些文字！全部代碼如下：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

# app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://tmp:9898@192.168.1.40/tmp_db?charset=utf8"
# app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True

db = SQLAlchemy(app)

article_tag = db.Table(
    "article_tag",
    db.Column("article_id", db.Integer, db.ForeignKey("article.id"), primary_key=True),
    db.Column("tag_id", db.Integer,db.ForeignKey("tag.id"), primary_key=True)
)


class User(db.Model):
    """user 用戶表"""

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

    def __init__(self):
        self.name = self.address = ""

    def __repr__(self):
        return "<Name: %r>" % self.name


class Article(db.Model):
    """article 文章表"""

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(64), nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)

    user = db.relationship("User", backref="articles")
    tags = db.relationship("Tag", secondary="article_tag", backref="articles")

    def __init__(self):
        self.title = ""

    def __repr__(self):
        return "<Title: %r>" % self.title


class Tag(db.Model):
    """tag 標簽表"""

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

    def __init__(self):
        self.name = ""

    def __repr__(self):
        return "<Name: %r>" % self.name


@app.route("/")
def index():
    return "Hello World"

if __name__ == "__main__":
    db.create_all()
    app.run(debug=True)