测开之路一百四十四：ORM之SQLAlchemy查询

Posted 2022-10-16 zhongyehai

 

在上一篇的基础上，插入数据

 

查询

Department.query.all() # 用表对象查
db.session.query(Department).all() # 用db对象查

 

查询前两条，直接python截取前两条数据

 

排序

默认升序：order_by(排序的字段名)

倒序：从sqlalchemy导入desc

查询单个指定字段

查询多个指定字段

只用一个变量接收多个查询结果的时候，SQLAlchemy默认用namedtuple进行命名处理

namedtuple结构：

别名：label(别名)

限制查询返回数量：limit()

和排序组合用

跳过一部分，分页需要用到此功能：offset()

查询指定条件：filter_by(name=‘xxx‘)

也可以用面向对象的形式：filter(对象.xxx==‘xxx‘)

使用python语法模糊查询：

包含：contains()

以xx开始：startwith()

以xx结尾：endwith()

比较：直接写运算符

多条件：无限.filter()，每家一个条件就加一个.filter()

由于.filter()和.filter_by()需要依附query对象，而.filter()和.filter_by()本身返回的就是query对象，所以可以无限过滤下去

联合查询：从Employee里面找出Department.name == ‘技术部‘的数据

框架的模糊查询：

.like(%%)

in

not in，取反

取某个字段为空的数据

取某个字段不为空的数据：isnot()，或者取反

and：filter()里面加条件默认是and

也可以多个filter筛选

sqlalchemy自带的and，需导入

或：sqlalchemy自带的or，需导入

取查询的第一个：first()、one()、one_or_none()

单个值：scalar()

使用原本的sql片段，需要从sqlalchemy导入text()：from sqlalchemy import text

sqlalchemy与sql混用

查询个数：count()

也可以使用sqlalchemy自带的func.count()

 

from pms.models import *

all = Department.query.all()  # 用表对象查
print(all)
all_department = db.session.query(Department).all()  # 用db对象查
print(all_department)
for d in all:
    print(f‘d.id-->d.name‘)

for d in Department.query.all()[:2]:
    print(f‘d.id-->d.name‘)

# 默认升序
for d in db.session.query(Department).order_by(Department.id):
    print(f‘d.id-->d.name‘)

# 倒序
from sqlalchemy import desc

for n in db.session.query(Department).order_by(desc(Department.id)):  # 根据部门倒序排，取前两条
    print(n)

for i in db.session.query(Department.id):  # 查询id字段
    print(i)

for n in db.session.query(Department.name):  # 查询name字段
    print(n)

for i, n in db.session.query(Department.id, Department.name):  # 查询多个指定字段
    print(f‘i--n‘)

for row in db.session.query(Department.id, Department.name):  # 查询多个指定字段, 返回SQLAlchemy的类型
    print(f‘type(row)：\\nrow‘)
    print(f‘row.id-->row.name‘)  # 取值时不用下标，用对象方式

# namedtuple，类似数据库结构
from collections import namedtuple

name = namedtuple(‘Book‘, [‘title‘, ‘price‘, ‘auther‘])
book1 = name(‘python‘, 30, ‘aaa‘)
print(f‘book1.title、book1.price、book1.auther‘)

for n in db.session.query(Department.name.label("部门")):  # 查询name字段
    print(n.部门)

for n in Department.query.limit(2):  # 前两条
    print(n)

for n in db.session.query(Department).order_by(Department.id).limit(2):  # 根据部门升序排，取前两条
    print(n)

for n in db.session.query(Department).order_by(Department.id).offset(2).limit(2):  # 根据部门升序排，跳过两条，再取前两条
    print(n)

# 查询指定内容：filter_by
for d in db.session.query(Department).filter(Department.name == ‘技术部‘):
    print(d)

for d in db.session.query(Department).filter_by(name=‘技术部‘):
    print(d)

# 模糊查询，python语法
for d in db.session.query(Department).filter(Department.name.contains(‘技‘)):
    print(d)

for d in Department.query.filter(Department.name.startswith(‘财‘)):
    print(d)

for d in Department.query.filter(Department.name.endswith(‘部‘)):
    print(d)

for d in Department.query.filter(Department.id > 3):
    print(d)

for d in Department.query.filter(Department.id > 3).filter(Department.name.contains(‘市场部‘)):
    print(d)

for d in Department.query.filter(Department.id > 3).filter_by(name=‘市场部‘):
    print(d)

# 联合查询
for emp in db.session.query(Employee).filter(Department.name == ‘技术部‘):
    print(emp)

for emp in db.session.query(Employee).filter(Department.name == ‘技术部‘):
    print(emp)

for d in db.session.query(Department).filter(Department.name.like(‘%术%‘)):
    print(d)

# 第二个字是术的，一个横线代表一个位置
for d in db.session.query(Department).filter(Department.name.like(‘_术%‘)):
    print(d)

for d in db.session.query(Department).filter(Department.name.like(‘技%‘)):
    print(d)

for d in db.session.query(Department).filter(Department.name.like(‘%部‘)):
    print(d)

for d in db.session.query(Department).filter(Department.name.in_([‘技术部‘, ‘财务部‘])):
    print(d)

for d in db.session.query(Department).filter(~Department.name.in_([‘技术部‘, ‘财务部‘])):
    print(d)

for d in db.session.query(Employee).filter(Employee.name.is_(None)):
    print(d)

for d in db.session.query(Employee).filter(Employee.name == None):
    print(d)

for d in db.session.query(Employee).filter(Employee.name.isnot(None)):
    print(d)

for d in db.session.query(Employee).filter(~Employee.name.is_(None)):
    print(d)

# and
for d in db.session.query(Department).filter(Department.name.endswith(‘部‘), Department.id == 1):
    print(d)
for d in db.session.query(Department).filter(Department.name.endswith(‘部‘)).filter(Department.id == 1):
    print(d)
from sqlalchemy import and_
for d in db.session.query(Department).filter(and_(Department.name.endswith(‘部‘), Department.id == 1)):
    print(d)

# 或
from sqlalchemy import or_
for d in db.session.query(Department).filter(or_(Department.name.startswith(‘技‘), Department.id == 2)):
    print(d)

# 取第一个：
# first()，性能最优，且没有结果的时候返回空，不会抛异常
print(db.session.query(Department).order_by(desc(Department.id)).first())
# one()：没有返回结果的时候会抛异常
print(db.session.query(Department).order_by(desc(Department.id)).one())
# one_or_none()没有返回结果的时候不会抛异常，会返回none
print(db.session.query(Department).order_by(desc(Department.id)).one_or_none())

# 一行一列，单个值 scalar()
print(db.session.query(Department.id).filter(Department.name == ‘技术部‘).scalar())

# 使用原生sql条件片段：text(片段)
from sqlalchemy import text
for dep in db.session.query(Department).filter(text(‘id < 3‘)).order_by(text(‘id desc‘)).all():
    print(dep)

# 使用原生sql与sqlalchemy混用
from sqlalchemy import text
data = db.session.query(Department).from_statement(text(‘select * from department where name=:n‘)).params(n=‘技术部‘).one()
print(data)
stmt = text(‘select id, name, salary from employee where name=:name‘)
stmt = stmt.columns(Employee.id, Employee.name, Employee.salary)
data = db.session.query(Employee).from_statement(stmt).params(name=‘tom2‘).all()
print(data)

# 查询个数：count()
print(db.session.query(Department).count())
print(db.session.query(Employee).filter(Employee.name.startswith(‘t‘)).count())

from sqlalchemy import func
data = db.session.query(func.count(Employee.gender), Employee.gender).group_by(Employee.gender).all()
print(data)