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