一 简介和安装

1 简介

1 简介

SQLALchemy 是一个第三方的ORM 框架，大量使用元编程实现其基本功能

创新互联主要从事成都网站制作、做网站、网页设计、企业做网站、公司建网站等业务。立足成都服务错那,10余年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:13518219792

---

相关参考文档:https://docs.sqlalchemy.org/en/13/

2 基本组成

一个连接池引擎中存放着连接池和方言(dialect)，dialect不同的数据库引擎对应不同的方言。

---

DBAPI通过API调用数据库的各种功能，如控制语句等。

2 安装

1 安装

pip install sqlalchemy

2 版本检测

pip show  sqlalchemy

二 基本实践

1 创建连接

数据库链接的事情，交给引擎处理

#!/usr/bin/poython3.6
#conding:utf-8
import pyMySQL
import sqlalchemy
from sqlalchemy import create_engine

#创建存储引擎
username='root'
password='Admin@Root123'
ip='192.168.1.200'
port=3306
databases='test'

# echo =True 打印SQL执行过程，主要应用于调试模式

engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)

2 创建实体类

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String

#创建存储引擎
username='root'
password='Admin@Root123'
ip='192.168.1.200'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

3 实例化

#实例化
L=Login(name='admin')
L.age=30

4 创建表

可以使用SQLAIchemy 来创建，删除表

删除继承自Base的所有表
Base.metadata.drop_all(engine)
创建继承自Base的所有表
Base.metadata.create_all(engine)

生产环境很少这样创建表，都是系统上线的时候由脚本生成的，生产环境很少删除表，废弃都不能删除。

---

具体代码如下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine) # 此处通过引擎提供的连接数据库的能力来创建默认字符集和相关检测方式，其被存储在Base元类中。

执行结果如下

数据库结果如下

5 创建会话

在一个会话中操作数据库，会话建立在连接上，连接被引擎管理

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作，此处不会commit
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

6 CRUD 操作(增，删，改，查)

1 增

1 增加单行数据

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:
    ed_login = Login(name='admin', loginname='admin', password='admin')
    # 另一种插入方式 
    # ed_login=Login()
    # ed_login.name='root'
    # ed_login.password='root'
    # ed_login.loginname='root'
    session.add(ed_login)  #增加状态
    session.commit()
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

结果如下

数据库结果如下

2 同时插入多行

add_all() 相关源码

    def add_all(self, instances):
        """Add the given collection of instances to this ``Session``."""

        if self._warn_on_events:
            self._flush_warning("Session.add_all()")

        for instance in instances:  # 此处也是通过add的方式进行遍历加入的
            self.add(instance, _warn=False)

代码如下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:
    lst=[]
    for  i in range(5):
        lst.append(Login(name='admin'+str(i), loginname='admin'+str(i), password='admin'+str(i)))
    print (lst)
    session.add_all(lst)
    session.commit()
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

结果如下

数据库结果如下

3 多次提交commit是否提交多次问题

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:
    lst=[]
    for  i in range(5):
        lst.append(Login(name='admin'+str(i), loginname='admin'+str(i), password='admin'+str(i)))
    print (lst)
    session.add_all(lst)
    print  ('-----commit------')
    session.commit()
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

结果如下

结论：如上，多次提交时commit只执行一次。

4 修改更新数据并提交

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases), echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:
    ed_login = Login(name='admin', loginname='admin', password='admin')
    session.add(ed_login)
    session.commit()

    # 修改相关属性
    ed_login.loginname='1322534564356464564'
    session.add(ed_login)
    print  ('-----commit------')
    session.commit()
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

执行代码结果显示如下

数据库结果如下

结论:

常规来说，上述代码应该是提交了两次，其并未进行相关的限制，但其显示结果却是修改了，而不是两次提交，此处便涉及到了状态变化，应该是进行了某种判断，此处add后状态未发生变化，因此只添加了一条

---

insert 只能返回你添加了几行，下面的情况会修改，其修改会使用update进行处理，更新的时候，一般需要给ID(主键)进行处理，如果没有，则insert后拿取对应的id进行相关的处理，及提交完成后若出现更改，则通过一定方式获取上次提交的主键，并进行相关的操作处理，当一次commit提交完成后，实例的无主键状态会变为有主键状态。

2 简单查询语句

使用query()方法，返回一个Query对象

---

查询相关源码

    def query(self, *entities, **kwargs):
        """Return a new :class:`.Query` object corresponding to this
        :class:`.Session`."""

        return self._query_cls(entities, self, **kwargs)

上述可知，其查询参数是可变参数和关键字参数，及支持多级查询

---

相关代码如下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:
    login=session.query(Login)  # 此处是实体类,此处是未进行限制，默认查询所以
    for  i  in  login:
        print (i)
    print ('----------------')
    data=session.query(Login).get(10)  # 通过主键id进行相关的查询
    print (data)
    print ('```````````````````')
    data1=session.query(Login).filter(Login.id==1)  # 此处通过filter进行操作，注意，此处是python命令，是双==号
    print (data1)
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

结果如下

3 修改

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases),echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:

    data=session.query(Login).get(11)  # 通过主键id进行相关的查询
    print  (data)
    data.password='passwd'
    data.loginname='abc'
    session.add(data)
    session.commit()
except  Exception as  e:   # 若抛出异常，则直接回滚
    print (e)
    session.rollback()
finally:
    pass

结果如下

数据库结果如下

具体步骤如下

1 先获取到查询结果
2 进行相关的修改操作
3 最后提交

4 删除

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases),echo=True)
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

#  实例化并进行插入操作
try:

    login=Login(id=10)  # 实例化
    session.delete(login)  # 删除id对应的值
    session.commit()  # 提交 
except  Exception as  e:   # 若抛出异常，则直接回滚
    print ('异常如下：--------',e)
    session.rollback()
finally:
    pass

结果如下

5 状态处理

上述产生一个异常，是未进行持久化的异常，每一个实体，都有状态属性_sa_instance_state，其类型是sqlchemy.orm.state.InstanceState,可以使用sqlalchemy.inspect(entity)函数查看状态

---

常见的状态值有 transient，pending，persistent，deleted，detached。

状态	说明
transient	实体类尚未加入到session中，同时并没有保存到数据库中
pending	transient的实体被add()到session中，状态切换到pending，但是还没有flush到数据库中
persistent	session中的实体对象对应着数据库中的真实记录，pending状态在提交成功后可以变成persistent状态，或者查询成功返回的实体也是persistent状态
deleted	实体被删除且已经flush但未commit完成，事务提交成功了，实体变成detached，事务失败，返回persistent
detached	删除成功的实体进入此状态

---

解释

新建一个实体，状态是transient临时的
一旦add()后从transient状态变为pending状态
成功commit()后从pending变为persistent状体
成功查询返回的实体对象，也是persistent状态
及就是 persistent状态可以由两种形式组成(commit和 get成功)
persistent状态的实体，修改依然是persistent状态
persistent状态的实体，删除后，flus后但没有commit，就变成了deleted状态，成功提交，变成detached状态，失败，还原persistent状态
删除，修改操作，需要对应一个真实的记录，所以要求实体对象必须是persistent状态

---

状态处理实例

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String,inspect
# 导入会话模块
from sqlalchemy.orm import sessionmaker

#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 导入基类
Base=declarative_base()

# 创建实体类
class  Login(Base):
    __tablename__='login'
    id=Column(Integer,primary_key=True)  # 设置id列为整形，并设置其为主键 ，整形时Inteager
    name=Column(String(48))  # 定义name名称，并定义其字段为charset类型和长度为48
    loginname=Column(String(48))
    password=Column(String(256))
    def __repr__(self):
        return ""  %(self.id,self.name,self.loginname,self.password)

#实例化
L=Login(name='admin')
L.age=30

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)

#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()

# 创建一个状态查询函数
def show(entiry):
    ins=inspect(entiry)
    return (ins.transient,ins.pending,ins.persistent,ins.deleted,ins.detached)
#  实例化并进行插入操作
try:
    login=Login(id=14)  #  实例化，此处的iD必须不存在，否则会报主键冲突
    print  ('实例化对应状态:',show(login))
    session.add(login)
    print  ('添加成功对应状态:',show(login))
    session.commit()  # 提交
    print  ('提交成功对应状态:',show(login))
    session.delete(login)  # 删除id对应的值
    print  ('删除对应状态:',show(login))
    session.commit()  # 提交
    print  ('删除提交对应状态:',show(login))

except  Exception as  e:   # 若抛出异常，则直接回滚
    print ('异常如下：--------',e)
    session.rollback()
finally:
    pass

结果如下

7 复杂查询语句

1基本查询

1 简单条件查询语句

    emps=session.query(Employess).filter(Employess.emp_no > 10025)  # 工号大于10025 的显示

2 in 语句

emps=session.query(Employess).filter(Employess.emp_no.in_(emplist))

3 not in 语句

 emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist))

4 消费者方法

消费者方法调用后，Query对象(可迭代)就转换成了一个容器

 # 消费者方法
    emps=session.query(Employess)  # 查询所有
    print (emps.all())  # 返回列表
    print (emps.count())  # 聚合函数查询
    print (len(list(emps))) # 返回列表元素数量，及返回值数量
        emps=session.query(Employee).filter(Employee.emp_no==10025)

5 LIKE

    emps=session.query(Employess).filter(Employess.last_name.like('P%'))

6 上述结果如下

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String,Enum,DATE
# 导入会话模块
from sqlalchemy.orm import sessionmaker
# 导入枚举类型
import enum
#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 导入基类
Base=declarative_base()
# 创建要查询的表对应的实体类
#表字段如下
'''
CREATE TABLE `employees` (
  `emp_no` int(11) NOT NULL COMMENT '主键',
  `birth_date` date NOT NULL COMMENT '生日',
  `first_name` varchar(14) NOT NULL COMMENT '用户-姓',
  `last_name` varchar(16) NOT NULL COMMENT '用户-名',
  `gender` enum('M','F') NOT NULL COMMENT '性别',
  `hire_date` date NOT NULL COMMENT '入职时间',
  PRIMARY KEY (`emp_no`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
'''

# 创建枚举类
class  MyEnum(enum.Enum):
    M='M'
    F='F'

# 创建对应实体类
class  Employess(Base):
    __tablename__='employees'
    emp_no=Column(Integer,nullable=False,primary_key=True)  # 整形，非空，主键
    birth_date=Column(DATE,nullable=False)  # 日期类型，非空
    first_name=Column(String(14),nullable=False)  # 字符串类型，长度为14，非空
    last_name=Column(String(14),nullable=False)  # 字符串类型，长度为16，非空
    gender=Column(Enum(MyEnum),nullable=False)  # 枚举，非空
    hire_date=Column(DATE,nullable=False)  # 字符类型，非空
    def __repr__(self):  # 定义查询返回
        return "< {}  emp_no:{}  name:{}>  ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.emp_no,self.last_name))
    __str__=__repr__

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)
#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()
#  实例化并进行插入操作

def  show(emps):
    for x  in emps:
        print (x)
    print ('------------',end='\n\n')

try:
    # 简单查询语句

    emps=session.query(Employess).filter(Employess.emp_no > 10025)  # 工号大于10025 的显示
    print ("简单查询语句")
    show(emps)
    # in 语句查询
    emplist=[10001,10002,10025]
    emps=session.query(Employess).filter(Employess.emp_no.in_(emplist))
    print ("in 语句查询")
    show(emps)
    # not in 语句查询
    emps=session.query(Employess).filter(~Employess.emp_no.in_(emplist))
    print ("not in 语句查询")
    show(emps)
    # 消费者方法
    emps=session.query(Employess)  # 查询所有
    print ("消费者方法")
    print (emps.all())  # 返回列表
    print (emps.count())  # 聚合函数查询
    print (len(list(emps))) # 返回列表元素数量，及返回值数量
    # like
    emps=session.query(Employess).filter(Employess.last_name.like('P%'))
    print ('Like方法')
    print (show(emps))
except  Exception as  e:   # 若抛出异常，则直接回滚
    print ('异常如下：--------',e)
finally:
    pass

结果如下

2 聚合，分组

1 聚合函数

# 导入聚合函数
from  sqlalchemy  import  func

try:
    # count命令
    print  (session.query(func.count(Employess.emp_no)).all())
    print  (session.query(func.count(Employess.emp_no)).one())  # 若使用one,则只能由一个结果，若有多个结果，则会报错
    print  (session.query(func.count(Employess.emp_no)).scalar())  #取回one()元组返回的第一个元素
    # AVG 平均值
    print  (session.query(func.avg(Employess.emp_no)).all())
    print  (session.query(func.avg(Employess.emp_no)).scalar())

    # min 最小值
    print  (session.query(func.min(Employess.emp_no)).all())
    print  (session.query(func.min(Employess.emp_no)).scalar())
    # max 最大值
    print  (session.query(func.max(Employess.emp_no)).all())
    print  (session.query(func.max(Employess.emp_no)).scalar())
except  Exception as  e:   # 若抛出异常，则直接回滚
    print ('异常如下：--------',e)
finally:
    pass

结果如下

2 分组

    print (session.query(Employess.gender,func.count(Employess.emp_no)).group_by(Employess.gender).all())

结果如下

3 逻辑与或非

# 导入与或非模块
from  sqlalchemy  import  or_,and_,not_

    # AND 第一个和第二个都要成立
    # 方式一and
    emps=session.query(Employess).filter(Employess.emp_no>10015).filter(Employess.emp_no<10018)
    show(emps)
    # 方式二 and
    emps=session.query(Employess).filter(and_(Employess.emp_no>10015,Employess.emp_no<10018))
    show(emps)
    # 方式三 &
    emps=session.query(Employess).filter((Employess.emp_no>10015)&(Employess.emp_no<10018))
    show(emps)

    # OR 或
    # 方式一
    emps=session.query(Employess).filter((Employess.emp_no>10025)|(Employess.emp_no<10005))
    show(emps)
    # 方式二
    emps=session.query(Employess).filter(or_(Employess.emp_no>10025),(Employess.emp_no>10005))
    show(emps)

    #NOT
    # 方式一
    emps=session.query(Employess).filter(not_(Employess.emp_no>10002))
    show(emps)

    #方式二
    emps=session.query(Employess).filter(~(Employess.emp_no>10002))
    show(emps)

4 排序和分页

    # 排序，默认是升序排列
    # 升序排列
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no)
    print ("升序排列")
    show(emps)
    emps = session.query(Employess).filter(Employess.emp_no > 10025).order_by(Employess.emp_no.asc())
    show(emps)
    # 降序排列
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.emp_no.desc())
    print ("降序排列")
    show(emps)

    # 多列排序,first_name 为降序排列,emp_no为升序排列，默认只有两个first_name相同时，后面的才有意义
    emps=session.query(Employess).filter(Employess.emp_no>10025).order_by(Employess.first_name.desc()).order_by(Employess.emp_no)
    print  ("多列排序")
    show(emps)
    # 分页
    emps=session.query(Employess).limit(4)
    print  ("分页")
    show(emps)
    emps=session.query(Employess).limit(4).offset(20)
    show(emps)

结果如下

5 查询删除

    emps=session.query(Employess.emp_no).filter(Employess.emp_no>10025).delete()
    print ('查询结果为:',emps)
    # session.commit()

结果如下

查询到的东西就是想要的，直接删除，此处对其进行了优化，直接使用where语句进行处理，而没有经过select处理

6 关联查询

1 需求

查询10010 员工所在的部门编号
使用隐式内连接

---

    # 一般连接方式处理如下
    results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隐式内连接第一种写法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all()
    show(results)
    # 隐式内连接第二种写法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    print (results)

第一种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date, dept_emp.emp_no AS dept_emp_emp_no, dept_emp.dept_no AS dept_emp_dept_no, dept_emp.from_date AS dept_emp_from_date, dept_emp.to_date AS dept_emp_to_date 
FROM employees, dept_emp 
WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s

第二种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no 
WHERE employees.emp_no = %(emp_no_1)s

第三种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no 
WHERE employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s

第四种写法最终翻译结果如下

SELECT employees.emp_no AS employees_emp_no, employees.birth_date AS employees_birth_date, employees.first_name AS employees_first_name, employees.last_name AS employees_last_name, employees.gender AS employees_gender, employees.hire_date AS employees_hire_date 
FROM employees INNER JOIN dept_emp ON employees.emp_no = dept_emp.emp_no AND employees.emp_no = %(emp_no_1)s
2019-09-21 15:20:08,776 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10010}

显示如下

这种写法，都返回一行数据，原因是query(Employee)这只能返回一个实体对象中去，为了解决这个问题，需要修改实体类Employee，增加属性用来存放部门信息
sqlalchemy.orm.relationship(实体类名字符串)

2 使用relationship来显示其他表的信息

#!/usr/bin/poython3.6
#conding:utf-8
import pymysql
import sqlalchemy
from sqlalchemy import create_engine
# 导入基类模块
from sqlalchemy.ext.declarative import declarative_base
# 导入实体类相关属性模块
from sqlalchemy import Column, Integer, String,Enum,DATE,ForeignKey
# 导入会话模块
from sqlalchemy.orm import sessionmaker,relationship
# 导入枚举类型
import enum
#创建存储引擎
username='root'
password='666666'
ip='192.168.1.120'
port=3306
databases='test'
engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}'.format(username,password,ip,port,databases))
# 导入基类
Base=declarative_base()
# 创建要查询的表对应的实体类
class Dept_emp(Base):
    __tablename__='dept_emp'
    emp_no=Column(Integer,ForeignKey('employees.emp_no',ondelete='CASAED'),primary_key=True)  # 整形，非空，主键,ForeignKey用于定义外键，
    dept_no=Column(String(4),ForeignKey('departments.dept_no',ondelete='CASCAED'),primary_key=True)
    from_date=Column(DATE,nullable=False)
    to_date=Column(DATE,nullable=False)
    def __repr__(self):
        return "< {}  empno={}  dept_no={}>".format(self.__class__.__name__,self.emp_no,self.dept_no)
# 创建枚举类
class  MyEnum(enum.Enum):
    M='M'
    F='F'
# 创建对应实体类
class  Employess(Base):
    __tablename__='employees'
    emp_no=Column(Integer,nullable=False,primary_key=True)  # 整形，非空，主键
    birth_date=Column(DATE,nullable=False)  # 日期类型，非空
    first_name=Column(String(14),nullable=False)  # 字符串类型，长度为14，非空
    last_name=Column(String(14),nullable=False)  # 字符串类型，长度为16，非空
    gender=Column(Enum(MyEnum),nullable=False)  # 枚举，非空
    hire_date=Column(DATE,nullable=False)  # 字符类型，非空
    # 增加属性
    depts=relationship('Dept_emp')  # 后面接一个类名，这几个的外键关系已经被记录了，在dept_mp中已经表达过了，生成语句的投影和employee有关， 此处生成一个属性来存储部门之间的关系，如果相反，则无法记录上面的不存在的名字，是依靠Colum中的外键中进行定义处理这个东西的
    def __repr__(self):  # 定义查询返回
        return "< {}  emp_no:{}  name:{}  depts:{}>  ".format(self.__class__.__name__,self.emp_no,"{} {}".format(self.first_name,self.last_name),self.depts)
    __str__=__repr__

class  Department(Base):
    __tablename__='departments'
    dept_no=Column(String(32),primary_key=True)
    dept_name=Column(String(40),nullable=False,unique=True)
    def __repr__(self):
        return "< {}  no={}  name={}>".format(self.__class__.__name__,self.dept_no,self.dept_name)

    __str__=__repr__

#将对应实体类加入存储引擎。执行相关指令
Base.metadata.create_all(engine)
#创建会话
Session = sessionmaker(bind=engine)  #此处返回的是一个类，需要对其进行实例化相关操作
#对返回的类进行实例化
# session对象线程不安全，所以不同线程使用不能的session对象，。Session和engine都是线程安全的，有一个就行了
session=Session()
#  实例化并进行插入操作

def  show(emps):
    for x  in emps:
        print (x)
    print ('------------',end='\n\n')

try:
    # 一般连接方式处理如下
    results=session.query(Employess,Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隐式内连接第一种写法，此处的query 写谁，最终其显示结果就是谁的表的列处理 results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==10010).all()
    show(results)
    # 隐式内连接第二种写法
    results=session.query(Employess).join(Dept_emp).filter(Employess.emp_no==Dept_emp.emp_no).filter(Employess.emp_no==10010).all()
    show(results)
    # 隐式连接第三种想法
    results=session.query(Employess).join(Dept_emp,(Employess.emp_no==Dept_emp.emp_no)&(Employess.emp_no==10010)).all()
    show(results)
except  Exception as  e:   # 若抛出异常，则直接回滚
    print ('异常如下：--------',e)
finally:
    pass

结果如下

3 总结 :

第二种写法中join(Dept_empo)中没有等值条件，会自动生成等值条件，如果后面有filter，哪怕是filter(Employess.emp_no==Dept_emp.emp_no),这个条件会在where中出现，第二种自动增加的join等值条件的方式不好，不建议这样写

第三种在join中增加等值条件，组织了自动的等值条件生成，这种方式是推荐的

第四种方式和第三种是相同的，也可以使用

8 总结

在开发中，一般都会采用ORM框架，这样就可以使用对象操作表了，

---

定义表映射的类，使用Cloumn的描述器定义类，使用Foreignkey来定义外键约束
如果在一个对象中，想看看其他表对应的独享内容，就要使用relationship来定义关系。

当前题目：python之SQLAIchemy
分享地址：http://myzitong.com/article/ijesjh.html