介绍

有时候需要用 Python 直接操作数据库, 手工撸 SQL 倒也不是不能行, 但是总觉得有点背离了 人生苦短, 我用 Python 的初心, 习惯了 SQLAlchemy 的直接操作后, 就变懒了. 但是对于比较简单的数据库操作场景, 再跑一套 SQLAlchemy 又显得有些臃肿, 这次上手轻量 Python ORM 选手 peewee

上 Demo

# 安装依赖
pip install peewee pymysql faker

from peewee import *
from datetime import datetime
from faker import Faker# 定义数据库连接
db = MySQLDatabase(database="peewee_db", user="root", password="lpwm86", host="debian.lan", port=3306
)# 定义基础模型
class BaseModel(Model):# Peewee 会自动创建一个自动编号的主键 id, 所以没啥特别场景可以不用定义class Meta:database = db# 定义用户模型
class User(BaseModel):login = CharField(unique=True)password = CharField()name = CharField(unique=True)join_date = DateField()# 定义角色模型
class Role(BaseModel):name = CharField()users = ManyToManyField(User, backref="roles")  # backref 后面的名字可以被关联的对象通过 user.roles 方式访问# 用户和角色多对多的映射表
UserRole = Role.users.get_through_model()# 定义文章模型
class Post(BaseModel):title = CharField()content = TextField()author = ForeignKeyField(User)# 连接数据库
db.connect()
print("Db connected.")# 创建数据表
db.create_tables([User, Role, Post, UserRole])
print("Table created.")db.execute_sql("set foreign_key_checks=0")
table: BaseModel
for table in [User, Role, Post, UserRole]:table.truncate_table()print(f"{table._meta.name} truncated")db.execute_sql("set foreign_key_checks=1")# 插入测试数据
faker = Faker()
role_admin: Role = Role.create(name="管理员")
role_normal: Role = Role.create(name="普通用户")admin_user: User = User.create(login="admin", password="admin123", name="管理员", join_date=datetime.now().date()
)
admin_user.roles.add(role_admin)for x in range(0, 10):# 添加用户user = User.create(login=faker.user_name(),password=faker.password(),name=faker.name(),join_date=faker.date_object(),)user.roles.add(role_normal)# 添加测试文章for y in range(0, 10):post = Post.create(title=faker.text(50), content=faker.text(800), author=user)# 查询数据
print("-" * 10)
print("查询所有用户:")
print("-" * 10)
_user: User
for _user in User.select():print(_user.name)print("-" * 10)
print("查询文章总数:")
print("-" * 10)
print(Post.select().count())print("-" * 10)
print("查询所有role.name==普通用户")
print("-" * 10)
_role = Role.select().where(Role.name == "普通用户").get()
for _user in _role.users:print(_user.name)print("-" * 10)
print("查询用户 id==2 的所有文章")
print("-" * 10)
user_2 = User.select().where(User.id == 2)
_post: Post
for _post in Post.select().where(Post.author == user_2):print(_post.title)