一、介绍
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() # 关闭sessionModel.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()
![拓展欧几里得 [Noi2002]Savage](https://img-blog.csdn.net/20170815211246353?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvUVRZMjAwMQ==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/SouthEast)
