API 索引
模型
class Model(**kwargs)
模型提供对表的一对一的映射。Model的子类声明任意多的Field实例当作class属性。这些字段对应表上的列。
表层次的操作，例如select(),update(),insert(),and delete(),实现类方法。行层次的的操作比如
save(),delete_instance()实现为实例方法。
参数：kwargs-实例模型，给定的键值对赋值给相应的字段。
Example：
class User(Model): username = CharField() join_date = DateTimeField(default=datetime.datetime.now) is_admin = BooleanField() u = User(username='charlie', is_admin=True)
classmethod select(*selection)
参数：
- selection-模型类，字段实例，函数或表达式的列表。如果参数没有提供，给定模型的所有列将被选择
- 返回类型：给定模型的SelectQuey
选择所有列的例子：
选择所有Tweet和关联模型的所有列 。当user外键通过Tweet实例获取时，没有第二次查询：
(Tweet
.select(Tweet, User)
.join(User)
.order_by(Tweet.created_date.desc()))
classmethod update(*update)
参数：
- update -字段名称到表达式的映射
- 返回类型：UpdateQuery
显示注册过期的用户然后标定为非活跃用户：
q = User.update(active=False).where(User.registration_expired == True) q.execute() # Execute the query, updating the database.
原子性更新的例子：
q = PageView.update(count=PageView.count + 1).where(PageView.url == url) q.execute() # execute the query, updating the database.
更新查询执行时，修改行的数量返回
classmethod insert(**insert)
插入新行。如果字段有默认值，在insert参数字典里没有显式设置的字段使用这些默认值。
参数：
- insert-field或者字段名称到表达式的映射
- 返回类型：InsertQuery
创建新用户的例子：
q = User.insert(username='admin', active=True, registration_expired=False) q.execute() # perform the insert.
你也可以使用Field对象作为键：
User.insert(**{User.username: 'admin'}).execute()
如果你有一个拥有字段默认值的模型，如果插入参数没有指名它，默认值将被使用：
class User(Model): username = CharField() active = BooleanField(default=True) # This INSERT query will automatically specify `active=True`: User.insert(username='charlie')
当插入查询在有自增主键的表上执行时，新行的主键将返回。
insert_many(rows)
一次插入多行。rows参数必须是一个字典的可迭代对象，和insert相同，字典中没有声明的字段使用默认值。
由于批量插入的原生性，每行应该有相同字段。下面的例子不行：
Person.insert_many([ {'first_name': 'Peewee', 'last_name': 'Herman'}, {'first_name': 'Huey'}, # Missing "last_name"! ])
- 参数：rows -包含键值对的可迭代对象
- 返回类型：InsertQuery
使用批量插入的例子：
usernames = ['charlie', 'huey', 'peewee', 'mickey'] row_dicts = ({'username': username} for username in usernames) # Insert 4 new rows. User.insert_many(row_dicts).execute()
因为rows参数可以为任意的迭代对象，你也可以使用生成器：
def get_usernames(): for username in ['charlie', 'huey', 'peewee']: yield {'username': username} User.insert_many(get_usernames()).execute()
classmethod insert_from(fields,query)
使用查询作为数据源插入数据。该API使用INSERT INTO ...SELECT FROM 查询。
参数：
- fields-将选择的数据映射到字段对象
- query-新行的查询
返回类型：InsertQuery
非正常的跨表插入数据：
source = (User
.select(User.username, fn.COUNT(Tweet.id))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username))
UserTweetDenorm.insert_from(
[UserTweetDenorm.username, UserTweetDenorm.num_tweets],
source).execute()
classmethod delete()
返回类型：DeleteQuery
删除非活跃用户的例子：
q = User.delete().where(User.active == False) q.execute() # remove the rows
这个方法在整表上删除，删除一个实例使用Model.delete_instance()
classmethod raw(sql,*params)
参数：sql -SQL表达式
参数-操作时插入的参数
选择所有用户例子：
q = User.raw('select id, username from users') for user in q: print user.id, user.username
raw作为保留使用，是为了显著优化查询的场景使用。因为它返回模型的实例所以select插询很有用。
classmethod create(**attrubutes)
参数：
- attributes-模型属性的键值对
返回类型 拥有提供属性的模型实例
用户创建的例子（新行增加到数据库）：
user = User.create(username='admin', password='test')
create方法是实例化对象然后保存的简单快捷方式
classmethod get(*args)
- 参数：args-插询表达式的列表
- 返回类型：Model实例或者DoesNotExist错误
得到匹配的单行。若没有行返回，则抛出错误DoesNotExist
user = User.get(User.username == username, User.active == True)
这个方法在SelectQuery上也可以使用，虽然没有接受参数：
active = User.select().where(User.active == True) try: user = active.where( (User.username == username) & (User.active == True) ).get() except User.DoesNotExist: user = None
get方法是选择单一记录的简单方式。当没有记录匹配加上抛出错误行为。如果多行返回，数据库游标的首行返回。
classsmethod get_or_create([deaults=None[,**kwargs]])
- 参数：defaults-在新增对象实例上的默认值，是一个字典
- kwargs：Diango风格的过滤器指名get的模型，以及赋给新实例的属性值
返回值： 一个二元元组包含模型实例和boolean值标明是否对象是创建的。
这个函数试图根据提供的过滤器取得模型实例。如果没有匹配的结果，新模型根据默认值和过滤器中的参数被创建。
当调用get_or_create 加上autocommit=False时小心，因为这个方法调用Database.atomic()来创建对象，而不是使用事务或保存节点。
没有使用get_or_create的例子：
# Without `get_or_create`, we might write:
try:
person = Person.get(
(Person.first_name == 'John') &
(Person.last_name == 'Lennon'))
except Person.DoesNotExist:
person = Person.create(
first_name='John',
last_name='Lennon',
birthday=datetime.date(1940, 10, 9))
等效于使用get_or_create：
person, created = Person.get_or_create( first_name='John', last_name='Lennon', defaults={'birthday': datetime.date(1940, 10, 9)})
classmethod create_or_get([**kwargs])
参数：kwargs-字段名称和值来尝试创建新对象。
返回值：一个二元元组包含一个对象实例以及Boolean值标明对象是否被创建
这个函数试图创建对象实例根据提供的关键字参数。如果IntegrityError发生，标明约束冲突，Peewee返回匹配过滤器的模型。
Peewee将不尝试匹配所有的关键字参数来遇到IntegrityError发生。只有主键的字段或者拥有唯一约束的字段来取到对象实例。
当使用autocommit=False来create_or_get时小心，因为这个方法将调用Database.atomic()来创建对象，而不是事务或保存节点。
例子：
# This will succeed, there is no user named 'charlie' currently. charlie, created = User.create_or_get(username='charlie') # This will return the above object, since an IntegrityError occurs # when trying to create an object using "charlie's" primary key. user2, created = User.create_or_get(username='foo', id=charlie.id) assert user2.username == 'charlie'
classmethod alias()
返回类型 ModelAlias实例
alias方法来创建自关联查询：
例子：
Parent = Category.alias() sq = (Category .select(Category, Parent) .join(Parent, on=(Category.parent == Parent.id)) .where(Parent.name == 'parent category'))
当使用ModelAlias时，你必须显示声明关联条件。
classmethod create_table([fail_silently=False])
参数：fail_silently（Boolean）-设置为True，将检查表名是否已存在。
创建根据模型的表，同样创建模型中定义的约束和索引：
例子：
database.connect() SomeModel.create_table() # Execute the create table query.
classmethod drop_table([fair_silently=False,[,cascase=False]])
参数：fail_silently（Boolean）将检查表是否存在
cascade-CASCADE选项情况下删除表
删除模型对应的表
classmethod table_exists（）
返回类型：标明模型对应的表是否数据库中已经存在
classmethod sqlall()
返回值：返回需要创建表和索引的一系列查询
save([force_insert=False,[,only=None]])
参数：
- force_insert(bool)-是否强制插入
- only(list)-需要持久化的字段-当提供此参数时，只有提供的字段被持久化。
保存给定的实例，当创建或者更新根据它是否有主键。如果force_insert=True INSERT将忽略是否主键存在而执行插入。
保存模型实例的例子：
user = User() user.username = 'some-user' # does not touch the database user.save() # change is persisted to the db
delete_instance([recursive=False,[,delete_nullable=False]])
参数：
- recursive-删除实例和依赖它的别的实例，选择性参数更新那些nullable的依赖
- delete_nullable-如果递归删除，不论依赖的对象是否可以被更新为NULL，都删除它们
删除给定的实例。任何已删除对象为外键的对象也被自动删除。更详细的程序控制，你可以用recursive=True，任何相关联的non-nullable模型也被删除（相关联的nullable设置为NULL）。
如果你想要同时删除nullable模型，设置delete_nullable=True。
例子：
some_obj.delete_instance() # it is gone forever
dependencies([search_nullable=False])
参数：
search_nullable(bool)-通过nullable外键搜索关联的模型
返回类型：生成查询和外键字段的生成器表达式
生成依赖模型的查询列表。生成二元元组，其包含查询饿相关联的外键字段。对于查询模型的依赖有用，例如当删除事件发生时，需要做些别的操作。
dirty_fields
返回手动设置的字段列表
返回类型：list
如果你只是想要持久化修改的字段，你可以调用model.save(only=model.dirty_fields)
如果你想要只是保存模型的dirty_fields，你可以用Meta设置only_save_dirty=True。这样，每次你调用Model.save(),默认只是dirty fields才会保存：
class Person(Model):
first_name = CharField()
last_name = CharField()
dob = DateField()
class Meta:
database = db
only_save_dirty = True
is_dirty()
返回是否任何字段已经手动修改
返回类型：bool
prepared()
这个方法提供了钩子，行数据生成后执行模型初始化
Fields
Field(null=False,index=False,unique=False,verbose_name=None,help_text=None, db_column=None, default=None, choices=None, primary_key=False, sequence=None, constraints=None, schema=None, **kwargs)
参数：
- null(bool)- 是否这个列接受None或者NULL值
- index(bool)-是否在这个列创建索引
- unique(bool)-是否在此列上创建唯一索引
- verbose_name(string)-表示这个字段的verbose名字,元数据目的有用
- help_text(string)-表示这个字段的帮助描述文字
- db_column(string)-存储时映射到的字段名字,对与数据库兼容性有用.
- default-未初始化是的默认值
- choices-一个二元数组的可迭代对象,映射值本身到显示的值
- primary_key(bool)-是否为表的主键
- sequence(string)-序列的名字(如果向后数据库支持)
- constraints(list)约束的数组例如[check('price>0')]
- schema(string)-schema的名字(如果向后数据库支持)
- kwargs-包含可能用来标明特殊字段子类型的命名属性,比如"max_length或者decimal_places"
db_field='<some field type'>'
映射这个字段到的列类型,例如:string或者DateTimeField
_is_bound
标明这个字段绑定到模型类的Boolean标志
model_class
对应的模型类.只在绑定的字段上有用
name
字段的名称.只在绑定的字段上有用
db_value(value)
参数:value-对应数据库存储的python数据类型
返回类型:python数据类型转换以后的数据库类型
python_value(value)
参数:value-从后端存储的数据
返回类型:python数据类型
coerce(value)
用于db_value和python_value的快捷方法.你可以尽管实现
参数:value-app或者数据库任意的类型
返回类型:python数据类型
class IntegerField
存储:integers
db_field = 'int'
class BigIntegerField
存储:大整型数
db_field = 'bigint'
class PrimaryKeyField
存储:作为主键的合适的自增整型字段
db_field = 'primary_key'
class FloatField
存储:浮点数字
db_field = 'float'
class DoubleField
存储:双精度浮点数字
db_field = 'double'
class DecimalField
存储:数字,使用python标准库中的Decimal对象
额外的属性和值:
max_digits |
10 |
decimal_places |
5 |
auto_round |
False |
rounding |
decimal.DefaultContext.rounding |
db_field = 'decimal'
class CharField
存储:小的字符串(0-255字节)
额外的熟悉和值:
max_length |
255 |
db_field = 'string'
class TextField
存储:任意大字符串
db_field = 'text'
class DateTimeField
存储:python datetime.datetime实例
接收特殊的参数formats，包含时间格式列表。默认的格式是：
'%Y-%m-%d %H:%M:%S.%f' # year-month-day hour-minute-second.microsecond '%Y-%m-%d %H:%M:%S' # year-month-day hour-minute-second '%Y-%m-%d' # year-month-day
注意
如果拿来的值不匹配格式，它将返回原来的样子
db_field='datetime'
year
提取年份的表达式，例如：查询2013年以来的所有博客：
Blog.select().where(Blog.pub_date.year == 2013)
month
从存储的日期里提取月。
day
从存储日期里提取天。
hour
从存储日期里提取小时
minute
从存储日期里提取分钟
second
从存储日期里提取秒
class DateField
存储：python datetime.date实例
接受参数formats，包含格式日期的格式列表。默认格式是：
'%Y-%m-%d' # year-month-day '%Y-%m-%d %H:%M:%S' # year-month-day hour-minute-second '%Y-%m-%d %H:%M:%S.%f' # year-month-day hour-minute-second.microsecond
注意：
如果来的值不匹配一个格式，返回原来的样子
db_field='date'
year
从日期里提取年的表达式，例如取得1980年出生的人：
Person.select().where(Person.dob.year == 1983)
month
提取月
day
提取天
class TimeField
存储：python datetime.time实例
接受formats参数，包含格式列表。默认值有：
'%H:%M:%S.%f' # hour:minute:second.microsecond '%H:%M:%S' # hour:minute:second '%H:%M' # hour:minute '%Y-%m-%d %H:%M:%S.%f' # year-month-day hour-minute-second.microsecond '%Y-%m-%d %H:%M:%S' # year-month-day hour-minute-second
注意：
如果来的值没有匹配的格式，返回原值
db_field=‘time'
hour
从时间里提取小时.例如得到发生在晚上的事件:
Event.select().where(Event.time.hour > 17)
minute
提取分钟
second
提取秒
class BooleanField
存储:True/False
db_field='bool'
class BlobField
存储任意的二进制数据
class UUIDField
存储UUID值
注意:
此字段只支持PostgresqlDatabase.
class BareField
在SQLite中使用.因为数据类型不强制,你可以声明没有任何数据类型的字段.对于SQLite虚拟表来说使用元列和未名类型列来说很普遍,所以你可使用未名类型字段.
接受coerce参数,一个函数,用来将数据库类型和Python数据类型相互转换.
注意:
当前此字段只支持SQLiteDatabase
class ForeignKeyField(rel_model,[ralated_name=None[,on_delete=None[,on_update=None,[,to_field=None[,..]]]]])
存储:到另一个模型的关系
参数:
- rel_model-关联的Model类或者self字符来表示自引用的外键
- related_name(string)-对应关联模型的属性
- on_delete(string)-当删除时的行为,例如:on_delete='CASCADE'
on_update(string)-当更新时的行为
to_field-rel_model外键引用的字段.默认主键字段当作rel_model
class User(Model):
name = CharField()
class Tweet(Model):
user = ForeignKeyField(User, related_name='tweets')
content = TextField()
# "user" attribute
>>> some_tweet.user
<User: charlie>
# "tweets" related name attribute
>>> for tweet in charlie.tweets:
... print tweet.content
Some tweet
Another tweet
Yet another tweet
外键没有特殊的db_field因为他们的字段类型取决于他们关联的模型主键的字段类型
如果你手动的标示to_field，这个字段必须是主键或者有唯一约束
class CompositeKey(*fields)
表示模型的复合主键.不像其它字段,复合主键在所有字段定义完成后定义在Meta 类中.它接受用作主键的字段名字的字符列表作参数:
class BlogTagThrough(Model):
blog = ForeignKeyField(Blog, related_name='tags')
tag = ForeignKeyField(Tag, related_name='blogs')
class Meta:
primary_key = CompositeKey('blog', 'tag')
Query Types
class Query
是所有其它查询类的父类.你可能不直接在你的代码里处理Query,所有的查询类型实现了一些方法.
where(*expressions)
参数:表达式-一个或多个表达式
返回类型:查询实例
选择用户名为somebody的用户:
sq = SelectQuery(User).where(User.username == 'somebody')
选择编辑和管理员的tweets
sq = SelectQuery(Tweet).join(User).where(
(User.is_editor == True) |
(User.is_admin == True))
删除不活跃用户的tweets:
dq = DeleteQuery(Tweet).where(
Tweet.user << User.select().where(User.active == False))
dq.execute() # perform the delete query
where调用是链式的,多次调用使用AND连接
join(model,join_type=None,on=None)
参数:
- model-关联的模型,必须有ForeignKeyField来关联查询上下文和关联的模型.
- join_type-显式指名关联类型:JOIN.INNER,JOIN.LEFT_OUTER,JOIN.FULL
- on-如果两个模型间有多个外键,指名使用哪个外键来关联
返回类型:查询实例
查询上下文和join的参数来生成JOIN子句,建立新的查询上下文.
例如:选择admin发表的tweets和所属users:
sq = SelectQuery(Tweet).join(User).where(User.is_admin == True)
把特殊的外键字段当做关联选择users.请看example app实例使用:
sq = SelectQuery(User).join(Relationship, on=Relationship.to_user)
switch(model)
参数:model-切换查询上下文到模型
返回类型:返回新的查询上下文的查询
切换查询上下文到指定的model,如果model没有选择或者在前面关联会剖出异常.通常在单表执行多次连接时有用.
下面查询选择blog同时连接到entry和user:
sq = SelectQuery(Blog).join(Entry).switch(Blog).join(User)
alias(alias=None)
参数:alias(str) 查询结果的昵称
返回类型:查询实例
sql()
返回类型:包含合适的ＳＱＬ查询和参数元组的二元元组
execute()
执行查询
scalar([as_tuple=False])
参数:as_tuple(bool)-把行当作元组或单值返回
返回类型:结果行,一个元组或单值
提供了从选择查询里得到单值的方式,例如执行聚合:
>>> PageView.select(fn.Count(fn.Distinct(PageView.url))).scalar() 100 # <-- there are 100 distinct URLs in the pageview table
class SelectQuery(model_class,*selection)
目前为止peewee里最复杂的查询类.它支持select语句中所有子句.
SelectQuery上方法可链式调用
SelectQuery实现了__iter__()方法,允许迭代返回结果实例
参数:
- model-执行查询的模型
- selection-模型,字段,函数或表达式的列表
如果没有selection提供,默认给的模型的所有字段都返回
例如选择user实例,只有id和username被选择.当迭代结果时,返回User模型实例:
sq = SelectQuery(User, User.id, User.username)
for user in sq:
print user.username
例如选择users和每个人的tweets数量,User实例将拥有额外的count属性,对应发表的tweets数量:
sq = (SelectQuery(
User, User, fn.Count(Tweet.id).alias('count'))
.join(Tweet)
.group_by(User))
select(*selection)
参数:selection-表达式列表,可以是模型类,字段,若果不写默认给定模型的所有字段
返回类型:SelectQuery
通常实例创建后将表明selection .这个方法之所以存在,是为了在查询上修改select子句.
query = User.select() query = query.select(User.userna
from_(*args)
参数:args-一个或多个表达式,可以是Model或SelectQuery实例.如果为空,默认给定模型的表
返回类型:SelectQuery
# rather than a join, select from both tables and join with where. query = User.select().from_(User, Blog).where(Blog.user == User.id)
group_by(*clauses)
参数:clauses-表达式列表,可是模型类或单独的字段实例
返回类型:SelectQuery
以单列或多列分组.如果模型类当参数,模型类所有字段将被当作分组字段.
例如:选择users关联tweets,并且以user分组,所以每个人的tweets可以被计算
sq = (User
.select(User, fn.Count(Tweet.id).alias('count'))
.join(Tweet)
.group_by(User))
having(*expressions)
参数:expressions-一个或多个表达式列表
返回类型:SelectQuery
同上例:限制发表100以上tweets的用户:
sq = (User
.select(User, fn.Count(Tweet.id).alias('count'))
.join(Tweet)
.group_by(User)
.having(fn.Count(Tweet.id) > 100))
order_by(*clauses)
参数:clauses-字段的列表,可以调用字段的field.[asc|desc]()
返回类型:SelectQuery
根据名字排序用户
User.select().order_by(User.username)
根据user名字排序,然后根据发表时间排序选择tweets
query = (Tweet
.select()
.join(User)
.order_by(
User.username,
Tweet.created_date.desc()))
也可以使用+或-前缀来标明升降序:
query = (Tweet
.select()
.join(User)
.order_by(
+User.username,
-Tweet.created_date))
复杂例子:先根据tweets发表数量,然后用户名字排序:
tweet_ct = fn.Count(Tweet.id)
sq = (User
.select(User, tweet_ct.alias('count'))
.join(Tweet)
.group_by(User)
.order_by(tweet_ct.desc(), User.username))
window(*windows)
参数:windows(Window)-一个或多个window定义.
在查询上添加一个或若干个window定义
limit(num)
参数rs:num(int)-限制结果返回num个行
offset(num)
参数:num(int) -结果跳过num行
paginate(page-num,paginate_by=20)
参数：page_num-从1开始的页数，用来将返回结果分页
paginate_by-返回结果的每页记录数量
返回类型：SelectQuery
paginate是在查询应用LIMIT和OFFSET的简单方式
页数切片是从1开始的，所以第一页是page 1
distinct([is_distinct=True])
参数:is_distinct-下面将说明
返回类型:SelectQuery
表明查询结果返回不同的行.相当于SELECT DISTINCT 查询
注意:
is_distinct是个Boolean值,而不是DISTINCT
也可以用一个或多个表达式来生成DISTINCT ON查询.例如:.distinct([Model.col1,Model.col2])
for_update([for_update=True,[,nowait=False]])
表明查询更新行时加锁.如果nowait设为True,数据库如果不能获得锁将抛出OperationalError
naive()
返回类型:SelectQuery
标志查询仅仅视图为返回的每行重建单一模型.若多表参与参与查询,返回的列直接分发到单个模型实例上.
通常这个方法对于给的数据库游标来说加快其重建模型实例的速度.
注意:
当在大的数据结果集中做简单的迭代时可以得到显著的速度提升.
iterator()
返回类型:iterable
默认peewee将游标返回的数据行都缓存,这样方便了多个迭代,切片,索引而不进行额外查询.
但是当迭代大量的行时,这种缓存占据大量内存.使用iterator()可以不存储所有的返回的模型实例从而节约内存.
# iterate over large number of rows.
for obj in Stats.select().iterator():
# do something.
pass
tuples()
返回类型:SelectQuery
将游标查询结果只返回原生的元组数组.当你不想要模型实例而是简单的结果时使用此方法.
dicts()
返回类型:SelectQuery
将游标结果只返回字典数组.当你只需要简单的查询结果时使用此方法.
aggregate_rows()
返回类型:SelectQuery
此方法提供了避免N+1查询难题的一种方式
比如一个页面需要展示一系列用户及他们相关的tweets.你可以通过列出所有用户然后为每一个用户执行单独的查询找到他的tweets,这就是N+1行为,因为查询次数取决于查询出用户的数量.明智的方法将尽量少执行查询.Peewee提供了若干方法来解决此问题.
你可以使用prefetch()帮助器,它使用IN查询子句来得到用户的tweets.
另一种方法是在一个查询里查出所有用户和tweet行,然后再根据用户的不同来聚合tweets/
原来查出的数据是这样子的:
# user.id, user.username, tweet.id, tweet.user_id, tweet.message [1, 'charlie', 1, 1, 'hello'], [1, 'charlie', 2, 1, 'goodbye'], [2, 'no-tweets', NULL, NULL, NULL], [3, 'huey', 3, 3, 'meow'], [3, 'huey', 4, 3, 'purr'], [3, 'huey', 5, 3, 'hiss'],
我们可以推断出JOIN子句中用户数据将重复，因此我们减少重复用户数据后，可以循环的收集用户的tweets，然后透明的迭代用户和他们的tweets
query = (User
.select(User, Tweet)
.join(Tweet, JOIN.LEFT_OUTER)
.order_by(User.username, Tweet.id)
.aggregate_rows()) # .aggregate_rows() tells peewee to de-dupe the rows.
for user in query:
print user.username
for tweet in user.tweets:
print ' ', tweet.message
# Producing the following output:
charlie
hello
goodbye
huey
meow
purr
hiss
no-tweets
注意：
确保ORDER BY子句可以保证重复的数据出现在相关联的行中
你也可以用作任意复杂的查询，但是使用prefetch来执行复杂的查询更有效。可以两种方法尝试一下看你的数据集适合哪种方式
annotate(related_model,aggregation=None)
参数:related_model-执行聚合的相关联的模型, 必须通过ForeignKeyField连接
aggregation-聚合使用的类型:比如fn.Count(Tweet.id).alias('count')
返回类型:SelectQuery
通过在相关联的模型上进行聚合来注释查询,例子:得到用户和他们每个人的tweets数量
>>> User.select().annotate(Tweet)
如果 aggregate设为None,默认行为为fn.Count(related_model.id).alias('count'),但是也可以为其它:
user_latest = User.select().annotate(Tweet, fn.Max(Tweet.created_date).alias('latest'))
如果ForeignKeyField为nullable,LEFT OUTER 连接需要使用在join中:
query = (User
.select()
.join(Tweet, JOIN.LEFT_OUTER)
.switch(User) # Switch query context back to `User`.
.annotate(Tweet))
aggregate(aggregate)
参数:aggregation-表明聚合函数,例如:fn.Max(Tweet.created_date).
用聚合行的函数返回数量值,比如行数,或者特殊列的平均值.
count([clear_limit=False])
参数:clear_limit(bool)-在count前移除查询里limit和offset子句
返回类型:查询返回结果的行数
如果查询里有GROUP BY,DISTINCT,LIMIT或者OFFSET子句时，wrapped_count()方法将使用.
>>> sq = SelectQuery(Tweet) >>> sq.count() 45 # number of tweets >>> deleted_tweets = sq.where(Tweet.status == DELETED) >>> deleted_tweets.count() 3 # number of tweets that are marked as deleted
wrapped_count([clear_limit=True])
参数:clear_limit(bool)-count前移除查询里的limit和offset子句.
返回类型:查询结果行数
在子查询中包装count查询,当执行DISTINCT查询或带有GROUP BY子句时返回正确的结果
注意
count()将在遇到DISTINCT或GROUP BY自动使用wrapped_count()
exists()
返回类型:返回当前查询是否返回有行的Boolean值,当优化查询时,代替get()方法
sq = User.select().where(User.active == True)
if sq.where(User.username == username, User.active == True).exists():
authenticated = True
get()
返回类型:模型实例或抛出DoesNotExist意外
从数据库里得到满足查询条件的单行,如果没有返回行抛出<model-class>.DoesNotExist意外
active = User.select().where(User.active == True)
try:
user = active.where(User.username == username).get()
except User.DoesNotExist:
user = None
此方法也通过Model API暴露,此情况下接收的参数翻译为where子句:
user=User.get(User.active==True,User.username==username)
first()
返回类型:模型实例或None
得到查询结果首记录,结果会缓存以免将来会迭代整个查询结果集
execute()
返回类型:QueryResultWrapper
执行查询,返回QueryResultWrapper来迭代结果集.查询结果由查询内部管理,当可能改变
结果集的子句加进来时,查询会重新执行.
__iter__()
执行查询,返回模型实例迭代对象:
__len__()
返回查询结果集中的行数量,若你只是想要数量,而不要结果集本身,可调用count()方法
注意:
SELECT查询被执行,结果集被加载.
__getitem__(value)
参数:value-索引或者切片对象
返回模型实例结果集的切片.例如:得到首个记录:
__or__(rhs)
参数:rhs-SelectQuery或CompoundSelect
返回类型:CompoundSelect
使用符号左右边对象创建联合查询,结果同时包含左右边的查询结果.
customers = Customer.select(Customer.city).where(Customer.state == 'KS')
stores = Store.select(Store.city).where(Store.state == 'KS')
# Get all cities in kansas where we have either a customer or a store.
all_cities = (customers | stores).order_by(SQL('city'))
SQLite不允许ORDER BY或LIMIT查询子句在复合查询的组件上,但是允许在最终结果集中使用这些子句.同样对UNION(ALL),INTERSECT(),EXCEPT也是这样.
__and__(rhs)
参数:rhs-SelectQuery或CompoundSelect
返回类型:CompoundSelect
创建INTERSECT查询.结果集包含符号左右边的结果的交集.
customers = Customer.select(Customer.city).where(Customer.state == 'KS')
stores = Store.select(Store.city).where(Store.state == 'KS')
# Get all cities in kanasas where we have both customers and stores.
cities = (customers & stores).order_by(SQL('city'))
__sub__(rbs)
参数:rhs-SelectQuery或CompoundSelect
返回类型:CompoundSelect
创建EXCEPT查询.结果返回符号包含在左边查询但是不包含在右边查询的结果.
customers = Customer.select(Customer.city).where(Customer.state == 'KS')
stores = Store.select(Store.city).where(Store.state == 'KS')
# Get all cities in kanasas where we have customers but no stores.
cities = (customers - stores).order_by(SQL('city'))
__xor__(rhs)
参数:SelectQuery或CompoundSelect
返回类型:CompoundSelect
创建一个对称差分查询.结果包含或者在左边查询里的数据或者在右边查询里的数据,但没有同时在两边查询里的数据.
customers = Customer.select(Customer.city).where(Customer.state == 'KS')
stores = Store.select(Store.city).where(Store.state == 'KS')
# Get all cities in kanasas where we have either customers with no
# store, or a store with no customers.
cities = (customers ^ stores).order_by(SQL('city'))
class UpdateQuery(model_class,**kwargs)
参数:
- model-更新的模型
- kwargs-字段值对映射,用来更新列值
例子:用户注册过期将用户标识为非活跃用户:
uq = UpdateQuery(User, active=False).where(User.registration_expired == True) uq.execute() # Perform the actual update
例子:原子性更新:
atomic_update = UpdateQuery(PageCount, count = PageCount.count + 1).where(
PageCount.url == url)
atomic_update.execute() # will perform the actual update
execute()
返回类型:更新的行数
执行更新
returning(*returning)
参数:RETURNING-模型类,字段实例,函数或表达式的列表.如果没有参数,给定模型的所有列都
将被选择.清空所有存在的值,传参数None
返回类型:给定模型的UpdateQuery
在查询上添加RETURNING子句,会使UPDATE在实际更新的每行上计算值当作返回.
当更新执行时,不是返回受影响的行数,而是返回迭代器用来生成更新的对象.
当前只有PostgresqlDatabase支持此特性.
例子:
# Disable all users whose registration expired, and return the user
# objects that were updated.
query = (User
.update(active=False)
.where(User.registration_expired == True)
.returning(User))
# We can iterate over the users that were updated.
for updated_user in query.execute():
send_activation_email(updated_user.email)
tuples()
返回类型:UpdateQuery
此方法结合returning()调用来使用
更新结果以行元组返回
dicts()
返回类型:UpdateQuery
此方法结合returning()来使用
更新结果以行字典返回.每个字典映射列字段到值.
on_conflict([action=None])
加上SQL ON CONFLICT子句给UPDATE查询，当conflict发生时会发生指定的动作。有效的动作有：
- ROLLBACK
- ABORT
- FAIL
- IGNORE
- REPLACE
指定None时时执行正常的UPDATE查询
注意：
这个特性只支持SQLite 数据库
class InsertQuery(model_class,[field_dict=None,[rows=None[,fields=None,[,query=None[,validate_fields=False]]]]])
创建模型的InsertQuery实例.
参数:
- field_dict(dict)-字段到值的映射.
- rows(iterable)-可迭代的字典,其为字段到值的映射
- fields(list)-插入值到的字段列表(仅仅与query参数联合使用)
- query-用作SelectQuery的数据源
- validate_fields(bool)-检查插入查询中每一列在模型上都有对应的字段.如果validation开启,但是验证失败,抛出KeyError异常.
基本使用例子:
>>> fields = {'username': 'admin', 'password': 'test', 'active': True}
>>> iq = InsertQuery(User, fields)
>>> iq.execute() # insert new row and return primary key
2L
插入多行:
users = [
{'username': 'charlie', 'active': True},
{'username': 'peewee', 'active': False},
{'username': 'huey', 'active': True}]
iq = InsertQuery(User, rows=users)
iq.execute()
从查询数据源插入:
query = (User
.select(User.username, fn.COUNT(Tweet.id))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User.username))
iq = InsertQuery(
UserTweetDenorm,
fields=[UserTweetDenorm.username, UserTweetDenorm.num_tweets],
query=query)
iq.execute()
execute()
返回类型:新行的主键
执行查询
upsert([upsert=True])
在SQLite上执行INSERT 或REPLACE查询.MySQL数据库是欧阳REPLACE查询.当前特性在Postgresql数据库上不能用,但是9.5版本语法将添加此功能.
注意
此特性只在SQLite和MySQL上有效.
on_conflict([action=None])
在INSERT查询上添加ON CONFLICT子句.指名为REPLACE等价于使用upsert方法.有效的动作有:
- ROLLBACK
- ABORT
- FAIL
- IGNORE
- REPLACE
当为None时表示动作为正常的INSERT查询.
此特性只支持SQLite数据库
return_id_list([return_id_list=True])
默认,批量插入时peewee不会返回插入的主键列表.但是如果数据库支持通过INSERT...RETURNING来返回主键列表,此方法让peewee来返回生成的IDs
注意:
当前只有Postgresql支持此行为.别的数据库支持批量插入,简单返回True
usernames = [
{'username': username}
for username in ['charlie', 'huey', 'mickey']]
query = User.insert_many(usernames).return_id_list()
user_ids = query.execute()
print user_ids
# prints something like [1, 2, 3]
returning(*returning)
参数:returning-模型类,字段实例,函数或表达式列表.如果无参数,模型的所有列都返回,传递None当参数清除任何已存在的值.
返回类型:模型的InsertQuery
给查询添加RETURNING子句,会在新插入的行中计算返回的值.
当执行查询时,而不是返回插入的行的主键,将返回生成插入对象的迭代器.
当前此特性仅仅支持PostgresqlDatabase
# Create some users, retrieving the list of IDs assigned to them.
query = (User
.insert_many(list_of_user_data)
.returning(User))
# We can iterate over the users that were created.
for new_user in query.execute():
# Do something with the new user's ID...
do_something(new_user.id)
tuples()
返回类型:InsertQuery
此方法只支持与returning()方法联合使用
把插入对象的结果以行元组的形式返回
dicts()
返回类型:InsertQuery
此方法只支持与returning()方法联合使用
把插入对象结果以字典列表的形式返回,其中字典是列到值的映射.
class DeleteQuery（model_class)
创建给定模型的DELETE查询
注意：
DeleteQuery不会在外键上删除或者不能保证遵守约束，所以小心使用
使用DeleteQuery删除非活跃用户：
dq = DeleteQuery(User).where(User.active == False)
execute()
返回类型:删除的行数
执行查询
returning(*returning)
参数:returning- 模型类,字段实例,函数和表达式的列表,如果没有提供参数.给定模型的所有列都将返回,清除所有已存在的值,传递None
返回类型:给定模型的DeleteQuery
给查询加上RETURNING子句,引起DELETE计算从数据库中成功删除的行的返回值.
当查询执行时,不是返回删除的行数,而是返回生成删除对象的迭代器.
注意:
当前只有PostgresqlDatabase支持此特性.
例子:
# Create some users, retrieving the list of IDs assigned to them.
query = (User
.delete()
.where(User.account_expired == True)
.returning(User))
# We can iterate over the user objects that were deleted.
for deleted_user in query.execute():
# Do something with the deleted user.
notify_account_deleted(deleted_user.email)
tuples()
返回类型:DeleteQuery
注意:
此方法与returning()一起调用
删除的结果返回以行元组列表返回.
dict()
返回类型:
DeleteQuery
此方法与returning()一起使用.
删除的结果以映射列到值的字典列表形式返回
class RawQuery(model_class,sql,*params)
通过QueryResultsWrapper来执行任意的查询,返回查询的模型实例
注意:
通常使用此方法来执行高度优化的查询
如果你执行参数化的查询,比如根据你的数据库使用正确的填充字符串.SQLite使用
'?',其它使用'%s'.
根据用户名字选择用户:
>>> rq = RawQuery(User, 'SELECT * FROM users WHERE username = ?', 'admin') >>> for obj in rq.execute(): ... print obj <User: admin>
tuples()
返回类型:RawQuery
标志查询是否从游标返回原生元组列表.当你不需要全部的模型实例时此方法很有用.
dicts()
返回类型:RawQuery
标志查询是否从游标返回原生字典列表.当你不需要全部的模型实例时此方法很有用.
execute()
返回类型:
迭代结果集的QueryResultWrapper.结果集是给定模型的模型实例列表
执行查询
class CompoundSelect(model_class,lhs,operator,rhs)
组合选择查询.
参数:
- model_class-返回模型的类型,默认是操作符左边模型类型
- lhs-左边查询,SelectQuery或CompoundQuery
- operator-关联两个查询的Node实例,例如SQL('UNION')
- rhs-右边查询,SelectQuery或CompoundQuery.
prefetch(sq,*subqueries)
参数:
- sq-SelectQuery实例
- subqueries-一个或多个SelectQuery实例来取得sq.你也可以传递模型实例,但是它们被转换为SelectQuery.如果你想要指名与某个模型关联,你可以传递(query_or_model,join_model)这样的二元元组
返回类型:带有预生成相关联实例的SelectQuery
从子查询中预先得到相应的实例数据,然后在外层查询中赋给它们的父行.这个函数在子查询中饥饿加载相关联的实例.这是解决O(n)查询问题的技巧,而不是执行k次子查询.
例如.你有用户列表,想要显示他们的tweets:
# let's impost some small restrictions on our queries
users = User.select().where(User.active == True)
tweets = Tweet.select().where(Tweet.published == True)
# this will perform 2 queries
users_pf = prefetch(users, tweets)
# now we can:
for user in users_pf:
print user.username
for tweet in user.tweets_prefetch:
print '- ', tweet.content
你可以预先得到任意数量的结果行.例如,我们有图片网站,User->Photo->(Comments,Tags).用户可以发表照片,照片有标签和评论.如果我们想要得到一系列用户,以及他们各自的照片,以及照片相关的评论和标签:
users = User.select()
published_photos = Photo.select().where(Photo.published == True)
published_comments = Comment.select().where(
(Comment.is_spam == False) &
(Comment.num_flags < 3))
# note that we are just passing the Tag model -- it will be converted
# to a query automatically
users_pf = prefetch(users, published_photos, published_comments, Tag)
# now we can iterate users, photos, and comments/tags
for user in users_pf:
for photo in user.photo_set_prefetch:
for comment in photo.comment_set_prefetch:
# ...
for tag in photo.tag_set_prefetch:
# ...
注意:
子查询必须通过外键关联,并且可以为任意深度.
关于更详细的信息.可以看<<避免N+1查询部分>>章节
若结果集很大prefetch()会消耗大量内存,并且你在做危险的事情时不会给你警告提示,所以取决于你的应用.另外,由于子查询的语义不明,可能有时候你预得到的数据不是你想要的(例如,如果你在子查询中使用LIMIT子句,可能应用到别的模型上)-如果你认为有bug的地方请向github提交报告.
Database 和子类
class Database(database[,threadlocals=True[,autocommit=True,[,fields=None,[,opt=None,[,autorollback=False,[,use_speedups=True,[**connect_kwargs]]]]]]])
参数：
- database-数据库名称(或者SQLite数据库的文件名)
- threadlocals(bool)是否在threadlocal中存储连接
- autocommit(bool)-通过调用execute()自动提交每一个查询
- fields(dict)-db_field到数据库列类型的映射,eg:'string'=>'varchar'
- ops(dict)-查询编译器理解的操作符到表达式的映射
- autorollback(bool)-当执行查询发生错误时是否自动回滚
- use_speedups(bool)-使用Cython加速查询性能
- connect_kwargs-任意的参数用在数据库连接上
- connect_kwargs字典用在特定的参数,该参数向后直接传递到数据库引擎上,让你可以指名user,host,password.
注意:
如果当类定义时数据库名称未知,你可传递None当作数据库名称,然后数据库将为deferred,此时
任何尝试连接数据库将抛出错误.调用Database.init()来初始化数据库.
关于运行时数据库配置的深度学习,请看运行时数据库配置章节.
与支持的数据库引擎的层次的API.数据库类:
- 管理潜在的数据库连接
- 执行查询
- 管理事务和保存节点
- 创建表和索引,删除表和索引
- 检查数据库
commit_select=False
在选择查询时是否执行提交,可以在有些数据库引擎上避免隐式堆积的事务
compiler_class=QueryCompiler
编译查询的类
Compound_operations=['UNION','INTERSECT,'EXCEPT']
支持的符合查询操作
Compound_select_parentheses=False
是否UNION或其他复合的SELECT查询在查询上加上括号
distinct_on=False
是否数据库支持DISTINCT ON语句
drop_cascade=False
是否数据库支持级联地删除表查询
field_overrides={}
字段类型到数据库列类型的映射.例如{'primary_key':'SERIAL'}
foreign_keys=True
是否给定的向后强制外键约束
for_update=False
是否给定的向后支持选择行更新
for_update_nowait=False
是否给定的向后支持选择行更新
insert_many=True
是否数据库支持在INSERT查询中多个VALUES子句
insert_returning=False
是否数据库支持新增加的行返回主键
interpolation='?'
数据库内插查询参数的字符串
op_overrides={}
操作code和字符操作的映射{OP.LIKE:'LIKE BINARY'}
quote_char='"'
数据库引擎应用名字的字符串
reversed_tables=[]
向后保留的表名字-如果在应用中遇到则会出现警告
returning_clause=False
是否数据库支持在UPDATE,INSERT和DELETE查询上的RERURNING子句
当前只支持PostgresqlDatabase数据库.
看下面信息:
- UpdateQuery.returning()
- InsertQuery.returning()
- DeleteQuery.returning()
savepoints=True
是否给定向后支持保存节点.
sequences=False
是否给定向后支持序列
subquery_delete_same_table=True
是否向后支持使用从表中的子查询来删除行
window_functions=False
是否给定向后支持window函数.
init(database[,**connect_kwargs])
此方法用来初始化deferred数据库.关于在运行时配置数据库,请看运行时数据库配置章节
参数:
- database-数据库名称(或sqlite的文件名)
- connect_kwargs-当连接数据库引擎时的任意参数
connect()
建立到数据库的连接
注意:
默认情况下,连接保存在threadlocal中,保证连接不被跨线程分享.为了取消此行为,可以初始化数据库时设threadlocals=False.
close()
关闭数据库连接
注意:
若你初始化时设置threadlocals=True,仅仅本地对调用的thread连接关闭.
initialize_connection(conn)
在新打开连接时得初始化其他.例如,如果你正使用SQLite想要开启外键强制约束(默认关闭)
下面是如何用这个钩子来加载SQLite扩展
class CustomSqliteDatabase(SqliteDatabase):
def initialize_connection(self, conn):
conn.load_extension('fts5')
get_conn()
返回类型:数据库的连接,如果不存在创建一个连接
get_cursor()
返回类型:执行查询的游标
last_insert_id(cursor,model)
- cursor-用于执行插入查询的数据库游标
- model-刚刚创建的模型类
返回类型-最近插入的实例的主键
rows_affected(cursor)
返回类型:最后查询影响的行数
compiler()
返回类型:使用重载过的字段和操作的QueryCompiler实例
execute(clause)
参数:clause(Node)-Node实例或子类(例如:SelectQuery)
子句被编译成SQL然后发给execute_sql()方法
execute_sql(sql[,params=None[,require_commit=True]])
参数:
- sql-sql查询字符串
- params-用来填充的参数的列表或元组
注意:
你可以用set_autocommit()和Database.get_autocommit()配置查询是否自动提交.
begin()
开启新的事务.默认没有实现,因为不是DB_API 2.0的一部分,但是为了API兼容性提供了.
commit()
在当前活动连接上调用commit(),提交当前事务
rollback()
在当前活动连接上调用rollback(),回滚当前事务.
set_autocommit(autocommit)
参数:
autocommit-是否开启自动提交的Boolean值
get_autocommit()
返回类型:是否自动提交开启的Boolean值
get_tables([schema=None])
返回类型:数据库中的表名列表
get_indexes(table,[,schema=None])
返回类型:IndexMetadata实例列表,代表给定表的索引
get_columns(table[,schema=None])
返回类型:ColumnMetadata实例列表,表示给定表的列
get_primary_keys(table[,schema=None])
返回类型:给定表的主键列名称列表
get_foreign_keys(table,[,schema=None])
返回类型:ForeignKeyMetadata实例列表,代表给定表的外键
sequence_exists(sequence_name)
返回类型:Boolean值,表示序列是否存在
create_table(model_class[,safe=True])
参数
- model_class-模型类
- safe(bool)-为True时,表将不创建如果已经存在.
注意:
不像Model.create_table(),此方法不创建索引和约束.这个方法只是创建表本身.如果你想要同时创建索引和约束,使用Model.createtable或Database.create_tables().
create_index(model_class,fields[,unique=False])
参数:
- model_class -将要创建索引的模型表
- fields-创建索引的字段(或字段实例或字段名称)
- unique-索引是否唯一
create_foreign_key(model_class,field[,constraint=None])
参数:
- model_class-在哪个Model表上创建外键约束
- field-Field对象
- constraint(str)-外键约束名称
使用ALTER TABLE来手动的创建外键约束.当创建循环外键依赖时非常有用.例子:
DeferredPost = DeferredRelation()
class User(Model):
username = CharField()
favorite_post = ForeignKeyField(DeferredPost, null=True)
class Post(Model):
title = CharField()
author = ForeignKeyField(User, related_name='posts')
DeferredPost.set_model(Post)
# Create tables. The foreign key from Post -> User will be created
# automatically, but the foreign key from User -> Post must be added
# manually.
User.create_table()
Post.create_table()
# Manually add the foreign key constraint on `User`, since we could
# not add it until we had created the `Post` table.
db.create_foreign_key(User, User.favorite_post)
create_sequence(sequence_name)
参数:sequence_name-将要创建的序列名称
注意:
仅仅支持有序列支持的数据库引擎
drop_table(model_class[,fail_silently=False[,cascade=False]])
参数:
- model_class-要删除的表的模型
- fair_silently(bool)-如果为True,查询将增加一条IF EXISTS子句
- cascade(bool)-选项CASCADE的删除表
drop_sequence(sequence_name)
参数:
sequence_name-删除的序列的名字
注意:
仅仅支持有序列支持支持的数据库引擎
create_tables(model[,safe=False])
参数:
- model(lsit)-模型列表
- safe(bool)-检查是否表已存在,然后再去创建表
此方法应该在用作创建表,因为它会解析模型的依赖关系,所以确保表的创建顺序.此方法也创建模型上的索引和约束.
用法:
db.create_tables([User, Tweet, Something], safe=True)
drop_tables(models[,safe=False[,cascade=False]])
参数:
- models(list)-模型列表
- safe(bool)-检查表是否存在在试图删除前
- cascade(bool)-选项CASCASE删除表
db.drop_tables([User, Tweet, Something], safe=True)
这个方法用于删除表,因为它将解析表之间的依赖关系,确保按照正确的顺序删除.
atomic()
在事务或保存节点中执行语句.外层的调用将使用事务,任何嵌套的内层调用使用保存节点.
atomic可以用作上下文管理器或者装饰器.
注意:
大多数情况下,当你想要在事务中查询时总是使用atomic()是好办法.使用atomic的方便之处是你不必手动跟踪事务栈的深度,它将为你管理事务栈.
用作上下文管理器代码:
with db.atomic() as txn:
perform_some_operations()
with db.atomic() as nested_txn:
do_other_things()
if something_bad_happened():
# Roll back these changes, but preserve the changes
# made in the outer block.
nested_txn.rollback()
用作装饰器代码:
@db.atomic()
def create_user(username):
# This function will execute in a transaction/savepoint.
return User.create(username=username)
transaction()
在事务中执行语句,可用作上下文管理器或装饰器.如果错误抛出在包装的语句块中,事务回滚,否则语句在完成后提交.事务语句块中的事务通过调用rollback()或commit()可以显式的回滚或提交.如果你手动的回滚,一个新事务将自动开始.
用transaction可以包装语句块-数据库将保持栈状态,仅仅语句块到达最外层的函数或块结构时才提交.
用作上下文管理器的例子:
# delete a blog instance and all its associated entries, but
# do so within a transaction
with database.transaction():
blog.delete_instance(recursive=True)
# Explicitly roll back a transaction.
with database.transaction() as txn:
do_some_stuff()
if something_bad_happened():
# Roll back any changes made within this block.
txn.rollback()
用作装饰器的例子:
@database.transaction()
def transfer_money(from_acct, to_acct, amt):
from_acct.charge(amt)
to_acct.pay(amt)
return amt
commit_on_success(func)
注意:
用atomic()或者transaction()代替
savepoint([sid=None])
使用上下文管理器或者装饰器来执行保存节点中的语句.如果在包装的语句块中出现异常,保存节点将回滚,否则语句提交.和transaction()不同,保存节点可以显式的回滚和提交,分别调用rollback()和commit().如果你手动的提交或回滚,新的保存节点不会创建.
保存节点可以认为是嵌套的事务.
参数:sid(str)-选择性的字符串来代表这个保存节点.
用作上下文管理器的例子:
with db.transaction() as txn:
do_some_stuff()
with db.savepoint() as sp1:
do_more_things()
with db.savepoint() as sp2:
even_more()
# Oops, something bad happened, roll back
# just the changes made in this block.
if something_bad_happened():
sp2.rollback()
execution_context([with_transaction=True])
创建一个ExecutionContext上下文管理器或装饰器.用ExecutionContext包装的语句将使用各自的连接.默认情况下,包装的语句在一个事务里执行,可以通过指定参数with_transaction=False来关闭此功能.
关于ExecutionContext的详细解释,请看<<高级连接管理>>章节.
注意:
ExecutionContext是新功能,没有测试完全.
classmethod register_fields(fields)
为数据类注册字段映射.用于注册自定义字段或者重载默认字段映射.
参数:
- fields(dict)-db_field到列类型的映射
classmethod register_ops(ops)
注册能被QueryComplier理解的操作符到等价的SQL语句的映射.例如:{OP.EQ:'='}.用于扩展字段的比较类型.
extract_date(date_part,date_field)
返回从日期字段中抽取日期一部分的表达式.例如,从DateTimeField里提取年份.
参数:
- date_part(str)-想得到的日期属性.有效的选项有:'year','month','day','hour','minute'以及'second'
- date_field(Field)-存储datetime,date或time的字段实例
返回类型:表达式对象
truncate_date(date_part,date_field)
返回截短date/datetime到指定的方案的合适的表达式.例如,用于将timestamps按天来分组.
参数:
- date_part(str)-截短到的日期部分,有效的选项有:'year','month','day','hour','minute','second'
- date_field(Field)-存储datetime,date或time的字段实例
返回类型:表达式对象.
例子:
# Get tweets from today.
tweets = Tweet.select().where(
db.truncate_date('day', Tweet.timestamp) == datetime.date.today())
class SqliteDatabase(Database)
Database 与sqlite3驱动或pysqlite2.除了默认的数据库参数,SqliteDatabase可以接收journal_mode参数来配置journal模式
注意:
如果你在系统同时安装了sqllite3和pysqlite2,peewee将使用指向SQLite最新版本的那个模块.
SQLite是Peewee唯一支持的可在宿主应用中高度可定制的数据库.意味着你可以用Python写自己的函数和聚合,然后在SQL查询中调用.这个特性可以通过SqliteExtDatabase来获得学习.强烈推荐你使用SqliteExtDatabase,它暴露了使得SQLite更强大的许多特性.
自定义参数:
参数:
- journal_mode(str)-日志模式
- pragmas(list)-包含PGRAMA语句的二元元组列表,以便处理与新连接的竞争
SQLite允许通过PRAGMA语句来在运行时配置许多参数.PRAGMA用来与新连接到来时处理情形.在新连接到来时运行一个或多个PRAGMA语句,你可以将它们以包含PRAGMA名字和值的二元元组列表的形式指名:
db = SqliteDatabase('my_app.db', pragmas=(
('journal_mode', 'WAL'),
('cache_size', 10000),
('mmap_size', 1024 * 1024 * 32),
))
insert_many=True 如果使用SQLite 3.7.11.0或更高版本
class MySQLDatabase(Database)
Database 与MySQLdb或pymysql一起工作的子类.
commit_select=True
compound_operations=['UNION']
for_update=True
subquery_delete_same_table=False
class PostgresqlDatabase(Database)
Database 子类与psycopg2驱动器一起工作
commit_select=True
Compound_select_parentheses=True
distinct_on=True
for_update=True
for_update_nowait=True
insert_returning=True
returning_clause=True
sequences=True
window_functions=True
register_unicode=True
控制是否UNICODE和UNICODEARRAY psycopg2扩展模块自动载入
(Transaction,Savepoint和ExecutionContext事务,保存节点和执行上下文
最简单使用事务和保存节点的方式是使用Database.atomic().atomic()方法根据嵌套的层次创建transaction或者savepoint.
with db.atomic() as txn:
# The outer-most call will be a transaction.
with db.atomic() as sp:
# Nested calls will be savepoints instead.
execute_some_statements()
class transaction(database)
上下文管理器封装了数据库事务.在包装以后语句块中的语句将在没有意外抛出时在最后提交,否则事务回滚.
注意:
事务应该不嵌套,以免在嵌套的语句块中发生意外时导致未知行为.如果你想要嵌套事务,使用atomic()方法,会在最外层创建事务,嵌套中使用保存节点.
实际中你最好不用transaction对象,而是使用Database.transaction()方法.
commit()
手动提交挂起的更改,然后开始新事务
rollback()
手动回滚挂起的更改,然后开始新的事务.
class savepoint(database,[sid=None])
上下文管理器封装了保存节点(嵌套事务).在包装的语句中在没有意外抛出情况下最后提交事务,否则最后回滚事务.
注意:
保存节点必须在一个事务中使用,所以推荐使用使用atomic()方法来代替手动管理事务+保存节点的上下文栈.
实际中,最好不直接创建savepoint,而采用Database.savepoint()方法.
commit()
手动提交任何挂起的更改.如果保存节点手动提交并且有改变,它们将在最外层的上下文中执行.
rollback()
手动回滚任何挂起的更改,如果保存节点手动回滚并且有更改,它们将在最外层上下文中执行.
class ExecutionContext(database[,with_transaction=True])
ExecutionContext显式提供了专用的连接来运行SQL.典型地每个线程使用单一的数据库连接,但是如果你想要强制性的使用新数据库连接.这时你可以使用ExecutionContext.在包装的语句块中默认语句在一个事务中运行,你可以通过with_transaction=False来不用事务.
注意:
而不是直接使用ExecutionContext,使用Database.execution_context().
例子:
conn = db.get_conn()
with db.execution_context():
# This will be a new connection object. If you are using the
# connection pool, it may be an unused connection from the pool.
ctx_conn = db.get_conn()
# This statement is executed using the new `ctx_conn`.
User.create(username='huey')
# At the end of the wrapped block, the connection will be closed and the
# transaction, if one exists, will be committed.
# This statement is executed using the regular `conn`.
User.create(username='mickey')
class Using(database,models,[,with_transaction=True])
在包装语句期间,所有models的查询都可用这里指定的数据库.通过指定with_transaction=True,所有的查询可以在事务里运行.
Using提供了,在一系列模型上运行查询,但是使用手动指定的数据库.
参数:
- database-Database实例
- models-使用此数据库的一系列模型类
- with_transaction-是否包装的语句在事务中运行
Metadata 类型
class IndexMetadata(name,sql,columns,unique,table)
name
索引名字
sql
SQL查询去生成索引
columns
索引包含的列
unique
表示索引是否为唯一约束
table
包含索引的表名
class ColumnMetadata(name,data_ype,null,primary_key,table)
name
列名
data_type
列类型
null
NULL是否允许
primary_key
是否列为主键
table
包含列的表名
class ForeignKeyMetadata(column,dest_table,dest_column,table)
column
包含外键的列
dest_table
引用此外键的表
dest_column
引用此外键的列
table
包含此外键的表名
Misc
class fn
帮助器转换任意的函数调用到SQL函数调用
peewee中使用表达式函数，使用fn对象。它'.'操作符右边的事物被当作函数对待.你可以
你可以将函数传递任意参数,参数也可以为其它有效的表达式.
例如:
| Peewee expression | Equivalent SQL |
|---|---|
fn.Count(Tweet.id).alias('count') |
Count(t1."id") AS count |
fn.Lower(fn.Substr(User.username, 1, 1)) |
Lower(Substr(t1."username", 1, 1)) |
fn.Rand().alias('random') |
Rand() AS random |
fn.Stddev(Employee.salary).alias('sdv') |
Stddev(t1."salary") AS sdv |
over([partition_by=None,[,order_by=None[,window=None]]])
基本的SQL窗口函数.
参数:
- partition_by(list)-分区依据的Node实例列表
- order_by(list)-用来排序的Node实例列表
- window(Window)-聚合的Window实例
例子:
# Get the list of employees and the average salary for their dept.
query = (Employee
.select(
Employee.name,
Employee.department,
Employee.salary,
fn.Avg(Employee.salary).over(
partition_by=[Employee.department]))
.order_by(Employee.name))
# Rank employees by salary.
query = (Employee
.select(
Employee.name,
Employee.salary,
fn.rank().over(
order_by=[Employee.salary])))
# Get a list of page-views, along with avg pageviews for that day.
query = (PageView
.select(
PageView.url,
PageView.timestamp,
fn.Count(PageView.id).over(
partition_by=[fn.date_trunc(
'day', PageView.timestamp)]))
.order_by(PageView.timestamp))
# Same as above but using a window class.
window = Window(partition_by=[fn.date_trunc('day', PageView.timestamp)])
query = (PageView
.select(
PageView.url,
PageView.timestamp,
fn.Count(PageView.id).over(window=window))
.window(window) # Need to include our Window here.
.order_by(PageView.timestamp))
class SQL(sql,*params)
在peewee查询上添加SQL片段.例如,你需要引用一个别名的查询.
参数:
- sql(str)-任意的SQL字符串
- params-任意的查询参数
# Retrieve user table and "annotate" it with a count of tweets for each
# user.
query = (User
.select(User, fn.Count(Tweet.id).alias('ct'))
.join(Tweet, JOIN.LEFT_OUTER)
.group_by(User))
# Sort the users by number of tweets.
query = query.order_by(SQL('ct DESC'))
class Window([partition_by=None[,order_by=None]])
创建WINDOW定义
参数:
- partition_by(list)-分区的Node实例列表
- order_by(list)-用来排序的Node实例列表
例子:
# Get the list of employees and the average salary for their dept.
window = Window(partition_by=[Employee.department]).alias('dept_w')
query = (Employee
.select(
Employee.name,
Employee.department,
Employee.salary,
fn.Avg(Employee.salary).over(window))
.window(window)
.order_by(Employee.name))
class DeferredRelation
用来引用尚未创建的模型类.作为外键关联的模型的别名.在循环引用有用.
DeferredPost = DeferredRelation()
class User(Model):
username = CharField()
# `Post` is not available yet, it is declared below.
favorite_post = ForeignKeyField(DeferredPost, null=True)
class Post(Model):
# `Post` comes after `User` since it refers to `User`.
user = ForeignKeyField(User)
title = CharField()
DeferredPost.set_model(Post) # Post is now available.
set_model(model)
用合适的模型类代替占位符
class Proxy
Proxy类在当你想要推迟对象的初始化时非常有用,例如,你想要定义你的模型,但是你在运行上线前还未确定数据库引擎.
例子:
database_proxy = Proxy() # Create a proxy for our db.
class BaseModel(Model):
class Meta:
database = database_proxy # Use proxy for our DB.
class User(BaseModel):
username = CharField()
# Based on configuration, use a different database.
if app.config['DEBUG']:
database = SqliteDatabase('local.db')
elif app.config['TESTING']:
database = SqliteDatabase(':memory:')
else:
database = PostgresqlDatabase('mega_production_db')
# Configure our proxy to use the db we specified in config.
database_proxy.initialize(database)
initialize(obj)
参数:代理的对象
一旦初始化,obj上的属性和方法可以直接通过Proxy实例来访问.
class Node
Node类是查询的所有组件的父类,形成了peewee的基本表达式API的基础,下面的类就都继承自Node:
SelectQuery,UpdateQuery,InsertQuery,DeleteQuery, andRawQuery.FieldFunc(andfn())SQLExpressionParamWindowClauseEntityCheck
重载的操作符:
- 位操作符 and -and和or-(& 和 |):使用当作连接符连接多个node
- +,-,*,/和^ (加,减,乘,除和异或)
- ==,!=,<,<=,>,>=:使用它们创建一个二元表达式.
- <<:创建IN表达式
- >>:创建IS表达式．
- % 和** :LIKE和ILIKE
contains(rhs)
使用大小写不敏感的字符串查询创建二元表达式
startswith(rhs)
使用大小写不敏感的前缀搜索创建二元表达式
endswith(rhs)
使用大小写不敏感的后缀搜索创建二元表达式
between(low,high)
创建匹配满足low和high之间的表达式
regexp(expression)
基于正则表达式匹配
concat(rhs)
将当前node和rhs连接.
is_null([is_null=True])
创建表达式来测试Node是否为NULL
# Find all categories whose parent column is NULL. root_nodes = Category.select().where(Category.parent.is_null()) # Find all categores whose parent is NOT NULL. child_nodes = Category.select().where(Category.parent.is_null(False))
为了简化,peewee生成对应的正确的SQL.is_null()方法提供阅读简单性.
__invert__()
取node的相反.这将粗略的转成NOT(<node>)
alias([name=None])
将node定义别名.这转换为<node> AS <name>
asc()
应用升序到node上,转换为<node> ASC.
desc()
应用降序到node上,转换为<node> DESC
bind_to(model_class)
将表达式的结果绑定到指定的模型类型上.当表达式的结果放在指定的关联实例上,需要给select添加表达式时非常有用.
classmethod extend([name=None,[,clone=False]])
在Node和它的子类上添加装饰函数作为新方法,当自己在所有的node类型上实现了特定的新特性时非常有用.
参数:
- name(str)-方法名称,若没有则包装的函数名称被使用
- clone(bool)-此方法是否返回克隆版本.当方法改变node的内部状态时通常设为true
例子:
# Add a `cast()` method to all nodes using the '::' operator.
PostgresqlDatabase.register_ops({'::', '::'})
@Node.extend()
def cast(self, as_type):
return Expression(self, '::', SQL(as_type))
# Let's pretend we want to find all data points whose numbers
# are palindromes. Note that we can use the new *cast* method
# on both fields and with the `fn` helper:
reverse_val = fn.REVERSE(DataModel.value.cast('str')).cast('int')
query = (DataPoint
.select()
.where(DataPoint.value == reverse_val))
调动类的delattr方法把原来定义的方法作为参数移除扩展的方法.