创新互联公司坚持“要么做到，要么别承诺”的工作理念，服务领域包括：成都做网站、网站制作、成都外贸网站建设、企业官网、英文网站、手机端网站、网站推广等服务，满足客户于互联网时代的洪江网站设计、移动媒体设计的需求，帮助企业找到有效的互联网解决方案。努力成为您成熟可靠的网络建设合作伙伴！
目录
简单条件查询：...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}
###########
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)
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}
(
(
###########
注：
SELECT
*
FROM
employee,
dept_emp
WHERE
employee.emp_no = dept_emp.emp_no
AND employee.emp_no = 10010;
# 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;
解决：
多表查询中的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}
###########
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
应用在项目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
