SQlAlchemy的增刪改查

一.創建數據表

# 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)

二.增加數據

 

#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()

 

三.查詢數據

 

# 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() #關閉會話

 

四.修改數據

 

# 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() #關閉會話

 

五.刪除數據

 

# 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() #關閉會話

 

六,高級版查詢

 

# 高級版查詢操作
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()

 

七.高級版修改數據

#高級版更新操作
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()