SQLAlchemy 增刪改查 一對多 多對多

好久沒有更新Blog了

今天來聊一聊 Python 的 ORM 框架 SQLAlchemy 有些同學已經聽說過這個框架的大名了,也聽說了 SQLAlchemy 沒有 Django 的 Models 好用

我在這里官宣辟謠一下啊, Models 緊緊只是配置和使用比較簡單(這特么就是廢話),因為他是Django自帶的ORM框架,也正是因為是Django原生的,所以兼容性遠遠不如SQLAlchemy

真正算得上全面的ORM框架必然是我們的SQLAlchemy ORM框架,它可以在任何使用SQL查詢時使用

當然了,無論是使用什么ORM框架,都是為了方便不熟練數據庫的同學使用的,我個人還是比較推崇原生 SQL ,也建議每位同學攻克 SQL 的難關

廢話不多說,我們來看一下 SQLAlchemy 如何使用:

1.創建數據表

# ORM中的數據表是什么呢?
# Object Relation Mapping
# Object - Table 通過 Object 去操縱數據表
# 從而引出了我們的第一步創建數據表 - 創建Object
# 1. 創建Object
# class User(object):
#     pass

# 2. 讓Object與數據表產生某種關系 也就是讓Object與數據表格式極度相似
# 導入官宣基礎模型
from sqlalchemy.ext.declarative import declarative_base
# 實例化官宣模型 - Base 就是 ORM 模型
Base = declarative_base()
# 當前的這個Object繼承了Base也就是代表了Object繼承了ORM的模型
class User(Base):  # 相當於 Django Models中的 Model
    # 為Table創建名稱
    __tablename__ = "user"
    # 創建ID數據字段 , 那么ID是不是一個數據列呢? 也就是說創建ID字段 == 創建ID數據列
    from sqlalchemy import Column,Integer,String
    # id = Column(數據類型,索引,主鍵,外鍵,等等)
    # int == Integer
    id = Column(Integer,primary_key=True,autoincrement=True)
    # str == char(長度) == String(長度)
    name = Column(String(32),index=True)

# 3.去數據庫中創建數據表? or 先連接數據庫?
# 3.去連接數據庫 創建數據引擎
from sqlalchemy import create_engine
# 創建的數據庫引擎
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8")

# Base 自動檢索所有繼承Base的ORM 對象 並且創建所有的數據表
Base.metadata.create_all(engine)

 

2.增刪改查操作

2.1.增加數據

#insert 為數據表增加數據
# insert One 增加一行數據
# insert into user(name) values ("DragonFire")
# 在ORM中的操作:
# 1.首先導入之間做好的ORM 對象 User
from my_create_table import User
# 2.使用Users ORM模型創建一條數據
user1 = User(name="DragonFire")
# 數據已經創建完了,但是需要寫入到數據庫中啊,怎么寫入呢?
# 3.寫入數據庫:
# 首先打開數據庫會話 , 說白了就是創建了一個操縱數據庫的窗口
# 導入 sqlalchemy.orm 中的 sessionmaker
from sqlalchemy.orm import sessionmaker
# 導入之前創建好的 create_engine
from my_create_table import engine
# 創建 sessionmaker 會話對象,將數據庫引擎 engine 交給 sessionmaker
Session = sessionmaker(engine)
# 打開會話對象 Session
db_session = Session()
# 在db_session會話中添加一條 UserORM模型創建的數據
db_session.add(user1)
# 使用 db_session 會話提交 , 這里的提交是指將db_session中的所有指令一次性提交
db_session.commit()

# 當然也你也可很任性的提交多條數據
# 方法一:
user2 = User(name="Dragon")
user3 = User(name="Fire")
db_session.add(user2)
db_session.add(user3)
db_session.commit()
# 之前說過commit是將db_session中的所有指令一次性提交,現在的db_session中至少有兩條指令user2和user3
db_session.close()
#關閉會話

# 如果說你覺得方法一很麻煩,那么方法二一定非常非常適合你
# 方法二:
user_list = [
    User(name="Dragon1"),
    User(name="Dragon2"),
    User(name="Dragon3")
]
db_session.add_all(user_list)
db_session.commit()

db_session.close()

2.2.查詢數據

# ORM操作查詢數據
# 有了剛才Insert增加數據的經驗,那么查詢之前的准備工作,就不用再重復了吧
# 回想一下剛才Insert時我們的操作
from my_create_table import User, engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

# 1. select * from user 查詢user表中的所有數據
# 語法是這樣的 使用 db_session 會話 執行User表 query(User) 取出全部數據 all()
user_all_list = db_session.query(User).all()
print(user_all_list)  # [<my_create_table.User object at 0x0000016D7C4BCDD8>]
# 如何查看user_all_list其中的數據呢? 循環唄
for i in user_all_list:
    print(i.id, i.name)  # ORM對象 直接使用調用屬性的方法 拿出對應字段的值

db_session.close()
#關閉會話

# 2. select * from user where id >= 20
# 語法是這樣的 使用 db_session 會話 執行User表 query(User) 篩選內容User.id >=20 的數據全部取出 all()
user_all_list = db_session.query(User).filter(User.id >= 20).all()
print(user_all_list)
for i in user_all_list:
    print(i.id, i.name)

db_session.close()
#關閉會話

# 3. 除了取出全部還可以只取出一條
user = db_session.query(User).filter(User.id >= 20).first()
print(user.id, user.name)
db_session.close()
#關閉會話

# 4. 烏龍 之 忘了取出數據.......
wulong1 = db_session.query(User).filter(User.id >= 20)
print(wulong1)
#SELECT user.id AS user_id, user.name AS user_name
#FROM user
#WHERE user.id >= %(id_1)s
# Fuck我忘了取出數據了!!!!!!! 哎? wulong1給我顯示了原生SQL語句,因禍得福了
wulong2 = db_session.query(User)
print(wulong2)
#SELECT user.id AS user_id, user.name AS user_name
#FROM user
# Fuck我又忘了取出數據了!!!!!!! 哎? wulong2給我顯示了原生SQL語句,因禍得福了
db_session.close()
#關閉會話

2.3.修改數據

# ORM更新數據
# 無論是更新還是刪除,首先要做的事情,就應該是查詢吧
# 根據之前原有的經驗,接下來是不是要導入ORM對象了,是不是要創建db_session會話了
from my_create_table import User,engine
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# UPDATE user SET name="NBDragon" WHERE id=20 更新一條數據
# 語法是這樣的 :
# 使用 db_session 執行User表 query(User) 篩選 User.id = 20 的數據 filter(User.id == 20)
# 將name字段的值改為NBDragon update({"name":"NBDragon"})
res = db_session.query(User).filter(User.id == 20).update({"name":"NBDragon"})
print(res) # 1 res就是我們當前這句更新語句所更新的行數
# 注意注意注意
# 這里一定要將db_session中的執行語句進行提交,因為你這是要對數據中的數據進行操作
# 數據庫中 增 改 刪 都是操作,也就是說執行以上三種操作的時候一定要commit
db_session.commit()
db_session.close()
#關閉會話

# 更新多條
res = db_session.query(User).filter(User.id <= 20).update({"name":"NBDragon"})
print(res) # 6 res就是我們當前這句更新語句所更新的行數
db_session.commit()
db_session.close()
#關閉會話

2.4.刪除數據

# ORM 刪除一條多條數據
# 老規矩
# 導入 ORM 創建會話
from my_create_table import User,engine
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine)
db_session = Session()

# DELETE FROM `user` WHERE id=20
res = db_session.query(User).filter(User.id==20).delete()
print(res)
# 是刪除操作吧,沒錯吧,那你想什么呢?commit吧
db_session.commit()

db_session.close()
#關閉會話

2.5.搞基!高級版查詢操作

# 高級版查詢操作,厲害了哦
#老規矩
from my_create_table import User,engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

# 查詢數據表操作
# and or
from sqlalchemy.sql import and_ , or_
ret = db_session.query(User).filter(and_(User.id > 3, User.name == 'DragonFire')).all()
ret = db_session.query(User).filter(or_(User.id < 2, User.name == 'DragonFire')).all()

# 查詢所有數據
r1 = db_session.query(User).all()

# 查詢數據 指定查詢數據列 加入別名
r2 = db_session.query(User.name.label('username'), User.id).first()
print(r2.id,r2.username) # 15 NBDragon

# 表達式篩選條件
r3 = db_session.query(User).filter(User.name == "DragonFire").all()

# 原生SQL篩選條件
r4 = db_session.query(User).filter_by(name='DragonFire').all()
r5 = db_session.query(User).filter_by(name='DragonFire').first()

# 字符串匹配方式篩選條件 並使用 order_by進行排序
r6 = db_session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='DragonFire').order_by(User.id).all()

#原生SQL查詢
r7 = db_session.query(User).from_statement(text("SELECT * FROM User where name=:name")).params(name='DragonFire').all()

# 篩選查詢列
# query的時候我們不在使用User ORM對象,而是使用User.name來對內容進行選取
user_list = db_session.query(User.name).all()
print(user_list)
for row in user_list:
    print(row.name)

# 別名映射  name as nick
user_list = db_session.query(User.name.label("nick")).all()
print(user_list)
for row in user_list:
    print(row.nick) # 這里要寫別名了

# 篩選條件格式
user_list = db_session.query(User).filter(User.name == "DragonFire").all()
user_list = db_session.query(User).filter(User.name == "DragonFire").first()
user_list = db_session.query(User).filter_by(name="DragonFire").first()
for row in user_list:
    print(row.nick)

# 復雜查詢
from sqlalchemy.sql import text
user_list = db_session.query(User).filter(text("id<:value and name=:name")).params(value=3,name="DragonFire")

# 查詢語句
from sqlalchemy.sql import text
user_list = db_session.query(User).filter(text("select * from User id<:value and name=:name")).params(value=3,name="DragonFire")

# 排序 :
user_list = db_session.query(User).order_by(User.id).all()
user_list = db_session.query(User).order_by(User.id.desc()).all()
for row in user_list:
    print(row.name,row.id)

#其他查詢條件
"""
ret = session.query(User).filter_by(name='DragonFire').all()
ret = session.query(User).filter(User.id > 1, User.name == 'DragonFire').all()
ret = session.query(User).filter(User.id.between(1, 3), User.name == 'DragonFire').all() # between 大於1小於3的
ret = session.query(User).filter(User.id.in_([1,3,4])).all() # in_([1,3,4]) 只查詢id等於1,3,4的
ret = session.query(User).filter(~User.id.in_([1,3,4])).all() # ~xxxx.in_([1,3,4]) 查詢不等於1,3,4的
ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='DragonFire'))).all() 子查詢
from sqlalchemy import and_, or_
ret = session.query(User).filter(and_(User.id > 3, User.name == 'DragonFire')).all()
ret = session.query(User).filter(or_(User.id < 2, User.name == 'DragonFire')).all()
ret = session.query(User).filter(
    or_(
        User.id < 2,
        and_(User.name == 'eric', User.id > 3),
        User.extra != ""
    )).all()
# select * from User where id<2 or (name="eric" and id>3) or extra != "" 

# 通配符
ret = db_session.query(User).filter(User.name.like('e%')).all()
ret = db_session.query(User).filter(~User.name.like('e%')).all()

# 限制
ret = db_session.query(User)[1:2]

# 排序
ret = db_session.query(User).order_by(User.name.desc()).all()
ret = db_session.query(User).order_by(User.name.desc(), User.id.asc()).all()

# 分組
from sqlalchemy.sql import func

ret = db_session.query(User).group_by(User.extra).all()
ret = db_session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).all()

ret = db_session.query(
    func.max(User.id),
    func.sum(User.id),
    func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()
"""

# 關閉連接
db_session.close()

2.6.高級修改數據操作

#高級版更新操作
from my_create_table import User,engine
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
db_session = Session()

#直接修改
db_session.query(User).filter(User.id > 0).update({"name" : "099"})

#在原有值基礎上添加 - 1
db_session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)

#在原有值基礎上添加 - 2
db_session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate")
db_session.commit()

 

 

 

3.一對多的操作 : ForeignKey

3.1.創建數據表及關系relationship:

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# 這次我們要多導入一個 ForeignKey 字段了,外鍵關聯對了
from sqlalchemy import Column,Integer,String,ForeignKey
# 還要從orm 中導入一個 relationship 關系映射
from sqlalchemy.orm import relationship

class ClassTable(Base):
    __tablename__="classtable"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),index=True)

class Student(Base):
    __tablename__="student"
    id=Column(Integer,primary_key=True)
    name = Column(String(32),index=True)

    # 關聯字段,讓class_id 與 class 的 id 進行關聯,主外鍵關系(這里的ForeignKey一定要是表名.id不是對象名)
    class_id = Column(Integer,ForeignKey("classtable.id"))

    # 將student 與 classtable 創建關系 這個不是字段,只是關系,backref是反向關聯的關鍵字
    to_class = relationship("ClassTable",backref = "stu2class")

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8")

Base.metadata.create_all(engine)

3.2.基於relationship增加數據

from my_ForeignKey import Student, ClassTable,engine
# 創建連接
from sqlalchemy.orm import sessionmaker
# 創建數據表操作對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 增加數據
# 1.簡單增加數據
# 添加兩個班級:
# db_session.add_all([
#     ClassTable(name="OldBoyS1"),
#     ClassTable(name="OldBoyS2")
# ])
# db_session.commit()
# 添加一個學生 DragonFire 班級是 OldBoyS1
# 查詢要添加到的班級
# class_obj = db_session.query(ClassTable).filter(ClassTable.name == "OldBoyS1").first()
# 創建學生
# stu = Student(name="DragonFire",class_id = class_obj.id)
# db_session.add(stu)
# db_session.commit()

# 2. relationship版 添加數據
# 通過關系列 to_class 可以做到兩件事
# 第一件事 在ClassTable表中添加一條數據
# 第二件事 在Student表中添加一條數據並將剛剛添加的ClassTable的數據id填寫在Student的class_id中
# stu_cla = Student(name="DragonFire",to_class=ClassTable(name="OldBoyS1"))
# print(stu_cla.name,stu_cla.class_id)
# db_session.add(stu_cla)
# db_session.commit()

# 3.relationship版 反向添加數據
# 首先建立ClassTable數據
class_obj = ClassTable(name="OldBoyS2")
# 通過class_obj中的反向關聯字段backref - stu2class
# 向 Student 數據表中添加 2條數據 並將 2條數據的class_id 寫成 class_obj的id
# class_obj.stu2class = [Student(name="BMW"),Student(name="Audi")]
# db_session.add(class_obj)
# db_session.commit()

# 關閉連接
db_session.close()

3.3.基於relationship查詢數據

from my_ForeignKey import Student, ClassTable,engine

from sqlalchemy.orm import sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.查詢所有數據,並顯示班級名稱,連表查詢
student_list = db_session.query(Student).all()
for row in student_list:
    # row.to_class.name 通過Student對象中的關系字段relationship to_class 獲取關聯 ClassTable中的name
    print(row.name,row.to_class.name,row.class_id)

# 2.反向查詢
class_list = db_session.query(ClassTable).all()
for row in class_list:
    for row2 in row.stu2class:
        print(row.name,row2.name)
# row.stu2class 通過 backref 中的 stu2class 反向關聯到 Student 表中根據ID獲取name


db_session.close()

3.4.更新數據

from my_ForeignKey import Student, ClassTable,engine

from sqlalchemy.orm import sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 更新
class_info = db_session.query(ClassTable).filter(ClassTable.name=="OldBoyS1").first()
db_session.query(Student).filter(Student.class_id == class_info.id).update({"name":"NBDragon"})
db_session.commit()

db_session.close()

3.5.刪除數據

from my_ForeignKey import Student, ClassTable,engine

from sqlalchemy.orm import sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 刪除
class_info = db_session.query(ClassTable).filter(ClassTable.name=="OldBoyS1").first()
db_session.query(Student).filter(Student.class_id == class_info.id).delete()
db_session.commit()

db_session.close()

 

 

4.多對多 : ManyToMany

4.1.創建表及關系

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship

class Hotel(Base):
    __tablename__="hotel"
    id=Column(Integer,primary_key=True)
    girl_id = Column(Integer,ForeignKey("girl.id"))
    boy_id = Column(Integer,ForeignKey("boy.id"))

class Girl(Base):
    __tablename__="girl"
    id=Column(Integer,primary_key=True)
    name = Column(String(32),index=True)

    #創建關系
    boys = relationship("Boy",secondary="hotel",backref="girl2boy")


class Boy(Base):
    __tablename__="boy"
    id=Column(Integer,primary_key=True)
    name = Column(String(32),index=True)


from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:DragonFire@127.0.0.1:3306/dragon?charset=utf8")

Base.metadata.create_all(engine)

4.2.基於relationship增加數據

from my_M2M import Girl,Boy,Hotel,engine

# 創建連接
from sqlalchemy.orm import sessionmaker
# 創建數據表操作對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.通過Boy添加Girl和Hotel數據
boy = Boy(name="DragonFire")
boy.girl2boy = [Girl(name="趙麗穎"),Girl(name="Angelababy")]
db_session.add(boy)
db_session.commit()

# 2.通過Girl添加Boy和Hotel數據
girl = Girl(name="珊珊")
girl.boys = [Boy(name="Dragon")]
db_session.add(girl)
db_session.commit()

4.3.基於relationship查詢數據

from my_M2M import Girl,Boy,Hotel,engine

# 創建連接
from sqlalchemy.orm import sessionmaker
# 創建數據表操作對象 sessionmaker
DB_session = sessionmaker(engine)
db_session = DB_session()

# 1.通過Boy查詢約會過的所有Girl
hotel = db_session.query(Boy).all()
for row in hotel:
    for row2 in row.girl2boy:
        print(row.name,row2.name)

# 2.通過Girl查詢約會過的所有Boy
hotel = db_session.query(Girl).all()
for row in hotel:
    for row2 in row.boys:
        print(row.name,row2.name)