62数据库7_SQLAlchemy复杂查询

发布时间：2020-06-30 23:55:04 作者：chaijowin
来源：网络阅读：268

简单条件查询：... 1

与and_（&）、或or_（|）、非not_（~）：... 4

in_、notin_、like、ilike：... 5

排序：... 5

分页：... 6

消费者方法：... 6

聚合、分组：... 7

关联查询：... 7

隐式内连接：... 7

使用join. 8

first方法：... 11

querying：

复杂查询：

注：

多对多，要加张表；

简单条件查询：

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from sqlalchemy import Column, Integer, String, Date, Enum, inspect, ForeignKey

import enum

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker, relationship

class MyEnum(enum.Enum): #enum为内建模块

M = 'M'

F = 'F'

# DROP TABLE IF EXISTS `employees`;

# CREATE TABLE `employees` (

# `emp_no` int(11) NOT NULL,

# `birth_date` date NOT NULL,

# `first_name` varchar(14) NOT NULL,

# `last_name` varchar(16) NOT NULL,

# `gender` enum('M','F') NOT NULL,

# `hire_date` date NOT NULL,

# PRIMARY KEY (`emp_no`)

# ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

class Employee(Base):

__tablename__ = 'employee'

emp_no = Column(Integer, primary_key=True)

birth_date = Column(Date, nullable=False)

first_name = Column(String(14), nullable=False)

last_name = Column(String(16), nullable=False)

gender = Column(Enum(MyEnum), nullable=False)

hire_date = Column(Date, nullable=False)

dept_emp = relationship('Dept_emp') #relationship中有userlist=False，表示一对一关系

def __repr__(self):

return '<{} emp_no:{} name:{}_{} gender={} dept_emp={}>'.format(

self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender, self.dept_emp

)

# DROP TABLE IF EXISTS `departments`;

# CREATE TABLE `departments` (

# `dept_no` char(4) NOT NULL,

# `dept_name` varchar(40) NOT NULL,

# PRIMARY KEY (`dept_no`),

# UNIQUE KEY `dept_name` (`dept_name`)

# ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

class Department(Base):

__tablename__ = 'department'

dept_no = Column(String(4), primary_key=True)

dept_name = Column(String(40), nullable=False, unique=True)

def __repr__(self):

return '<{} dept_no={} dept_name={}>'.format(self.__class__.__name__, self.dept_no, self.dept_name)

# DROP TABLE IF EXISTS `dept_emp`;

# CREATE TABLE `dept_emp` (

# `emp_no` int(11) NOT NULL,

# `dept_no` char(4) NOT NULL,

# `from_date` date NOT NULL,

# `to_date` date NOT NULL,

# PRIMARY KEY (`emp_no`,`dept_no`),

# KEY `dept_no` (`dept_no`),

# CONSTRAINT `dept_emp_ibfk_1` FOREIGN KEY (`emp_no`) REFERENCES `employees` (`emp_no`) ON DELETE CASCADE,

# CONSTRAINT `dept_emp_ibfk_2` FOREIGN KEY (`dept_no`) REFERENCES `departments` (`dept_no`) ON DELETE CASCADE

# ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

class Dept_emp(Base):

__tablename__ = 'dept_emp'

emp_no = Column(Integer, ForeignKey('employee.emp_no', ondelete='CASCADE'), primary_key=True,) #Integer类型不需要传参，没有__init__()方法；ForeignKey定义外键约束，注意FK里的参数是表名.字段（全小写），而不是类名.属性（类名的首字母大写），若用类名.字段会报错

dept_no = Column(String(4), ForeignKey('department.dept_no', ondelete='CASCADE'), primary_key=True, ) #注意关键字参数要放到最后

from_date = Column(Date, nullable=False)

to_date = Column(Date, nullable=False)

def __repr__(self):

return '<{} emp_no={} dept_no={}>'.format(self.__class__.__name__, self.emp_no, self.dept_no)

def show(entity):

for x in entity:

print(x)

print('###########', end='\n\n')

host = '10.113.129.2'

port = 3306

user = 'root'

password = 'rootqazwsx'

database = 'test1'

conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format(user, password, host, port, database)

engine = create_engine(conn_str, echo=True)

# Base.metadata.drop_all(engine)

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)

session = Session()

emps = session.query(Employee).filter(Employee.emp_no > 10015)

show(emps)

输出：

2018-10-11 16:30:13,657 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)

2018-10-11 16:30:13,658 INFO sqlalchemy.engine.base.Engine SELECT employee.emp_no AS employee_emp_no, employee.birth_date AS employee_birth_date, employee.first_name AS employee_first_name, employee.last_name AS employee_last_name, employee.gender AS employee_gender, employee.hire_date AS employee_hire_date

FROM employee

WHERE employee.emp_no > %(emp_no_1)s

2018-10-11 16:30:13,658 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10015}

###########

与and_（&）、或or_（|）、非not_（~）：

from sqlalchemy import and_, or_, not_

emps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.gender == MyEnum.F) #链式编程

emps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.gender == MyEnum.F)) #and_()

emps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.gender == MyEnum.F)) #&，两边的表达式要有括号

emps = session.query(Employee).filter(or_(Employee.emp_no > 10015, Employee.gender == MyEnum.F)) #or_()

emps = session.query(Employee).filter((Employee.emp_no > 10015) | (Employee.gender == MyEnum.F)) #|，两边的表达式要加括号

emps = session.query(Employee).filter(not_(Employee.emp_no < 10018)) #not_()

emps = session.query(Employee).filter(~(Employee.emp_no < 10018)) #~后的表达式要加括号

show(emps)

in_、notin_、like、ilike：

emplist = [10010, 10015, 10018]

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

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

mps = session.query(Employee).filter(Employee.last_name.like('P%')) #ilike，忽略大小写

show(emps)

排序：

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no) #默认升序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.asc()) #升序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.desc()) #降序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name).order_by(Employee.emp_no.desc()) #多列排序

show(emps)

分页：

网页展示用到；

emps = session.query(Employee).limit(4)

emps = session.query(Employee).limit(4).offset(12) #offset偏移，先偏移再展示指定的limit数

show(emps)

消费者方法：

调用后，query对象（可迭代）就转换成了一个容器；

emps = session.query(Employee)

print('~~~~~~~~~~', len(list(emps))) #总行数，返回大量结果集，再转为list，不要用

print('@@@@@@@@@@@', len(emps.all())) #同list(emps)，取所有数据，返回列表，不要用

print('###########', emps.count()) #聚合函数count(*)查询，实质调用的是count(*)

# print(emps.one()) #返回一行，如果查询结果是多行会抛异常，sqlalchemy.orm.exc.MultipleResultsFound: Multiple rows were found for one()

# print('~~~~~~~~~~~', emps.limit(1)) #emps.limit(1)是可迭代对象，要展示遍历即可，如emps=emps.limit(1);show(emps)

print('@@@@@@@@@@', emps.limit(1).one()) #返回一行，用limit(1).one()或get()取PK的一条

print('##########', emps.get(10010)) #同emps = session.query(Employee).filter(Employee.emp_no == 10010);emps.one()

session.query(Employee).filter(Employee.emp_no > 10018).delete() #delete by query

session.commit()

emps = session.query(Employee).filter(Employee.emp_no > 10010)

count = emps.count()

print(count)

emps = emps.limit(6)

show(emps)

聚合、分组：

from sqlalchemy import func

query = session.query(func.count(Employee.emp_no))

query = session.query(func.max(Employee.emp_no))

query = session.query(func.min(Employee.emp_no))

query = session.query(func.avg(Employee.emp_no))

print('~~~~~~~~~~', query.one()) #只能有一行结果，返回元组

print('@@@@@@@@@@', query.scalar()) #取one()返回元组的第一个元素

print(session.query(func.count(Employee.emp_no)).group_by(Employee.gender).all()) #分组后作聚合

关联查询：

需求：查询10010员工所在部门编号；

隐式内连接：

results = session.query(Employee, Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all() #隐式内连接

# results = session.query([Employee, Dept_emp]).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all() #用可迭代对象，[]或()，经测试有问题

show(results)

输出：

2018-10-12 09:42:57,050 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)

2018-10-12 09:42:57,053 INFO sqlalchemy.engine.base.Engine SELECT employee.emp_no AS employee_emp_no, employee.birth_date AS employee_birth_date, employee.first_name AS employee_first_name, employee.last_name AS employee_last_name, employee.gender AS employee_gender, employee.hire_date AS employee_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 employee, dept_emp

WHERE employee.emp_no = dept_emp.emp_no AND employee.emp_no = %(emp_no_1)s

2018-10-12 09:42:57,054 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10010}

(<Employee emp_no:10010 name:Duangkaew_Piveteau gender=MyEnum.F>, <Dept_emp emp_no=10010 dept_no=d004>)

(<Employee emp_no:10010 name:Duangkaew_Piveteau gender=MyEnum.F>, <Dept_emp emp_no=10010 dept_no=d006>)

###########

注：

SELECT

FROM

employee,

dept_emp

WHERE

employee.emp_no = dept_emp.emp_no

AND employee.emp_no = 10010;

使用join

# results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == 10010) #方式一，此种方式不要用，等值要自己写（如方式二.join(Dept_emp, Employee.emp_no == Dept_emp.emp_no)），生成的select语句中会自动加上ON，而自动生成的这个ON有时不是我们想要的，即便多加一个filter也没用，还是会自动加上ON

print(results.count()) #查询结果为1，但count()为2，解决sqlalchemy.orm.relationship('实体类名字符串')

results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all() #方式二；这两种写法，返回都只有一行数据，原因在于query(Employee)只能返回一个实体对象，解决：修改实体类Employee，增加属性用来存放部门信息

show(results)

输出：

2018-10-12 10:36:44,180 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)

2018-10-12 10:36:44,181 INFO sqlalchemy.engine.base.Engine SELECT employee.emp_no AS employee_emp_no, employee.birth_date AS employee_birth_date, employee.first_name AS employee_first_name, employee.last_name AS employee_last_name, employee.gender AS employee_gender, employee.hire_date AS employee_hire_date

FROM employee INNER JOIN dept_emp ON employee.emp_no = dept_emp.emp_no

WHERE employee.emp_no = %(emp_no_1)s

2018-10-12 10:36:44,181 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10010}

###########

注：

等价于SQL语句：

SELECT

FROM

(

SELECT

emp.emp_no AS emp_no,

emp.last_name AS emp_name,

dept.dept_no AS dept_no

FROM

employee AS emp

INNER JOIN dept_emp AS dept ON emp.emp_no = dept.emp_no

) AS mid_tab

WHERE

mid_tab.emp_no = 10010;

62数据库7_SQLAlchemy复杂查询

解决：

多表查询中的relationship；

from sqlalchemy.orm import relationship

class Employee(Base):

__tablename__ = 'employee'

emp_no = Column(Integer, primary_key=True)

birth_date = Column(Date, nullable=False)

first_name = Column(String(14), nullable=False)

last_name = Column(String(16), nullable=False)

gender = Column(Enum(MyEnum), nullable=False)

hire_date = Column(Date, nullable=False)

dept_emp = relationship('Dept_emp') #官方要求，类名要引起来，FK在Dept_emp中写明了此处可省

def __repr__(self):

return '<{} emp_no:{} name:{}_{} gender={} dept_emp={}>'.format(

self.__class__.__name__, self.emp_no, self.first_name, self.last_name, self.gender, self.dept_emp

) #若将此处self.dept_emp改为self.emp_no输出结果中将不会有dept_emp语句产生

# results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010) #写法1

# results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010) #写法2

results = session.query(Employee).join(Dept_emp, (Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010)) #写法3

show(results)

输出：

2018-10-12 11:03:24,694 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)

2018-10-12 11:03:24,695 INFO sqlalchemy.engine.base.Engine SELECT employee.emp_no AS employee_emp_no, employee.birth_date AS employee_birth_date, employee.first_name AS employee_first_name, employee.last_name AS employee_last_name, employee.gender AS employee_gender, employee.hire_date AS employee_hire_date

FROM employee INNER JOIN dept_emp ON employee.emp_no = dept_emp.emp_no

WHERE employee.emp_no = %(emp_no_1)s

2018-10-12 11:03:24,695 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 10010}

2018-10-12 11:03:24,710 INFO sqlalchemy.engine.base.Engine SELECT 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 dept_emp

WHERE %(param_1)s = dept_emp.emp_no

2018-10-12 11:03:24,710 INFO sqlalchemy.engine.base.Engine {'param_1': 10010}

###########

first方法：

emps = session.query(Employee).filter(Employee.emp_no > 500000).first()

print(emps, type(emps))

输出：

2018-11-29 09:27:49,179 INFO sqlalchemy.engine.base.Engine {'emp_no_1': 500000, 'param_1': 1}

None <class 'NoneType'>

应用在项目blog，blog/handler/user.py的reg接口，代码为：

@user_router.post('/reg')

def reg(ctx, request: MagWeb.Request):

# print(request)

# print(request.json)

payload = request.json

# print(payload, type(payload))

email = payload.get('email')

if session.query(User).filter(User.email == email).first() is not None:

raise exc.HTTPConflict()

注，first源码：

def first(self):

"""Return the first result of this ``Query`` or

None if the result doesn't contain any row.

first() applies a limit of one within the generated SQL, so that

only one primary entity row is generated on the server side

(note this may consist of multiple result rows if join-loaded

collections are present).

Calling :meth:`.Query.first` results in an execution of the underlying query.

.. seealso::

:meth:`.Query.one`

:meth:`.Query.one_or_none`

"""

if self._statement is not None:

ret = list(self)[0:1]

else:

ret = list(self[0:1])

if len(ret) > 0:

return ret[0]

else:

return None

62数据库7_SQLAlchemy复杂查询

简单条件查询：

与and_（&）、或or_（|）、非not_（~）：

in_、notin_、like、ilike：

排序：

分页：

消费者方法：

聚合、分组：

关联查询：

隐式内连接：

使用join

first方法：

相关阅读