SQLAlchemy 是一個 ORM框架：類對應表，類中的字段對應表中的列，類的對象對應表的一條記錄；作用：幫助我們使用類和對象快速實現數據庫操作
操作數據庫的方式：
1. 原生SQL - pymysql ：支持 python2 和 python3 - MySQLdb ：只支持 python2 2. ORM框架 ：框架本身要是有ORM應用自己的（如 Django），要是沒有就用 SQLAlchemy
安裝 SQLAlchemy：
pip install sqlalchemy
注： SQLAlchemy 默認不能修改表結構；想修改表結構需要引入第三方組件
SQLAlchemy 單表
創建數據庫的表 --- 單表：models.py
import datetime from sqlalchemy import create_engine # 引入 創建引擎 from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index # 引入列和數據類型 Base = declarative_base() # Base 要自己實例化 class Users(Base): # Users是類名；類 繼承 Base __tablename__ = 'users' # 數據庫的表名叫 "users" id = Column(Integer, primary_key=True) # Colume表示列； id 是表中的列，Interger 表示整形，primary_key=True 表示 是主鍵 name = Column(String(32), index=True, nullable=False) # name 為表中的列， String(32) 表示 MySQL中的 Varchar(32)， index=True 表示索引，nullable=False 表示不允許為空 # email = Column(String(32), unique=True) # ctime = Column(DateTime, default=datetime.datetime.now) # extra = Column(Text, nullable=True) def init_db(): """ 根據類創建數據庫的表 :return: """ engine = create_engine( # 創建數據庫連接 "mysql+pymysql://root:tj037778@127.0.0.1:3306/dbtest?charset=utf8", # mysql 表示要連接的數據庫；pymysql 表示用 pymysql 來連接；用戶名是 root，密碼是123，連接本地的 dbtest 這個數據庫 max_overflow=0, # 超過連接池大小外最多創建的連接；即超過 pool_size 后最多能溢出多少個連接 pool_size=5, # 連接池大小 pool_timeout=10, # 池中沒有線程（連接）最多等待的時間，否則報錯 pool_recycle=-1 # 多久之后對線程池中的線程進行一次連接的回收（重置）；-1表示不重建 ) Base.metadata.create_all(engine) # Base.metadata.create_all() ： 找到當前 py 文件下面 繼承了Base的所有的類，在數據庫中生成一張表 def drop_db(): """ 根據類刪除數據庫中的表 :return: """ engine = create_engine( "mysql+pymysql://root:tj037778@127.0.0.1:3306/dbtest?charset=utf8", max_overflow=0, # 超過連接池大小外最多創建的連接 pool_size=5, # 連接池大小 pool_timeout=30, # 池中沒有線程最多等待的時間，否則報錯 pool_recycle=-1 # 多久之后對線程池中的線程進行一次連接的回收（重置） ) Base.metadata.drop_all(engine) # Base.metadata.drop_all() ： 找到當前 py 文件下面 繼承了Base的所有的類，在數據庫中刪除相應的表 if __name__ == "__main__": init_db() # drop_db()
單表基本增刪改查示例：crud.py
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import Users # 導入 Users 類 # 創建數據庫連接（crud 這個py文件的數據庫連接） engine = create_engine("mysql+pymysql://root:tj037778@127.0.0.1:3306/dbtest", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) # Session可理解成 連接池 # 通過 SQLAlchemy 操作數據庫：通過 Users 類對 users 表進行增刪改查 # 每次執行數據庫操作時，都需要創建一個session （可理解成一個連接） session = Session() # ############# 執行ORM操作：增 ############# # 1. 單條增加 """ obj1 = Users(name="neo") # 實例化一個 Users 的對象；該對象是在內存中創建 session.add(obj1) # session 根據 obj1 在表中增加一條數據 session.commit() # session.add() 之后要 commit() 一下 session.close() # 關閉 連接 """ # 2. 多條增加 """ obj2 = Users(name="egon") obj3 = Users(name="alex") session.add_all( # 增加多個；也可以 session.add() 多次 [obj2,obj3] # 添加的記錄（對象）放入 列表 中 ) session.commit() session.close() """ # ############# 執行ORM操作：查 ############# # 1. 查詢所有 res = session.query(Users).all() # 查詢 Users 表中的所有數據 print(res) # 打印結果（按id 倒序排列）： [<models.Users object at 0x000002132AEFE400>, <models.Users object at 0x000002132AEFE470>, <models.Users object at 0x000002132AEFE4E0>] # 列表套對象的形式 for row in res: print(row.id, row.name) # 2. 條件查詢 res_filter = session.query(Users).filter(Users.id >= 2) # filter() 中直接加 表達式 for row in res_filter: print(row.id, row.name) # 3. 只取第一個 res_first = session.query(Users).filter(Users.id >= 2).first() # first() 取到的結果不再是一個列表，而是一個對象 print(res_first.id, res_first.name) # ############# 執行ORM操作：刪 ############# session.query(Users).filter(Users.id >= 3).delete() # 查詢結果 .delete() session.commit() # 刪除之后也要 commit() # ############# 執行ORM操作：改 ############# session.query(Users).filter(Users.id == 1).update({Users.name: "NEO"}) # 查詢結果.update({ }) ； 返回 更新的條數 # session.query(Users).filter(Users.id==1).update({"name":"neo"}) # 也可以用這種寫法 session.commit() # update() 補充： # session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False) # 在字段原有值的基礎上更新時，synchronize_session=False 表示是 字符串 的拼接 # session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") # 在字段原有值的基礎上更新時，synchronize_session="evaluate" 表示是 數值 的相加減 （默認是這種） # session.commit() session.close()
SQLAlchemy 單表的其它常用操作：
# ##################### 其它常用操作 ######################## # 1. 查詢某些字段 res = session.query(Users.id,Users.name).all() # 查詢 id字段 和 name字段；結果為列表 for item in res: print(item,type(item),item[0],item.name) # item 不是元組,只是形式像元組，並且可以利用 item[0]，item[1] 這種元組的用法去獲取item中的值，也可以利用 item.id 和 item.name 獲取item中的值 # 打印結果： # (2, 'egon') <class 'sqlalchemy.util._collections.result'> 2 egon # (1, 'NEO') <class 'sqlalchemy.util._collections.result'> 1 NEOsult'> # 2. 給某字段起個別名（sql 中的 as）： .label("別名") res = session.query(Users.id,Users.name.label("cname")).all() for item in res: print(item.cname) # 打印結果： # egon # NEO # 3. 查看sql語句：去掉 .all() query = session.query(Users.id,Users.name.label("cname")) print(query) # 打印結果： # SELECT users.id AS users_id, users.name AS cname FROM users # 4. 其它 ret1 = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() # filter() 中的 “,” 表示的是 sql 條件中的 and （默認條件為 and ） ret2 = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all() # .between ： sql 條件中的 between...and ret3 = session.query(Users).filter(Users.id.in_([1,3,4])).all() # Users.id.in_([1,3,4]) ：sql 條件中的 in ret4 = session.query(Users).filter(~Users.id.in_([1,3,4])).all() # ~Users.id.in_([1,3,4]) ：not in
# 5. 子查詢 ret5 = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all() # 一個查詢結果作為另一個的查詢條件 # 6. and 和 or 的嵌套 from sqlalchemy import and_,or_ # 先導入 and 和 or session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # and_() 中的條件是 and 關系 （默認也是 and） session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() # or_() 中的條件是 or 關系 session.query(Users).filter( or_( # 下面的三個條件是 or 的關系 Users.id < 2, and_(Users.name == 'eric', Users.id > 3), # 這兩個是 and 的關系 # Users.extra != "" # 這個是 not （非） )).all() ret6 = session.query(Users).filter_by(name='alex').all() # filter_by() 過濾 """ filter() 和 filter_by() 的區別： filter：column == expression 1. 傳入參數的寫法，要用：類名.列名 兩個等號 去判斷 2. 更復雜的查詢的語法，比如and_()，or_()等多個條件的查詢，只支持filter filter_by：keyword = expression 1. 傳入參數的寫法，只需要用：（不帶類名的）列名 單個等號 就可以判斷 """ # 7. 通配符 ret7 = session.query(Users).filter(Users.name.like('n%')).all() # like ret8 = session.query(Users).filter(~Users.name.like('n%')).all() # not like ret9 = session.query(Users).filter(Users.name.like('n_')).all() # like # 8. 切片 （限制/分頁） ret10 = session.query(Users)[1:2] # 9. 排序 ret11 = session.query(Users).order_by(Users.name.desc()).all() ret12 = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() # 10. 分組：group_by() from sqlalchemy.sql import func # 導入 func，func 中有聚合函數 # ret13 = session.query(Users).group_by(Users.depart_id).all() # group_by(字段)之后不要 query() 所有字段 ret14 = session.query( Users.name, func.max(Users.id), # 自己指定取哪個字段 func.sum(Users.id), # 非分組字段獲取時，使用聚合函數 func.min(Users.id) ).group_by(Users.name).all() ret15 = session.query( func.max(Users.id), func.sum(Users.id), func.min(Users.id) ).group_by(Users.name).having(func.count(Users.id) >2).all() # 根據 name 分組，func.count(Users.id) > 2 ；根據聚合函數進行二次篩選：having # 11. 組合（垂直/上下連表）：union 和 union all --- union all 去重，union 不去重 q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret16 = q1.union(q2).all() q1 = session.query(Users.name).filter(Users.id > 2) q2 = session.query(Favor.caption).filter(Favor.nid < 2) ret17 = q1.union_all(q2).all()