Peewee学习笔记

peewee是一个ORM数据库处理的库，我翻译了从快速入门开始的核心部分，后面还有API文档和扩展库的文档没有翻译。

2021年3月22日完成了翻译，中间也间断的学习了一部分。给我的感受是peewee正是我要找的那种可以像操作类的属性一样调用数据的库。在此之前，我也自学了一个叫TinyDB的库，它不用数据库文件，而是使用JSON文件保存数据。感觉还是peewee更专业一点。

后面的内容，预计花1个月的时间，将这个复杂的缩减到一个文档中。这就是所谓的书从厚读薄吧。

摘要：peewee 核心

Database
Model --> 创建表
Field -->表的字段
Query --> 查询

使用的顺序：

初始化数据库
创建表
增、删、改、查

一、初始化数据库

在执行下面所有命令之前，我们需要初始化一个数据库（Database）、表（Model）和它的字段（field）。以下是官方文档的标准范式，抄就完了，别问为什么？

from peewee import *

# db = SqliteDatabase(':memory:')  <-- 在内存里面搞一个数据库
db = SqliteDatabase('my_database.db') # <--在当前目录下生成一个my_database.db文件

class BaseModel(Model): # 标准写法
    class Meta:         # 下面的Model全部继承BaseModel
        database = db   # 连接数据库，Meta子类还可以设置其他的东西，这里别写

class User(BaseModel):
    name = CharField()
    message = TextField()

db.connect() # 生成/连接数据库
db.close()   # 关闭 数据库连接

二、创建表

在使用数据库之前，需要用已经声明的Model来在数据库里面创建表：

db.connect() # 没有必要，但是还是显示的建立连接会比较方便后面判断错误。
db.create_tables([User])

print(User.__dict__.keys())

测试下，没有问题。

PS：数据迁移问题

使用playhouse.migrate模块可以进行迁移（在已经存在的表里增加一个字段）

from playhouse.migrate import *
migrator = SqliteMigrator(db)

price_field = IntegerField(default=0)
migrate(migrator.add_column('product', 'price', price_field))

这里需要注意了，你增加了字段，那表的Model就要跟着改。如果你非要在同一个py文件中这么操作，peewee作者的答案是：

It's unusual to do this online:

1.declare a model/schema

2.migrate it

3.use the new schema

Because if you know you need a price field just put it on the model class from the start.

If you really need to do this, then you can call this after you've run the migration:
Product._meta.add_field('price', price_field)

数据库my_database.db文件里面实际上已经有了price这个字段，但是在同一个py文件中是无法调用的，为什么，因为我们定义的Model里面没有，所以，要修改前面的Model。

直接在Model里面写个字段

使用 pwiz 生成新文件

python -m pwiz -e sqlite path/to/sqlite_database.db

三、增删改查

1.创建或插入新记录

类方法
- Model.create()
- Model.insert().execute()
- 通过类方法不需要使用 .save()保存
实例属性
- user = User(name = 'xxx')
- user.name = 'xxx'
- 通过实例属性都要使用 .save()保存

1.1 `create()`

创建一条记录，并返回一个实例

user = User.create(name = 'watalo', message= '11111')
print(user)

1.2 `insert()`

插入一条记录，返回主键

User.insert(name = 'donggua', message = 'wangwangwang').execute()

1.3 传参的实例

user = User(name = 'jianhetao', message = 'houhouhou')
user.save()
# query = User.select().where(User.name == 'jianhetao')
# for q in query:
#     print(q.message)

1.4 不传参的实例

user = User()
user.name = 'jiajia' # 提示 NOT NULL constraint failed: user.message
user.message = '555555555555555' # 如果不写这条
user.save()

1.5 `insert_many()`

不管什么情况都使用db.atomic()会极大的提高运行速度。

数据源可以用两种类型：

字典列表：[{‘name’:‘ xxx’, 'message':'jjj'},]

data_source = [
    {'name':'aaa', 'message':'jjj'},
    {'name':'bbb', 'message':'jjj'},
    {'name':'ccc', 'message':'jjj'},
    {'name':'ddd', 'message':'jjj'},
    {'name':'eee', 'message':'jjj'},
    {'name':'fff', 'message':'jjj'},
    {'name':'ggg', 'message':'jjj'},
    ] # 字典列表形式的数据源
with db.atomic():
    User.insert_many(data_source).execute()

元组列表：[('xxx','jjj')]

data_source = [
    ('hhh', 'jjj'),
    ('hhh', 'jjj'),
    ('hhh', 'jjj'),
    ] # 元组列表形式的数据源
with db.atomic():
    # 这里需要制定fields
    User.insert_many(data_source, fields=[User.name, User.message]).execute()

数据源的量特别大的时候，SQLite会有批处理的数据量的上限，通常是999或这32776，这种情况使用分块会比较好：

chunked(): 把大量数据分成固定数量的块，一个一个的循环

from datetime import datetime
data_source = []
for i in range(40000):
    data_cell = {'name':'aaa'+str(i), 'message':'message'+str(i*2)}
    data_source.append(data_cell)

print(datetime.now())
from peewee import chunked
with db.atomic():
    for batch in chunked(data_source, 100):
        User.insert_many(batch).execute()
print(datetime.now())

用时：0.7089990000000004

还可以直接写个循环：

with db.atomic():
    for idx in range(0, len(data_source), 100):
        User.insert_many(data_source[idx:idx+100]).execute()

用时：0.7069879999999955

从用时上看，这两种方法都差不多。

1.6 `insert_from()`

从其他的表里面直接导入数据。没有试。

User_back.insert_from(User.select(User.name),fields=[User_back.name,]).execute()

1.7 其他选项

bulk_create() 没有试。

with db.atomic():
    User.bulk_create(users, batch_size=100)

参数batch_size

bulk_update() 没有试。

with database.atomic():
    User.bulk_update(list_of_users, fields=[`username`], batch_size=50)

2.删除记录

2.1 `delete_instance()`

删除一个实例，在实例后调用。

watalo = User.get(User.name == 'watalo')
watalo.delete_instance()

2.2 `delete()`

在类后面使用

hhh_to_be_delete = User.delete().where(User.name == 'hhh') 
hhh_to_be_delete.execute()

3.更新记录

3.1 修改单条记录

一旦实例有了主键，再进行save()操作就是 update，而不是 insert。

aaa = User.select().where(User.name == 'aaa').get()
aaa.message = '我改了'
aaa.save()

3.2 更新多条记录

使用update查询，直接进行批量更新

query = User.update(message = '').where(User.name.startswith('aaa'))
query.execute()

3.3 原子更新

atomic update 这个单词不知道具体是什么意思，感觉就是对某个很细微的数据进行更新。有三个案例可以学习下：

简单计数

query = Stat.update(counter=Stat.counter + 1).where(Stat.url == request.url)
query.execute()

复杂查询

query = Employee.update(bonus=(Employee.bonus + (Employee.salary * .1)))
query.execute()  # Give everyone a bonus!

使用子查询

subquery = Tweet.select(fn.COUNT(Tweet.id)).where(Tweet.user == User.id)
update = User.update(num_tweets=subquery)
update.execute()

3.4 插入

replace()

3.24之前的SQLite和MySQL可以使用。作用是直接替换某条记录，而且可以忽略限制条件。这里就不讨论了

class User(Model):
    username = TextField(unique=True)
    last_login = DateTimeField(null=True)

# Insert or update the user. The "last_login" value will be updated
# regardless of whether the user existed previously.
user_id = (User
           .replace(username=`the-user`, last_login=datetime.now())
           .execute())

on_conflict_replace()

用这个命令可以忽略限制条件。

user_id = (User
           .insert(username=`the-user`, last_login=datetime.now())
           .on_conflict_replace()
           .execute())

4.查询记录

4.1 查询单条记录

7种调用方式：

get() 如果没查到会跳出 DoesNotExist异常
```
User.get(User.id == 6)
```
get_by_id()
```
User.get_by_id(7)
```
User[8]
User.get(User.id == 9).name
User.get(User.name == 'aaa')
User.select().where(User.name == 'bbb').get()
get_or_none() 如果没有查到会直接返回None
```
User.select().where(User.name == 'bbb').get()
```

get_or_create()如果没有就创建一条记录，并且返回一个元组(实例，布尔值)。

User.get_or_create(name = 'wocalie',defaults = {'message':'successed'})
# (<User: 120017>, True)

这个函数的逻辑如下：

try:
    with db.atomic():
        return User.create(username=username)
except peewee.IntegrityError:
    # `username` is a unique column, so this username already exists,
    # making it safe to call .get().
    return User.get(User.username == username)

4.2 查询多条记录

4.2.1 `select()`: 查询所有

query = User.select()
print(query)

select()结果可以进行迭代：

User.select().iterator()
```
query = User.select()

for row in query.iterator():
    print(row)
```
在调用大型数据的时候，会节省内存。

还可以字典化、元组化，进一步加快速度。而且这些都可以进行iterator()。

User.select().dicts()

query = User.select().dicts().iterator()

User.select().tuple()

query = User.select().tuples().iterator()

User.select().nametuple()

query = User.select().namedtuples().iterator()

在做联表查询的大型迭代时，peewee实际上重构了model，把数据写入model，再读取。这样很慢，所以peewee提供了一个方法：

User.select().objects()

query = User.select()

for q in query.objects():
    print(q.name)

4.3 筛选查询记录

4.3.1 `where()`方法

User.select().where()

对单一Model的条件查询

User.select().where(User.name == 'bbb')

User.select().join(Tweet).where()

对联表Mode的条件查询

User.select().join(Tweet).where(Tweet.is_pulished = True)

4.3.2 条件设置

复杂查询时，可采用以下方式：

圆括号和python的按位或和和操作符

(User.username == `Charlie`) | (User.username == `Peewee Herman`)

函数表达式

fn.Lower(fn.Substr(User.username, 1, 1)) == a

比较表达式

Employee.salary < (Employee.tenure * 1000) + 40000

嵌套查询：in_()

a_users = User.select().where(fn.Lower(fn.Substr(User.username, 1, 1)) == `a`)
a_user_tweets = Tweet.select().where(Tweet.user.in_(a_users))

4.4 排序`order_by()`

基本用法
- 正序(+)：不用调整
- 倒序(-)：Tweet.created_date.desc() =- Tweet.created_date

query = Tweet.select().join(Member).where(Tweet.is_published).order_by(Tweet.created_date.desc())

query = Tweet.select().join(Member).where(Tweet.is_published).order_by(- Tweet.created_date)

高级用法

ntweets = fn.COUNT(Tweet.id)
query = (User
         .select(User.username, ntweets.alias(`num_tweets`))
         .join(Tweet, JOIN.LEFT_OUTER)
         .group_by(User.username)
         .order_by(ntweets.desc())

4.5 分页`paginate()`

for tweet in Tweet.select().paginate(5, 10):
    print(tweet.id)

4.6 随机取值`Random()`

for tweet in Tweet.select().order_by(fn.Random()).limit(5):
    print(tweet.id)

4.7 计数`count()`

带条件

Tweet.select().where(Tweet.id > 40).count()

不带条件

Tweet.select().count()

4.8 聚合`group_by()`

我理解的所谓聚合，是各种查询后打包成一个新的数据，然后再来利用，不知道对不对。

举例：查一个计数的结果，然后打包到Member里面：

query = (Member
         .select(Member, fn.Count(Tweet.id).alias('count'))
         .join(Tweet, JOIN.LEFT_OUTER)
         .group_by(Member))
for q in query:
    print(q.id, q.count)

[peewee]笔记

Peewee学习笔记

摘要：peewee 核心

一、初始化数据库

二、创建表

PS：数据迁移问题

三、增删改查

1.创建或插入新记录

1.1 `create()`

1.2 `insert()`

1.3 传参的实例

1.4 不传参的实例

1.5 `insert_many()`

1.6 `insert_from()`

1.7 其他选项

2.删除记录

2.1 `delete_instance()`

2.2 `delete()`

3.更新记录

3.1 修改单条记录

3.2 更新多条记录

3.3 原子更新

3.4 插入

4.查询记录

4.1 查询单条记录

4.2 查询多条记录

4.2.1 `select()`: 查询所有

4.3 筛选查询记录

4.3.1 `where()`方法

4.3.2 条件设置

4.4 排序`order_by()`

4.5 分页`paginate()`

4.6 随机取值`Random()`

4.7 计数`count()`

4.8 聚合`group_by()`

推荐阅读

[peewee]笔记

Peewee学习笔记

摘要：peewee 核心

一、初始化数据库

二、创建表

PS：数据迁移问题

三、增删改查

1.创建或插入新记录

1.1 create()

1.2 insert()

1.3 传参的实例

1.4 不传参的实例

1.5 insert_many()

1.6 insert_from()

1.7 其他选项

2.删除记录

2.1 delete_instance()

2.2 delete()

3.更新记录

3.1 修改单条记录

3.2 更新多条记录

3.3 原子更新

3.4 插入

4.查询记录

4.1 查询单条记录

4.2 查询多条记录

4.2.1 select(): 查询所有

4.3 筛选查询记录

4.3.1 where()方法

4.3.2 条件设置

4.4 排序order_by()

4.5 分页paginate()

4.6 随机取值Random()

4.7 计数count()

4.8 聚合group_by()

推荐阅读

1.1 `create()`

1.2 `insert()`

1.5 `insert_many()`

1.6 `insert_from()`

2.1 `delete_instance()`

2.2 `delete()`

4.2.1 `select()`: 查询所有

4.3.1 `where()`方法

4.4 排序`order_by()`

4.5 分页`paginate()`

4.6 随机取值`Random()`

4.7 计数`count()`

4.8 聚合`group_by()`