一、介紹

SQLAlchemy是一個基于Python實現的ORM框架。該框架建立在 DB API之上，使用關系對象映射進行數據庫操作，簡言之便是：將類和對象轉換成SQL，然后使用數據API執行SQL并獲取執行結果。

安裝：pip3 install sqlalchemy

組成部分：

• Engine，框架的引擎
• Connection Pooling ，數據庫連接池
• Dialect，選擇連接數據庫的DB API種類
• Schema/Types，架構和類型
• SQL Exprression Language，SQL表達式語言

SQLAlchemy本身無法操作數據庫，其必須以來pymsql等第三方插件，Dialect用于和數據API進行交流，根據配置文件的不同調用不同的數據庫API，從而實現對數據庫的操作，如：

#################### MySQL-Python ####################
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>#################### pymysql ####################
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]#################### MySQL-Connector ####################
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>#################### cx_Oracle ####################
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html

二、?使用

2.1??創建數據庫表

2.1.1 數據類型和數據庫的對應關系

類型名	MySQL類型	python類型	描述
SmallInteger	smallint	int	取值范圍較小，一般為16位
Integer	int	int	普通整數，一般32位
BigInteger	bigint	int/long	不限精度的整數
Float	float	float	浮點數
Numeric	decimal	decimal.Decimal	定點數
String	varchar	str	變長字符串
Text	tinytext	str	變長字符串，64K,216?1216?1 2^{16}-1216?1=65535bytes
Text(65536)	mediumtext	str	變長字符串，max16M,224?1224?1 2^{24}-1224?1=16777215bytes
Text(16777216)	longtext	str	變長字符串，max32G,232?1232?1 2^{32}-1232?1=4294967295bytes
LargeBinary	blob	str	二進制文件，64K
LargeBinary(65536)	mediumblob	str	二進制，max16M
LargeBinary(16777216)	longblob	str	二進制，max32G
PickleType	blob	任何python對象	自動使用Pickle序列化，只有blob
Unicode	varchar	unicode	變長字符串
UnicodeText	text	unicode	變長字符串，64K
Boolean	tinyint	bool	布爾值
Date	date	datetime.date	日期
Time	time	date.time	時間
DateTime	datetime	datetime.datetime	日期和時間
Interval	datetime	datetime.timedelta	時間間隔
Enum	enum	str	一組字符串

屬性	描述	生效值
primary_key	主鍵	True
unique	鍵值唯一性	True
index	索引	True
nullable	空值	True
default	默認值	null

2.1.2?示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
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
from sqlalchemy.orm import relationshipBase = declarative_base()###################### 單表示例 #########################
class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)email = Column(String(32), unique=True)ctime = Column(DateTime, default=datetime.datetime.now)extra = Column(Text, nullable=True)__table_args__ = (UniqueConstraint('id', 'name', name='uix_id_name'),  # 聯合唯一約束Index('ix_id_name', 'name', 'extra'),                # 普通索引)class Hosts(Base):__tablename__ = 'hosts'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)ctime = Column(DateTime, default=datetime.datetime.now)###################### 一對多示例 #########################
class Hobby(Base):__tablename__ = 'hobby'id = Column(Integer, primary_key=True)caption = Column(String(50), default='籃球')class Person(Base):__tablename__ = 'person'nid = Column(Integer, primary_key=True)name = Column(String(32), index=True, nullable=True)hobby_id = Column(Integer, ForeignKey("hobby.id"))# 與生成表結構無關，僅用于查詢方便hobby = relationship("Hobby", backref='pers')   # 第一個字段表示關聯的表，backref 表示反向查詢的字段名稱###################### 多對多示例 #########################
class Server2Group(Base):__tablename__ = 'server2group'id = Column(Integer, primary_key=True, autoincrement=True)server_id = Column(Integer, ForeignKey('server.id'))group_id = Column(Integer, ForeignKey('group.id'))class Group(Base):__tablename__ = 'group'id = Column(Integer, primary_key=True)name = Column(String(64), unique=True, nullable=False)# 與生成表結構無關，僅用于查詢方便servers = relationship('Server', secondary='server2group', backref='groups')  # secondary表示第三張表class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根據類創建數據庫表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0,   # 超過連接池大小外最多創建的連接pool_size=5,      # 連接池大小pool_timeout=30,  # 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1   # 多久之后對線程池中的線程進行一次連接的回收（重置）)Base.metadata.create_all(engine)def drop_db():"""根據類刪除數據庫表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0,   # 超過連接池大小外最多創建的連接pool_size=5,      # 連接池大小pool_timeout=30,  # 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1   # 多久之后對線程池中的線程進行一次連接的回收（重置）)Base.metadata.drop_all(engine)if __name__ == '__main__':drop_db()init_db()

指定關聯列：hobby = relationship("Hobby", backref='pers',foreign_keys="Person.hobby_id")

2.2? 操作數據庫表

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from?sqlalchemy.orm?import?sessionmaker
from?sqlalchemy?import?create_engine
from?models?import?Usersengine?=?create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session?=?sessionmaker(bind=engine)
session?=?Session()  # 每次執行數據庫操作時，都需要創建一個session# ############# 執行ORM操作 #############
obj1?=?Users(name="alex1")
session.add(obj1)session.commit()  # 提交事務
session.close()   # 關閉session

?????基于scoped_session實現線程安全

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
"""
# 線程安全，基于本地線程實現每個線程用同一個session
# 特殊的：scoped_session中有原來方法的Session中的以下方法：public_methods = ('__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested','close', 'commit', 'connection', 'delete', 'execute', 'expire','expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind','is_modified', 'bulk_save_objects', 'bulk_insert_mappings','bulk_update_mappings','merge', 'query', 'refresh', 'rollback','scalar'
)
"""# ############# 執行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)session.commit()  # 提交事務
session.close()   # 關閉session

多線程執行示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from db import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)def task(arg):# 每個線程創建一個sessionsession = Session()obj1 = Users(name="alex1")session.add(obj1)session.commit()for i in range(10):t = threading.Thread(target=task, args=(i,))t.start()

2.3? 基本增刪改查示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import textfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()# ################ 添加 ################
# 單個插入
obj1 = Users(name="fenglepeng")
session.add(obj1)
session.flush()
print(obj1.id)    # 輸出新插入數據的主鍵，當只有一個主鍵時才能生效，如果設置兩個及以上會取不到插入的數據# 批量插入
session.add_all([Users(name="wupeiqi"),Users(name="alex"),Hosts(name="c1.com"),
])
session.commit()   # 此時數據才插入到數據庫中# ################ 刪除 ################
session.query(Users).filter(Users.id > 2).delete()
session.commit()# ################ 修改 ################
# 使用 update
session.query(Users).filter(Users.id > 0).update({"name" : "099"})
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)       # 字符串
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")  # 數字# 或者直接修改查詢到的對象的屬性
obj1 = session.query(Users).filter(Users.id > 0).first()
obj1.name = 'fenglp'
session.commit()# ################ 查詢 ################
r1 = session.query(Users).all()                               # 不加all，顯示的是執行的sql
r2 = session.query(Users.name.label('xx'), Users.age).all()   # label 別名  
r3 = session.query(Users).filter(Users.name == "alex").all()  # query表示查詢的字段，filter表示過濾的字段
r4 = session.query(Users).filter_by(name='alex').all()        # 本表的字段
r5 = session.query(Users).filter_by(name='alex').first()# params 和Python中的format類似，用于替代前面的值
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all() # 嵌套查詢，子查詢session.close()

session 相關函數

db.session.add(obj)    # 添加
db.session.flush(obj)  # flush之后，可以取到obj的主鍵，但是當只有一個主鍵時才能生效，如果設置兩個及以上會取不到插
db.session.rollback()  # 回滾
db.session.commit() ?  # 提交事務，commit執行了，但是此時sql層面的commit并沒有執行
db.session.close()? ?  # 關閉session

Model.query是db.session.query(Model)的快捷方式，它不可調用。?如果您沒有查詢模型，請繼續使用db.session.query(...)就像使用常規SQLAlchemy一樣。

# 下面兩種方式結果相同，第二種是第一種的簡化版
db.session.query(Users).all() # 這種query中可以加參數
Users.query.all()             # 這種query中不能加參數

filter_by和filter的區別

• filter_by表內部精確查詢

User.query.filter_by(id=4).first()

• filter 全局查詢 id必須指明來源于那張表User，而且需要用等號，而不是賦值，建議使用filter

User.query.filter(User.id==4).first()

查詢操作

################# 條件查詢 ################
ret = session.query(Users).filter(Users.id = 1).all()   # 等于
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()  # between
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()   # in
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()  # 取反
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()  # 子查詢################# 組合查詢 ################
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()  # 組合查詢，默認是andfrom sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # _and里面的關系是and
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2,and_(Users.name == 'eric', Users.id > 3),Users.extra != "")).all()#################  通配符 % 表是多個，_ 表示一個 ################
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()################# 限制 ################
ret = session.query(Users).filter(Users.name.like('e%')).limit(10).all()  # offset
ret = session.query(Users).filter(Users.name.like('e%')).limit(10).offset(100).all()  # offset 和 limit
## 切片
ret = session.query(Users).filter(Users.name.like('e%')).slice(90,100).all()
ret = session.query(Users).filter(Users.name.like('e%'))[90:100]################# 排序 ################
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()################# 去重 ################
ret = session.query(Users).order_by(Users.name.desc()).distinct().all()################# 分組 ################
from sqlalchemy.sql import func################# 分組聚合 ################
ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).all()
ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()################# 連表  ################
# 已經設置好外鍵,它自動找到關聯的字段
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
ret = session.query(Person).join(Favor).all()                   # 默認是 inner join
ret = session.query(Person).join(Favor, isouter=True).all()     # isouter=True 表示左鏈接# 自己指定關聯關系
ret = session.query(Person).join(Favor, Person.id == Favor.nid).all()               # 默認是 inner join
ret = session.query(Person).join(Favor, Person.id == Favor.nid, isouter=True).all() # isouter=True 表示左鏈接,或者也可以
ret = session.query(Person).outerjoin(Favor, Person.id == Favor.nid).all()          # 左連接# 返回值是一個列表，包含你查詢的兩個對象，如[<app.models.Favor object at 0x7f0a12692290>, <app.models.Person object at 0x7f0a12692290>]################# 組合 ################
# union 對兩個結果集進行并集操作，不包括重復行，同時進行默認規則的排序
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()# union_all 對兩個結果集進行并集操作，包括重復行，不進行排序
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

基于relationship操作ForeignKey

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Personengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
########################## 添加 ############################
session.add_all([Hobby(caption='乒乓球'),Hobby(caption='羽毛球'),Person(name='張三', hobby_id=3),Person(name='李四', hobby_id=4),
])person = Person(name='張九', hobby=Hobby(caption='姑娘'))  # 同時創建，并自動添加關聯
session.add(person)hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飛'), Person(name='博雅')]
session.add(hb)
session.commit()########################## 查詢 ############################
# 使用relationship正向查詢
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)# 使用relationship反向查詢
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)session.close()

基于relationship操作m2m

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
########################## 添加 ############################
session.add_all([Server(hostname='c1.com'),Server(hostname='c2.com'),Group(name='A組'),Group(name='B組'),
])
session.commit()s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()gp = Group(name='C組')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()ser = Server(hostname='c6.com')
ser.groups = [Group(name='F組'),Group(name='G組')]
session.add(ser)
session.commit()########################## 查詢 ############################
# 使用relationship正向查詢
v = session.query(Group).first()
print(v.name)
print(v.servers)# 使用relationship反向查詢
v = session.query(Server).first()
print(v.hostname)
print(v.groups)session.close()

執行原生SQL語句

################# 第一種 ################
import?time
import?threading
import?sqlalchemy
from?sqlalchemy?import?create_engine
from?sqlalchemy.engine.base?import?Engineengine?=?create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",max_overflow=0,??# 超過連接池大小外最多創建的連接pool_size=5,??   # 連接池大小pool_timeout=30,?# 池中沒有線程最多等待的時間，否則報錯pool_recycle=-1??# 多久之后對線程池中的線程進行一次連接的回收（重置）
)def?task(arg):conn?=?engine.raw_connection()cursor?=?conn.cursor()cursor.execute("select * from t1")result?=?cursor.fetchall()cursor.close()conn.close()for?i?in?range(20):t?=?threading.Thread(target=task, args=(i,))t.start()################# 第二種 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):conn = engine.contextual_connect()with conn:cur = conn.execute("select * from t1")result = cur.fetchall()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第三種 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
from sqlalchemy.engine.result import ResultProxy
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):cur = engine.execute("select * from t1")result = cur.fetchall()cur.close()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第四種 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()# 關聯子查詢
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
result = session.query(Group.name, subqry)
"""
SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid 
FROM server 
WHERE server.id = `group`.id) AS anon_1 
FROM `group`
"""# 查詢
# cursor = session.execute('select * from users')
# result = cursor.fetchall()# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
session.commit()
print(cursor.lastrowid)session.close()

注意: 查看連接?show status like 'Threads%';

從 datetime 字段中獲取當日數據

from sqlalchemy import funcclass History(db.Model):__tablename__ = 'historys'id = db.Column(db.Integer, primary_key=True)date = db.Column(db.Date)datetime = db.Column(db.Datetime)# firstDay：某年某月的第一天，datetime類型
# lastDay：某年某月的最后一天，datetime類型
historys = History.query.filter(History.date.between(firstDay, lastDay)).all()
historys = History.query.filter(func.date(datetime).between(firstDay, lastDay)).all()# 格式化時間
func.date_format(History.datetime , "%Y-%m-%d %H:%i:%s")
# 將時間戳轉化為時間
func.from_unixtime((Table.timestamp), "%Y-%m-%d %H:%i:%s")
# 獲取12月份的所有數據
historys = History.query.filter(extract('month', History.date) == 12, extract('month', History.date) == 12).all()

動態增加條件

filter_by

filter_by用于查詢簡單的列名，不支持比較運算符。

filters = {’name': ‘fengyao', ‘age': 26}
User.query.filter_by(**filters).first()

filter

比filter_by的功能更強大，支持比較運算符，支持or_、in_等語法。

filters = {User.name == ‘fengyao’,User.age > 25
}
User.query.filter(*filters).first()