查询运算符

peewee 支持以下类型的比较

比较

含义

==

x 等于 y

<

x 小于 y

<=

x 小于或等于 y

>

x 大于 y

>=

x 大于或等于 y

!=

x 不等于 y

<<

x IN y,其中 y 是一个列表或查询

>>

x IS y,其中 y 是 None/NULL

%

x LIKE y,其中 y 可能包含通配符

**

x ILIKE y,其中 y 可能包含通配符

^

x XOR y

~

一元否定(例如,NOT x)

由于我用完了要覆盖的运算符,因此还有一些其他查询操作可作为方法使用

方法

含义

.in_(value)

IN 查找(与 << 相同)。

.not_in(value)

NOT IN 查找。

.is_null(is_null)

IS NULL 或 IS NOT NULL。接受布尔参数。

.contains(substr)

子字符串的通配符搜索。

.startswith(prefix)

搜索以 prefix 开头的值。

.endswith(suffix)

搜索以 suffix 结尾的值。

.between(low, high)

搜索 low <= value <= high 的值。

.regexp(exp)

正则表达式匹配(区分大小写)。

.iregexp(exp)

正则表达式匹配(不区分大小写)。

.bin_and(value)

二进制 AND。

.bin_or(value)

二进制 OR。

.concat(other)

使用 || 连接两个字符串或对象。

.distinct()

标记列以进行 DISTINCT 选择。

.collate(collation)

使用给定的排序规则指定列。

.cast(type)

将列的值转换为给定的类型。

要使用逻辑运算符组合子句,请使用

运算符

含义

示例

&

AND

(User.is_active == True) & (User.is_admin == True)

|(管道)

OR

(User.is_admin) | (User.is_superuser)

~

NOT(一元否定)

~(User.username.contains('admin'))

以下是你可以使用这些查询运算符的一些方法

# Find the user whose username is "charlie".
User.select().where(User.username == 'charlie')

# Find the users whose username is in [charlie, huey, mickey]
User.select().where(User.username.in_(['charlie', 'huey', 'mickey']))

# Find users whose salary is between 50k and 60k (inclusive).
Employee.select().where(Employee.salary.between(50000, 60000))

Employee.select().where(Employee.name.startswith('C'))

Blog.select().where(Blog.title.contains(search_string))

以下是你可以组合表达式的几种方法。比较可以任意复杂。

注意

请注意,实际比较用括号括起来。Python 的运算符优先级要求比较用括号括起来。

# Find any users who are active administrations.
User.select().where(
  (User.is_admin == True) &
  (User.is_active == True))

# Find any users who are either administrators or super-users.
User.select().where(
  (User.is_admin == True) |
  (User.is_superuser == True))

# Alternatively, use the boolean values directly. Here we query users who
# are admins and NOT superusers.
User.select().where(User.is_admin & ~User.is_superuser)

# Find any Tweets by users who are not admins (NOT IN).
admins = User.select().where(User.is_admin == True)
non_admin_tweets = Tweet.select().where(Tweet.user.not_in(admins))

# Find any users who are not my friends (strangers).
friends = User.select().where(User.username.in_(['charlie', 'huey', 'mickey']))
strangers = User.select().where(User.id.not_in(friends))

警告

虽然你可能很想在查询表达式中使用 python 的 inandorisnot 运算符,但这些运算符不起作用。in 表达式的返回值始终强制转换为布尔值。类似地,andornot 都将它们的实参视为布尔值,并且不能重载。

所以请记住

  • 使用 .in_().not_in() 代替 innot in

  • 使用 & 代替 and

  • 使用 | 代替 or

  • 使用 ~ 代替 not

  • 使用 .is_null() 代替 is None== None

  • 使用 ==!=TrueFalse 进行比较,或者你可以使用表达式的隐式值。

  • 使用逻辑运算符时,不要忘记用括号括起来你的比较。

有关更多示例,请参阅 表达式 部分。

注意

在 SQLite 中使用 LIKE 和 ILIKE

由于 SQLite 的 LIKE 操作在默认情况下不区分大小写,因此 peewee 将使用 SQLite GLOB 操作进行区分大小写的搜索。glob 操作使用星号作为通配符,而不是通常的百分号。如果你正在使用 SQLite 并且想要进行区分大小写的部分字符串匹配,请记住对通配符使用星号。

三值逻辑

由于 SQL 处理 NULL 的方式,因此有一些特殊操作可用于表示

  • IS NULL

  • IS NOT NULL

  • IN

  • NOT IN

虽然可以使用否定运算符 (~) 和 IS NULLIN 运算符,但有时为了获得正确的语义,你需要明确使用 IS NOT NULLNOT IN

使用 IS NULLIN 的最简单方法是使用运算符重载

# Get all User objects whose last login is NULL.
User.select().where(User.last_login >> None)

# Get users whose username is in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username << usernames)

如果你不喜欢运算符重载,你可以改用调用字段方法

# Get all User objects whose last login is NULL.
User.select().where(User.last_login.is_null(True))

# Get users whose username is in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username.in_(usernames))

要否定上述查询,可以使用一元否定,但为了获得正确的语义,你可能需要使用特殊 IS NOTNOT IN 运算符

# Get all User objects whose last login is *NOT* NULL.
User.select().where(User.last_login.is_null(False))

# Using unary negation instead.
User.select().where(~(User.last_login >> None))

# Get users whose username is *NOT* in the given list.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(User.username.not_in(usernames))

# Using unary negation instead.
usernames = ['charlie', 'huey', 'mickey']
User.select().where(~(User.username << usernames))

添加用户定义运算符

由于我用完了要重载的 python 运算符,peewee 中缺少一些运算符,例如 modulo。如果你发现需要支持表中没有的运算符,添加你自己的运算符非常容易。

以下是如何在 SQLite 中添加对 modulo 的支持

from peewee import *
from peewee import Expression  # The building block for expressions.

def mod(lhs, rhs):
    # Note: this works with Sqlite, but some drivers may use string-
    # formatting before sending the query to the database, so you may
    # need to use '%%' instead here.
    return Expression(lhs, '%', rhs)

现在,你可以使用这些自定义运算符来构建更丰富的查询

# Users with even ids.
User.select().where(mod(User.id, 2) == 0)

有关更多示例,请查看 playhouse.postgresql_ext 模块的源代码,因为它包含许多特定于 postgresql 的 hstore 的运算符。

表达式

Peewee 旨在提供一种简单、富有表现力且符合 python 习惯的方式来构建 SQL 查询。本节将快速概述一些常见的表达式类型。

有两种主要类型的对象可以组合起来创建表达式

  • Field 实例

  • 使用 fn 的 SQL 聚合和函数

我们假设一个简单的“User”模型,其中包含用于用户名和其他内容的字段。它看起来像这样

class User(Model):
    username = CharField()
    is_admin = BooleanField()
    is_active = BooleanField()
    last_login = DateTimeField()
    login_count = IntegerField()
    failed_logins = IntegerField()

比较使用 查询运算符

# username is equal to 'charlie'
User.username == 'charlie'

# user has logged in less than 5 times
User.login_count < 5

可以使用按位组合比较。运算符优先级由 python 控制,并且可以将比较嵌套到任意深度

# User is both and admin and has logged in today
(User.is_admin == True) & (User.last_login >= today)

# User's username is either charlie or charles
(User.username == 'charlie') | (User.username == 'charles')

# User is active and not a superuser.
(User.is_active & ~User.is_superuser)

比较也可以与函数一起使用

# user's username starts with a 'g' or a 'G':
fn.Lower(fn.Substr(User.username, 1, 1)) == 'g'

我们可以做一些相当有趣的事情,因为可以将表达式与其他表达式进行比较。表达式还支持算术运算

# users who entered the incorrect more than half the time and have logged
# in at least 10 times
(User.failed_logins > (User.login_count * .5)) & (User.login_count > 10)

表达式允许我们执行原子更新

# when a user logs in we want to increment their login count:
User.update(login_count=User.login_count + 1).where(User.id == user_id)

表达式可以在查询的所有部分中使用,所以请尝试一下!

行值

许多数据库支持行值,类似于 Python tuple 对象。在 Peewee 中,可以通过 Tuple 在表达式中使用行值。例如,

# If for some reason your schema stores dates in separate columns ("year",
# "month" and "day"), you can use row-values to find all rows that happened
# in a given month:
Tuple(Event.year, Event.month) == (2019, 1)

行值更常见的用途是在单个表达式中针对子查询中的多列进行比较。还有其他方法来表达这些类型的查询,但行值可能提供一种简洁且可读的方法。

例如,假设我们有一个表“EventLog”,其中包含事件类型、事件源和一些元数据。我们还有一个“IncidentLog”,其中包含事件类型、事件源和元数据列。我们可以使用行值将事件与某些事件相关联

class EventLog(Model):
    event_type = TextField()
    source = TextField()
    data = TextField()
    timestamp = TimestampField()

class IncidentLog(Model):
    incident_type = TextField()
    source = TextField()
    traceback = TextField()
    timestamp = TimestampField()

# Get a list of all the incident types and sources that have occured today.
incidents = (IncidentLog
             .select(IncidentLog.incident_type, IncidentLog.source)
             .where(IncidentLog.timestamp >= datetime.date.today()))

# Find all events that correlate with the type and source of the
# incidents that occured today.
events = (EventLog
          .select()
          .where(Tuple(EventLog.event_type, EventLog.source).in_(incidents))
          .order_by(EventLog.timestamp))

表达此类查询的其他方法是使用 联接联接子查询。上面的示例只是为了让你了解如何使用 Tuple

当新数据源自子查询时,你还可以使用行值来更新表中的多列。有关示例,请参阅 此处

SQL 函数

SQL 函数,如 COUNT()SUM(),可以使用 fn() 帮助程序来表达

# Get all users and the number of tweets they've authored. Sort the
# results from most tweets -> fewest tweets.
query = (User
         .select(User, fn.COUNT(Tweet.id).alias('tweet_count'))
         .join(Tweet, JOIN.LEFT_OUTER)
         .group_by(User)
         .order_by(fn.COUNT(Tweet.id).desc()))
for user in query:
    print('%s -- %s tweets' % (user.username, user.tweet_count))

fn 帮助程序将任何 SQL 函数公开,就好像它是方法一样。参数可以是字段、值、子查询,甚至嵌套函数。

嵌套函数调用

假设你需要获取所有用户名以a开头的用户列表。有几种方法可以做到这一点,但一种方法可能是使用一些 SQL 函数,如LOWERSUBSTR。要使用任意 SQL 函数,请使用特殊 fn() 对象来构造查询

# Select the user's id, username and the first letter of their username, lower-cased
first_letter = fn.LOWER(fn.SUBSTR(User.username, 1, 1))
query = User.select(User, first_letter.alias('first_letter'))

# Alternatively we could select only users whose username begins with 'a'
a_users = User.select().where(first_letter == 'a')

>>> for user in a_users:
...    print(user.username)

SQL 帮助程序

有时你可能只想传递一些任意 sql。你可以使用特殊 SQL 类来实现此目的。一种用例是引用别名

# We'll query the user table and annotate it with a count of tweets for
# the given user
query = (User
         .select(User, fn.Count(Tweet.id).alias('ct'))
         .join(Tweet)
         .group_by(User))

# Now we will order by the count, which was aliased to "ct"
query = query.order_by(SQL('ct'))

# You could, of course, also write this as:
query = query.order_by(fn.COUNT(Tweet.id))

有两种方法可以使用 peewee 执行手工制作的 SQL 语句

  1. Database.execute_sql() 用于执行任何类型的查询

  2. RawQuery 用于执行 SELECT 查询并返回模型实例。

安全性和 SQL 注入

默认情况下,peewee 会对查询进行参数化,因此用户传入的任何参数都将被转义。此规则的唯一例外是,如果您正在编写原始 SQL 查询或传入可能包含不可信数据的 SQL 对象。为了缓解此问题,请确保将任何用户定义的数据作为查询参数传入,而不是作为实际 SQL 查询的一部分

# Bad! DO NOT DO THIS!
query = MyModel.raw('SELECT * FROM my_table WHERE data = %s' % (user_data,))

# Good. `user_data` will be treated as a parameter to the query.
query = MyModel.raw('SELECT * FROM my_table WHERE data = %s', user_data)

# Bad! DO NOT DO THIS!
query = MyModel.select().where(SQL('Some SQL expression %s' % user_data))

# Good. `user_data` will be treated as a parameter.
query = MyModel.select().where(SQL('Some SQL expression %s', user_data))

注意

MySQL 和 Postgresql 使用 '%s' 来表示参数。另一方面,SQLite 使用 '?'。务必使用适合您数据库的字符。您还可以通过检查 Database.param 来查找此参数。