框架集成¶

对于Web应用程序，通常在接收到请求时打开连接，并在交付响应时关闭连接。本文档介绍了如何在您的Web应用中添加钩子，以确保正确处理数据库连接。

这些步骤将确保无论您使用的是简单的 SqliteDatabase 还是 PooledPostgresqlDatabase，peewee 都能正确处理连接。

模式总是相同的

# On request start:
db.connect()

# On request end (success or error):
if not db.is_closed():
    db.close()

每个框架都为此提供了钩子。下面各节展示了每种框架的惯用方法。

注意

处理大量流量的应用程序应使用连接池，以避免为每个请求建立新连接的开销。池化数据库可以作为其非池化数据库的直接替代品。

Flask¶

有关完整的 Flask + Peewee 应用程序示例，包括身份验证和其他常见的 Web 应用功能，请参阅示例应用。项目 examples/ 目录下还有一个完整的博客应用和一个分析应用。

最小的 Flask 集成确保数据库连接生命周期通过 before_request 和 teardown_request 钩子与请求/响应周期绑定。

from flask import Flask
from peewee import *

db = SqliteDatabase('my_app.db')
app = Flask(__name__)

@app.before_request
def _db_connect():
    db.connect()

@app.teardown_request
def _db_close(exc):
    if not db.is_closed():
        db.close()

无论请求成功还是引发异常，都会调用 teardown_request，这使其成为清理的正确钩子。

对于接收大量请求的应用程序，建议使用连接池

from flask import Flask
from playhouse.pool import PooledPostgresqlDatabase

db = PooledPostgresqlDatabase('app', host='10.8.0.1', user='postgres')
app = Flask(__name__)

# Note that when using the pooled implementation the hooks are the exact
# same. Opening and closing the connection simply acquires and releases it
# from the pool for the lifetime of the request.
@app.before_request
def _db_connect():
    db.connect()

@app.teardown_request
def _db_close(exc):
    if not db.is_closed():
        db.close()

另请参阅

Flask 工具扩展提供了用于常见任务的助手，例如声明式数据库配置、对象检索和分页。

FastAPI¶

FastAPI 是一个异步框架，可以与 Peewee 的异步支持集成或同步使用。Peewee 还提供了 Pydantic 集成支持，这与 FastAPI 配合良好。

关于 SQLModel 的简要说明¶

FastAPI 倡导使用 SQLModel 进行数据库访问。SQLModel 将 SQLAlchemy 和 Pydantic 组合成一个单一的类，这对于简单的示例可能运行良好。然而，有一些事情需要注意。

SQLModel 的官方教程只使用同步端点，FastAPI 在线程池上运行这些端点。异步用法被列为“高级”主题，目前没有文档。
在异步上下文中，惰性加载经常失效。SQLAlchemy 隐式地惰性加载关系，在与异步会话一起使用时可能触发 MissingGreenlet 错误。Peewee 也可能发生这种情况，但通过选择连接关系可以很容易地避免。
由于 SQLModel 对 DDL 和某些操作使用同步驱动，因此您通常需要安装同步和异步驱动，并进行单独的引擎配置。
SQLModel 使用继承来管理输入、输出和表模式。实际上，一个数据库表通常需要三到四个模型类，例如 UserBase、User、UserCreate 和 UserRead——所有这些都通过继承来管理。

Peewee 可能会提供更简单的体验——它只有一个内置连接池的数据库可供管理，更少的隐式惰性加载陷阱，并且使用 to_pydantic() 生成的 Pydantic 模式可以配置为包含/排除字段而无需继承。字段元数据会自动捕获：选项枚举、默认值、描述、标题和类型信息都捕获在 JSON 模式和 OpenAPI 文档中。

Peewee 需要少得多的机制来提供真正的 asyncio 数据库访问，当然也同样适用于同步 FastAPI 端点。

使用 Pydantic 的异步示例¶

下面是一个完整的 FastAPI 应用程序示例，演示了依赖注入风格的钩子、完全异步查询执行和 pydantic 集成

# example.py
from fastapi import Depends, FastAPI, HTTPException
from contextlib import asynccontextmanager
from peewee import *
from playhouse.pwasyncio import AsyncPostgresqlDatabase
from playhouse.pydantic_utils import to_pydantic


db = AsyncPostgresqlDatabase('peewee_test')

class User(Model):
    name = CharField(verbose_name='Full Name', help_text='Display name')
    email = CharField(unique=True)
    status = IntegerField(default=1, choices=(
        (1, 'Active'),
        (2, 'Inactive'),
        (3, 'Deleted')))
    class Meta:
        database = db

# Generate pydantic schemas suitable for create and responses.
# Schemas will include metadata from verbose_name, help_text, choices, and
# default settings.
UserCreate = to_pydantic(User, model_name='UserCreate')
UserResponse = to_pydantic(User, exclude_autofield=False, model_name='UserResponse')

async def get_db():
    async with db:
        yield db

@asynccontextmanager
async def lifespan(app):
    # Create tables (if they don't exist) at application startup.
    async with db:
        await db.acreate_tables([User])
    yield
    await db.close_pool()  # Shut-down pool and exit.

app = FastAPI(lifespan=lifespan)

@app.get('/users', response_model=list[UserResponse])
async def list_users(db=Depends(get_db)):
    rows = await db.list(User.select().dicts())
    return [UserResponse(**row) for row in rows]

@app.post('/users', response_model=UserResponse)
async def create_user(data: UserCreate, db=Depends(get_db)):
    user = await db.run(User.create, **data.model_dump())
    return UserResponse.model_validate(user)

@app.get('/users/{user_id}', response_model=UserResponse)
async def get_user(user_id: int, db=Depends(get_db)):
    try:
        user = await db.get(User.select().where(User.id == user_id))
    except User.DoesNotExist:
        raise HTTPException(status_code=404, detail='User not found')
    return UserResponse.model_validate(user)

运行示例

$ fastapi dev example.py

填充和查询数据

$ curl -X POST https://:8000/users \
     -H "Content-Type: application/json" \
     -d '{"name": "Alice", "email": "alice@example.com"}'

{"id":1,"name":"Alice","email":"alice@example.com","status":1}

$ curl -X POST https://:8000/users \
     -H "Content-Type: application/json" \
     -d '{"name": "Bob", "email": "bob@example.com", "status": 2}'

{"id":2,"name":"Bob","email":"bob@example.com","status":2}

$ curl https://:8000/users

[{"id":1,"name":"Alice","email":"alice@example.com","status":1},
 {"id":2,"name":"Bob","email":"bob@example.com","status":2}]

$ curl https://:8000/users/1

{"id":1,"name":"Alice","email":"alice@example.com","status":1}

我们还可以验证 Pydantic 模式是否捕获了我们的 Peewee 模型元数据

>>> UserCreate.model_json_schema()
{'properties': {
   'name': {
      'description': 'Display name',
      'title': 'Full Name',
      'type': 'string'},
   'email': {
      'title': 'Email',
      'type': 'string'},
  'status': {
      'default': 1,
      'description': 'Choices: 1 = Active, 2 = Inactive, 3 = Deleted',
      'enum': [1, 2, 3],
      'title': 'Status',
      'type': 'integer'}},
 'required': ['name', 'email'],
 'title': 'UserCreate',
 'type': 'object'}

另请参阅

依赖注入¶

以下是一个最小示例，演示了

确保使用数据库的端点自动打开和关闭连接。
在应用服务器启动时创建表/资源。
在应用服务器退出时关闭连接池。

from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI
from peewee import *
from playhouse.pwasyncio import *


app = FastAPI()

db = AsyncPostgresqlDatabase('peewee_test', host='10.8.0.1', user='postgres')

async def get_db():
    async with db:
        yield db

@asynccontextmanager
async def lifespan(app):
    async with db:
        await db.acreate_tables([User])
    yield
    await db.close_pool()

app = FastAPI(lifespan=lifespan)

@app.get('/users')
async def list_users(db=Depends(get_db)):
    return await db.list(User.select().dicts())

中间件和启动钩子¶

以下示例演示了如何使用中间件和启动钩子而不是依赖注入。

确保每个请求都打开和关闭连接。
在应用服务器启动时创建表/资源。
在应用服务器退出时关闭连接池。

from fastapi import FastAPI
from peewee import *
from playhouse.pwasyncio import *


app = FastAPI()

db = AsyncPostgresqlDatabase('peewee_test', host='10.8.0.1', user='postgres')

@app.middleware('http')
async def database_connection(request, call_next):
    async with db:  # Obtain connection from connection pool.
        response = await call_next(request)
    return response

@app.on_event('startup')
async def on_startup():
    async with db:
        await db.acreate_tables([Model1, Model2, Model3, ...])

@app.on_event('shutdown')
async def on_shutdown():
    await db.close_pool()

# Async queries.
@app.get('/users')
async def list_users():
    return await db.list(User.select().dicts())

@app.post('/users')
async def create_user(name: str):
    user = await db.run(User.create, name=name)
    return {'id': user.id, 'name': user.name}

同步 FastAPI¶

如果您正在使用 FastAPI 的同步端点，您可以使用同步 Peewee 数据库实现。这里是使用同步 Peewee 实现的上述“完整示例”

from fastapi import Depends, FastAPI, HTTPException
from contextlib import asynccontextmanager
from peewee import *
from playhouse.pydantic_utils import to_pydantic

db = PostgresqlDatabase('peewee_test')

class User(Model):
    name = CharField(verbose_name='Full Name', help_text='Display name')
    email = CharField(unique=True)
    status = IntegerField(default=1, choices=(
        (1, 'Active'),
        (2, 'Inactive'),
        (3, 'Deleted')))

    class Meta:
        database = db

# Generate pydantic schemas suitable for create and responses.
UserCreate = to_pydantic(User, model_name='UserCreate')
UserResponse = to_pydantic(User, exclude_autofield=False, model_name='UserResponse')

def get_db():
    with db.connection_context():
        yield db

@asynccontextmanager
async def lifespan(app):
    with db:
        db.create_tables([User])
    yield

app = FastAPI(lifespan=lifespan)

@app.get('/users', response_model=list[UserResponse])
def list_users(database=Depends(get_db)):
    rows = User.select().dicts()
    return [UserResponse(**row) for row in rows]

@app.post('/users', response_model=UserResponse)
def create_user(data: UserCreate, database=Depends(get_db)):
    user = User.create(**data.model_dump())
    return UserResponse.model_validate(user)

@app.get('/users/{user_id}', response_model=UserResponse)
def get_user(user_id: int, database=Depends(get_db)):
    try:
        user = User.get(User.id == user_id)
    except User.DoesNotExist:
        raise HTTPException(status_code=404, detail='User not found')
    return UserResponse.model_validate(user)

另请参阅

Pydantic 集成

Django¶

添加一个中间件，在视图运行前打开连接，并在响应准备好后关闭连接。将其放在 MIDDLEWARE 中第一个位置，以便它包装所有其他中间件。

# myproject/middleware.py
from myproject.db import database

def PeeweeConnectionMiddleware(get_response):
    def middleware(request):
        database.connect()
        try:
            response = get_response(request)
        finally:
            if not database.is_closed():
                database.close()
        return response
    return middleware

# settings.py
MIDDLEWARE = [
    'myproject.middleware.PeeweeConnectionMiddleware',
    # ... rest of middleware ...
]

Bottle¶

使用 before_request 和 after_request 钩子

from bottle import hook
from peewee import *

db = SqliteDatabase('my_app.db')

@hook('before_request')
def _connect_db():
    db.connect()

@hook('after_request')
def _close_db():
    if not db.is_closed():
        db.close()

Falcon¶

添加一个中间件组件

import falcon
from peewee import *

db = SqliteDatabase('my_app.db')

class DatabaseMiddleware:
    def process_request(self, req, resp):
        db.connect()

    def process_response(self, req, resp, resource, req_succeeded):
        if not db.is_closed():
            db.close()

app = falcon.App(middleware=[DatabaseMiddleware()])

Pyramid¶

设置自定义的 Request 工厂

from pyramid.request import Request
from peewee import *

db = SqliteDatabase('my_app.db')

class MyRequest(Request):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        db.connect()
        self.add_finished_callback(self._close_db)

    def _close_db(self, request):
        if not db.is_closed():
            db.close()

# In your application factory:
def main(global_settings, **settings):
    config = Configurator(settings=settings)
    config.set_request_factory(MyRequest)

Sanic¶

Sanic 是一个异步框架，可以与 Peewee 的异步支持集成使用。

from sanic import Sanic
from peewee import *
from playhouse.pwasyncio import *


app = Sanic('PeeweeApp')

db = AsyncPostgresqlDatabase('peewee_test', host='10.8.0.1', user='postgres')

@app.on_request
async def open_connection(request):
    await db.aconnect()  # Obtain connection from connection pool.

@app.on_response
async def close_connection(request, response):
    await db.aclose()  # Return connection to pool.

@app.before_server_start
async def setup_db(app):
    async with db:
        await db.acreate_tables([Model1, Model2, Model3, ...])

@app.before_server_stop
async def shutdown_db(app):
    await db.close_pool()

演示执行异步查询的示例

from sanic import json

@app.get('/message/')
async def message(request):
    # Get the latest message from the database.
    message = await db.get(Message.select().order_by(Message.id.desc()))
    return json({'content': message.content, 'id': message.id})

另请参阅

异步支持

CherryPy¶

订阅引擎的请求前/请求后事件

import cherrypy
from peewee import *

db = SqliteDatabase('my_app.db')

def _db_connect():
    db.connect()

def _db_close():
    if not db.is_closed():
        db.close()

cherrypy.engine.subscribe('before_request', _db_connect)
cherrypy.engine.subscribe('after_request', _db_close)

适用于任何框架的通用模式¶

如果您的框架未在此处列出，集成遵循相同的结构

找到在每个请求处理程序运行之前执行的钩子。
在那里调用 db.connect()。
找到在每个请求之后（无论是成功还是错误）运行的钩子。
如果连接已打开，请在那里调用 db.close()。

任何包装应用程序可调用对象的 WSGI 或 ASGI 中间件也可以管理此操作

class PeeweeMiddleware:
    def __init__(self, app, database):
        self.app = app
        self.db = database

    def __call__(self, environ, start_response):
        self.db.connect()
        try:
            return self.app(environ, start_response)
        finally:
            if not self.db.is_closed():
                self.db.close()

# Wrap your WSGI app:
application = PeeweeMiddleware(application, db)

框架集成¶

Flask¶

FastAPI¶

关于 SQLModel 的简要说明¶

使用 Pydantic 的异步示例¶

依赖注入¶

中间件和启动钩子¶

同步 FastAPI¶

Django¶

Bottle¶

Falcon¶

Pyramid¶

Sanic¶

CherryPy¶

适用于任何框架的通用模式¶

peewee

导航

相关主题