表达式转换
Method Meaning .in_(value) IN lookup (identical to <<). .not_in(value) NOT IN lookup. .is_null(is_null) IS NULL or IS NOT NULL. Accepts boolean param. .contains(substr) Wild-card search for substring. .startswith(prefix) Search for values beginning with prefix. .endswith(suffix) Search for values ending with suffix. .between(low, high) Search for values between low and high. .regexp(exp) Regular expression match (case-sensitive). .iregexp(exp) Regular expression match (case-insensitive). .bin_and(value) Binary AND. .bin_or(value) Binary OR. .concat(other) Concatenate two strings or objects using ||. .distinct() Mark column for DISTINCT selection. .collate(collation) Specify column with the given collation. .cast(type) Cast the value of the column to the given type.
in_(): IN not_in(): NOT IN regexp(): REGEXP is_null(True/False): IS NULL or IS NOT NULL contains(s): LIKE %s% startswith(s): LIKE s% endswith(s): LIKE %s between(low, high): BETWEEN low AND high concat(): ||
举例:
SELECT * FROM user WHERE username not like "%admin%" # ~(User.username.contains('admin')) SELECT * FROM user WHERE LENGTH(username)>45 # fn.length(User.username) > 45
参考: http://docs.peewee-orm.com/en/latest/peewee/query_operators.html
调用sql函数
使用fn
query = (User .select(User.username, fn.COUNT(Tweet.id).alias('ct')) .join(Tweet, JOIN.LEFT_OUTER, on=(User.id == Tweet.user_id)) .group_by(User.username) .order_by(fn.COUNT(Tweet.id).desc()))
参考:https://peewee.readthedocs.io/en/latest/peewee/api.html#fn
以下代码参考官方示例
示例代码:
# —*— coding: utf-8 —*— from peewee import * import datetime from chinesename import chinesename # py2解决编码问题 import sys reload(sys) sys.setdefaultencoding("utf-8") # 设置数据库 db = SqliteDatabase("demo.db") class BaseModel(Model): class Meta: database = db # 定义数据表 class User(BaseModel): name = CharField(unique=True) def __str__(self): return "[user] id: %d name: %s"%(self.id, self.name) class Tweet(BaseModel): user = ForeignKeyField(User, related_name ="tweets") message = TextField() created_date = DateTimeField(default=datetime.datetime.now) is_published = BooleanField(default=True) def __str__(self): return "[tweet] id: %d name: %s" % (self.id, self.user.name) # 创建数据表 db.connect() db.create_tables([User, Tweet], safe=True) db.close() # 添加数据 def add_data(): cn = chinesename.ChineseName() for i in range(100): user = User(name=cn.getName()) user.save() User.create(name=cn.getName()) Tweet.create(user=user, message="hello world") # add_data() print datetime.datetime.now() print datetime.date.today() # 查询 ret = User.get(User.name=="沈从") if ret: print ret usernames = ["马酿", "沈从"] users = User.select().where(User.name.in_(usernames)) for user in users: print user tweets = Tweet.select().where(Tweet.user.in_(users)) for tweet in tweets: print tweet tweets = Tweet.select().join(User).where(User.name.in_(usernames)) for tweet in tweets: print tweet count = (Tweet .select() .where( (Tweet.created_date >= datetime.date.today())& (Tweet.is_published == True)) .count()) print count # 分页 page 3 (users 41-60) users = User.select().order_by(User.name).paginate(3, 20) for user in users: print user # 更新 query = User.update(name="西门吹雪").where(User.id==1) query.execute() # 删除 query = User.delete().where(User.id==2) query.execute()