62数据库7_SQLAlchemy复杂查询

目录

简单条件查询：... 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}

<Employee emp_no:10016 name:Kazuhito_Cappelletti gender=MyEnum.M>

<Employee emp_no:10017 name:Cristinel_Bouloucos gender=MyEnum.F>

<Employee emp_no:10018 name:Kazuhide_Peha gender=MyEnum.F>

<Employee emp_no:10019 name:Lillian_Haddadi gender=MyEnum.M>

<Employee emp_no:10020 name:Mayuko_Warwick gender=MyEnum.M>

###########

与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}

<Employee emp_no:10010 name:Duangkaew_Piveteau gender=MyEnum.F>

###########

注：

等价于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}

<Employee emp_no:10010 name:Duangkaew_Piveteau gender=MyEnum.F dept_emp=[<Dept_emp emp_no=10010 dept_no=d004>, <Dept_emp emp_no=10010 dept_no=d006>]>

###########

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