FastApi(二) -- 集成peewee操作数据库

站长

2024年06月09日 20:07 · 阅读数 29

FastApi(二) -- 集成peewee操作数据库

peewee是什么

Peewee 是一个简单而强大的 Python ORM（对象关系映射）库，它提供了轻量级、简单易用的数据库操作功能。它的设计目标是提供一个简单而直观的 API，同时保持高性能和可扩展性，使得开发者能够轻松地在 Python 应用中进行数据库操作。

简单易用： Peewee 的 API 设计简洁清晰，易于理解和使用。它提供了类似于 Python 数据结构的对象模型，使得数据库操作变得非常直观和自然。
轻量级： Peewee 是一个轻量级的 ORM 库，它的代码库相对较小，没有过多的依赖。这使得它非常适合于小型项目或需要高性能的应用。
支持多种数据库后端： Peewee 支持多种流行的数据库后端，包括 SQLite、MySQL、PostgreSQL 等。这使得开发者可以根据项目需求选择最适合的数据库。
灵活性： Peewee 提供了丰富的查询构建器和条件表达式，使得开发者能够灵活地构建复杂的查询语句。同时，它还支持原生 SQL 查询，满足了更高级的需求。
性能优化： 尽管 Peewee 是一个轻量级的 ORM 库，但它仍然具有优秀的性能。它通过一些优化技巧和缓存机制来提高查询效率，同时保持了稳定性和可靠性。
Peewee 拥有一个活跃的开发者社区，提供了丰富的文档和教程，以及持续的更新和维护。这使得开发者能够快速解决问题，并获取帮助和支持。

FastApi如何集成Peewee

准备工作
1. 安装FastAPI与Peewee相关包
```
pip install fastapi uvicron peewee
```

搭建FastApi项目并集成Peewee

创建一个新的 Python 文件，比如 app.py，并编写以下代码来创建一个基本的 FastAPI 应用：

from fastapi import fastapi

app = fastapi()

if __name__ = "__main__":
    uvicoorn.run(app="main:app", host="localhost", port=8000)

引入peewee包，并连接数据库（以mysql数据库为例）

from peewee import MySQLDatabase

mysqlDb = MySQLDatabase(
    database='book_libs',
    user='root',
    password='passwrod',
    host='127.0.0.1',
    port=3306)

# 开启服务时连接数据库
app.on_event('startup')
def startup():
    mysqlDb.connect()

# 关闭服务时断开连接
@app.on_event('shutdown')
def shutdown():
    mysqlDb.close()

创建数据表的映

-- book_libs.`user` definition

CREATE TABLE `user` (
  `id` bigint unsigned NOT NULL COMMENT '用户id',
  `user_name` varchar(100) NOT NULL COMMENT '用户名称',
  `user_code` varchar(100) NOT NULL COMMENT '用户编码',
  `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '编辑时间',
  `create_by` bigint unsigned NOT NULL COMMENT '创建人',
  `gender` int NOT NULL DEFAULT '1' COMMENT '性别(1 男 0 女)',
  `avatar` varchar(100) DEFAULT NULL COMMENT '头像',
  `age` int DEFAULT NULL COMMENT '年龄',
  `update_by` bigint unsigned DEFAULT NULL COMMENT '编辑人',
  `deleted` tinyint(1) NOT NULL DEFAULT '0' COMMENT '是否删除',
  PRIMARY KEY (`id`),
  KEY `idx_user_name` (`user_name`) USING BTREE,
  KEY `idx_update_time` (`update_time`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='用户信息';

上面创建了一张user表，下面根据peewee提供的model类来进行表字段的映射：

from peewee import Model, CharField, IntegerField, BigIntegerField, DateTimeField, BooleanField
from utils.SnowFlakeUtil import snowflake
from config.DatabaseConfig import mysqlDb
from datetime import datetime

class User(Model):
	id = IntegerField(primary_key=True, default=snowflake.generate)
    user_name = CharField()
    user_code = CharField()
    age = IntegerField()
    gender = IntegerField()
    avatar = CharField()
    deleted = BooleanField()
    create_by = BigIntegerField()
    update_by = BigIntegerField()
    create_time = DateTimeField(default=datetime.now)
    update_time = DateTimeField(default=datetime.now)
    
    class Meta:
        database = mysqlDb

上面就是user表python代码映射试下，在代码中看到有一个 '''class Meta''' 的代码块，这个是peewee官方规定在集成时需要用这个与数据库建立联系

通过上面的例子我们已经成功的集成了peewee去进行mysql数据库的操作。那么下面我们就看下具体的一些操作接口吧

Peewee常用的方法

peewee的官方文档中给去了他的接口文档，我们可以跟着接口文档去尝试每一个接口，去进行数据库的操作。接下来我给大家讲解下常用的一些api

插入数据

model.save()

def create_user(user: User):
    count: int = user.save()
    
# 该方法会在数据库中新增一条数据，同时会返回一个数字代表这次操作数据库影响会有多少行

Model.create()

# 方法传递一个实例
def create_user(user: User):
    insert_user: User = User.create(**user.__dict__)

# create()方法也会在数据库中插入一条数据，在插入数据的同时还会返回一个数据库模型的实例User

# 方法传递一个字典
def create_user(user_dict: Dict):
    insert_user: user = User.create(**user_dict)

通过create()方法进行创建时，create()方法中的参数是通过关键词参数进行传参，因此不能够直接通过传递实体类。

同时我们看create()方法的源码发现，create()方法的底层调用的是save()方法。在create()方法中先实例化一个实体，然后再调用的实体的save()方法

class Model(with_metaclass(ModelBase, Node)):
    def __init__(self, *args, **kwargs):
        if kwargs.pop('__no_default__', None):
            self.__data__ = {}
        else:
            self.__data__ = self._meta.get_default_dict()
        self._dirty = set(self.__data__)
        self.__rel__ = {}

        for k in kwargs:
            setattr(self, k, kwargs[k])

    def __str__(self):
        return str(self._pk) if self._meta.primary_key is not False else 'n/a'


    @classmethod
    def insert(cls, __data=None, **insert):
        return ModelInsert(cls, cls._normalize_data(__data, insert))

    @classmethod
    def create(cls, **query):
        inst = cls(**query)
        inst.save(force_insert=True)
        return inst

Model.insert()

如果不想要返回数据的话我们可以使用insert()方法进行插入数据

class Model(with_metaclass(ModelBase, Node)):
    @classmethod
    def insert(cls, __data=None, **insert):
        return ModelInsert(cls, cls._normalize_data(__data, insert))

看源码知道，insert方法中有三个参数：

cls 类本身
__data 数据要插入的字典对象。这个参数通常用于插入单条记录。如果提供了 __data 参数，则其余的关键字参数会被忽略。
**insert: 插入数据的关键字参数。这些参数可以是字段名和对应的值，也可以是字段名和函数调用、子查询等。这个参数通常用于插入多条记录或者插入特殊格式的数据。如果提供了 __data 参数，则 **insert 参数会被忽略。

下面通过两种方式来进行说明：

User.insert(__data={'user_name': 'John', 'age': 30}).execute()
# 在这个示例中，我们通过 `__data` 参数传递了要插入的数据，其中键是字段名，值是对应的值。

data = [
    {'user_name': 'John', 'age': 30},
    {'user_name': 'Alice', 'age': 25}
]
User.insert_many(data).execute()
# 在这个示例中，我们使用 insert_many 方法插入多条记录，将数据以列表的形式传递给 insert_many 方法。

编辑数据

编辑数据有两种方式：

通过model.save()方法进行编辑，该方法如何主键存在，就回根据主键去匹配已有的数据，并对已有的数据进行覆盖

Model.update() 方法可以进行编辑操作

def update_user(user: User):
    try:
        User.update(__data={
            'user_name': user.user_name,
            'user_code': user.user_code,
        }).where(User.id == user.id).execute()
    except Exception as e:
        logger.error("编辑用户{}信息失败, {}".format(user.id, e.__cause__))
        raise CustomException(status_code=400, detail='编辑用户信息失败')

删除数据
1. Model.delete_instance() 该方法是根据现有的实例来进行删除我们可以先查询到该条数据，然后根据这条数据删除
```
user = User.get(User.id == 1)
user.delete_instance()
```
2. Model.delete() 该条语句可以配合条件进行删除
```
q = User.delete().where(User.active == False)
q.execute()  # Remove the rows, return number of rows removed.
```
查询数据
1. Model.get() 该方法用于获取单条数据的查询
```
User.get(User.id == 1)
```
2. Model.get_by_id() 该方法是根据主键id来进行查询
```
User.get_by_id(1)  # Same as above.
```
3. Model.select() 该方法可以查询多条，也可以配合着条件语句进行查询
```
query = User.select().where(User.active == True).order_by(User.username)
```

上面是一些常用且较为简单的api，具体的更多的我们可以去官网上进行插叙学习

使用中的一些坑

自定义save()函数进行保存时，数据一直保存不了，但是有没有报错？

通过save()函数的源码我们发现，在进行保存操作时，函数回去判断是否有主键，且主键是否有值。

class Model(with_metaclass(ModelBase, Node)):
        def save(self, force_insert=False, only=None):
        field_dict = self.__data__.copy()
        if self._meta.primary_key is not False:
            pk_field = self._meta.primary_key
            pk_value = self._pk
        else:
            pk_field = pk_value = None
        if only is not None:
            field_dict = self._prune_fields(field_dict, only)
        elif self._meta.only_save_dirty and not force_insert:
            field_dict = self._prune_fields(field_dict, self.dirty_fields)
            if not field_dict:
                self._dirty.clear()
                return False

        self._populate_unsaved_relations(field_dict)
        rows = 1

        if self._meta.auto_increment and pk_value is None:
            field_dict.pop(pk_field.name, None)

        if pk_value is not None and not force_insert:
            if self._meta.composite_key:
                for pk_part_name in pk_field.field_names:
                    field_dict.pop(pk_part_name, None)
            else:
                field_dict.pop(pk_field.name, None)
            if not field_dict:
                raise ValueError('no data to save!')
            rows = self.update(**field_dict).where(self._pk_expr()).execute()
        elif pk_field is not None:
            pk = self.insert(**field_dict).execute()
            if pk is not None and (self._meta.auto_increment or
                                   pk_value is None):
                self._pk = pk
                # Although we set the primary-key, do not mark it as dirty.
                self._dirty.discard(pk_field.name)
        else:
            self.insert(**field_dict).execute()

        self._dirty -= set(field_dict)  # Remove any fields we saved.
        return rows

下面是我自定义的save()方法：

    def save(self, *args, **kwargs):
        self.update_time = datetime.now()
        if self.id is None:
            self.id = snowflake.generate()
            self.create_time = datetime.now()
        try:
            count = super().save(*args, **kwargs, force_insert=True)
            print("Save count:", count)  # 调试输出保存的记录数
            return count
        except Exception as e:
            print("Error saving:", e)  # 打印保存过程中的错误信息
            raise e

在方法中我们看见，判断了主键id是否存在，如果不存在我们就新加入一个值，这个时候在去调用父类的save()方法时，就回一直执行编辑操作。但是编辑的时候，通过id去匹配数据我们又没有匹配到具体的数据，因此这儿对数据的操作影响行数一直未0，也并未报错；

解决方法：

可以根据save()函数中的强制插入参数，让save()函数一直只做插入操作

使用create()方法进行插入操作时，报错save()方法的强制确认参数重复？

由于上面重写save()函数，将强制确认参数设置为true。而create方法底层调用的是save()方法并传参了强制确认参数造成。

解决方法：
1. 不重写save()方法，peewee会自动在新增时加入id；
2. 两种方法不同时用

转载自:https://juejin.cn/post/7373955162128252955