peewee使用

前言

• 安装（版本：3.15.1）

 pip install peewee 

• 轻量化的ORM工具

建表

引用旧表

# 指定mysql，用户为root，host为localhost，数据库为user
python -m pwiz -e mysql -u root -H localhost --password user > UserModel.py

新建

from playhouse.shortcuts import model_to_dict
from peewee import MySQLDatabase, Model# 连接数据库
db = MySQLDatabase("user", user="root", host="0.0.0.0", port="3306", password="root")class BaseModel(Model):class Meta:database = dbdef to_dict(self) -> dict:"""转化为字典"""return model_to_dict(self)class UserModel(BaseModel):class Meta:table_name = "user"					# 用户表username = CharField(unique=True)		# 账号password = CharField()					# 密码isFrozen = BooleanField(default=False)	# 是否被冻结createTime = DateTimeField()			# 创建时间class UserInfoModel(BaseModel):class Meta:table_name = "user_info"		# 用户信息表user = ForeignKeyField(UserModel)	# 用户表关联外键nickname = CharField()				# 昵称email = CharField()					# 邮箱phone = CharField()					# 手机号class UserPermModel(BaseModel):class Meta:table_name = "user_perm"			# 用户权限表user = ForeignKeyField(UserModel)		# 用户表关联外键create = BooleanField(default=False)	# 增delete = BooleanField(default=False)	# 删update = BooleanField(default=False)	# 改search = BooleanField(default=False)	# 查if __name__ == '__main__':pass

操作

事物

# 装饰器方式
# @db.atomic()with database.atomic() as transaction:pass

增

user, info, perm = new_user["user"], new_user["info"], new_user["perm"]
user = UserModel(**user)
uid = user.save()
UserInfoModel.create(user=user, **info).save()
UserPermModel.create(user=user, **perm).save()

删

UserModel.get(UserModel.username == username
).delete_instance(# 同时删除外键关联表中内容recursive=True, delete_nullable=True
)

改

UserModel.update(info).where(UserModel.username == username).execute()

查

单表

query_data = UserModel.select(# 不填内容即所有字段# UserModel.username
).where(# 筛选条件UserModel.isFrozen == False
).order_by(# 升序：asc()，降序：desc()UserModel.createTime.asc()
)
res = [row.to_dict() for row in query_data]
print(query_data)
print(res)

多表

query_data = UserModel.select(UserModel, UserInfoModel, UserPermModel
).join_from(UserModel, UserInfoModel
).join_from(UserModel, UserPermModel
).order_by(UserModel.id.asc()
).dicts()
res = [row for row in query_data]
print(query_data)
print(res)# 或者
query_data = UserModel.select(UserModel, UserInfoModel, UserPermModel
).join(UserInfoModel
).switch(UserModel
).join(UserPermModel
).order_by(UserModel.id.asc()
).dicts()
res = [row for row in query_data]
print(query_data)
print(res)