?
創(chuàng)新互聯(lián)公司堅持“要么做到，要么別承諾”的工作理念，服務領域包括：成都做網(wǎng)站、網(wǎng)站制作、成都外貿(mào)網(wǎng)站建設、企業(yè)官網(wǎng)、英文網(wǎng)站、手機端網(wǎng)站、網(wǎng)站推廣等服務，滿足客戶于互聯(lián)網(wǎng)時代的洪江網(wǎng)站設計、移動媒體設計的需求，幫助企業(yè)找到有效的互聯(lián)網(wǎng)解決方案。努力成為您成熟可靠的網(wǎng)絡建設合作伙伴！
?
目錄
簡單條件查詢：...1
與and_（&）、或or_（|）、非not_（~）：...4
in_、notin_、like、ilike：...5
排序：...5
分頁：...6
消費者方法：...6
聚合、分組：...7
關聯(lián)查詢：...7
隱式內(nèi)連接：...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為內(nèi)建模塊
??? 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里的參數(shù)是表名.字段（全小寫），而不是類名.屬性（類名的首字母大寫），若用類名.字段會報錯
??? dept_no = Column(String(4), ForeignKey('department.dept_no', ondelete='CASCADE'), primary_key=True, )?? #注意關鍵字參數(shù)要放到最后
??? 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)
?
?
網(wǎng)頁展示用到；
?
emps = session.query(Employee).limit(4)
emps = session.query(Employee).limit(4).offset(12)?? #offset偏移，先偏移再展示指定的limit數(shù)
?
show(emps)
?
?
調(diào)用后，query對象（可迭代）就轉(zhuǎn)換成了一個容器；
?
emps = session.query(Employee)??
?
print('~~~~~~~~~~', len(list(emps)))?? #總行數(shù)，返回大量結(jié)果集，再轉(zhuǎn)為list，不要用
print('@@@@@@@@@@@', len(emps.all()))?? #同list(emps)，取所有數(shù)據(jù)，返回列表，不要用
print('###########', emps.count())?? #聚合函數(shù)count(*)查詢，實質(zhì)調(diào)用的是count(*)
?
# print(emps.one())?? #返回一行，如果查詢結(jié)果是多行會拋異常，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())?? #只能有一行結(jié)果，返回元組
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()?? #隱式內(nèi)連接
# results = session.query([Employee, Dept_emp]).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all()?? #用可迭代對象，[]或()，經(jīng)測試有問題
?
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())?? #查詢結(jié)果為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()?? #方式二；這兩種寫法，返回都只有一行數(shù)據(jù)，原因在于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')?? #官方要求，類名要引起來，F(xiàn)K在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輸出結(jié)果中將不會有dept_emp語句產(chǎn)生
?
# 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
重慶分公司
重慶分公司
