数据库工具¶

本节介绍用于管理连接、数据库 URL、模式迁移、内省、代码生成和测试的 playhouse 模块。

本页内容

数据库 URL
连接池
Schema Migrations
反射
pwiz - 模型生成器
测试工具

数据库 URL ¶

playhouse.db_url 模块允许您从连接字符串配置 Peewee，这在十二因素应用程序中很常见，其中数据库凭证存储在环境变量中。

import os
from playhouse.db_url import connect

db = connect(os.environ.get('DATABASE_URL', 'sqlite:////default.db'))

在查询字符串中传递其他关键字参数

db = connect('postgres://user:pass@host/db?max_connections=20')

URL 格式：scheme://user:password@host:port/dbname?option=value

常用 scheme

Scheme	数据库类
`sqlite:///path`	`SqliteDatabase`
`postgres://`	`PostgresqlDatabase`
`postgresext://`	`PostgresqlExtDatabase`
`mysql://`	`MySQLDatabase`

连接池实现

Scheme	数据库类
`sqlite+pool:///path`	`PooledSqliteDatabase`
`postgres+pool://`	`PooledPostgresqlDatabase`
`postgresext+pool://`	`PooledPostgresqlExtDatabase`
`mysql+pool://`	`PooledMySQLDatabase`

备用驱动程序

Scheme	数据库类
`psycopg3://`	`Psycopg3Database`
`psycopg3+pool://`	`PooledPsycopg3Database`
`cockroachdb://`	`CockroachDatabase`
`cockroachdb+pool://`	`PooledCockroachDatabase`
`cysqlite://`	`CySqliteDatabase`
`cysqlite+pool://`	`PooledCySqliteDatabase`
`apsw://`	`APSWDatabase`
`mariadbconnector://`	`MariaDBConnectorDatabase`
`mariadbconnector+pool://`	`PooledMariaDBConnectorDatabase`
`mysqlconnector://`	`MySQLConnectorDatabase`
`mysqlconnector+pool://`	`PooledMySQLConnectorDatabase`

connect(url, unquote_password=False, unquote_user=False, **connect_params)¶

参数

url – 数据库的 URL，请参阅示例。
unquote_password (bool) – 解码密码中的特殊字符。
unquote_user (bool) – 解码用户名中的特殊字符。
connect_params – 要传递给 Database 的其他参数。

解析 url 并返回一个适当的 Database 实例。

示例

sqlite:///my_app.db - 当前目录中的 SQLite 文件。
sqlite:///:memory: - 内存中的 SQLite。
sqlite:////absolute/path/to/app.db - 绝对路径的 SQLite。
postgresql://user:password@host:5432/dbname
mysql://user:password@host:3306/dbname

parse(url, unquote_password=False, unquote_user=False)¶

参数

url – 数据库的 URL，请参阅上面的 connect() 中的示例。
unquote_password (bool) – 解码密码中的特殊字符。
unquote_user (bool) – 解码用户名中的特殊字符。

解析 URL 并返回一个字典，其中包含 database、host、port、user 和 password 键，以及来自查询字符串的任何额外连接参数。

如果您需要手动构建数据库类，这将很有用

params = parse('postgres://user:pass@host:5432/mydb')
db = MyCustomDatabase(**params)

register_database(db_class, *names)¶

参数

db_class – Database 的子类。
names – 要用作 URL 中 scheme 的名称列表。

在 URL scheme 名称下注册自定义数据库类，以便 connect() 可以实例化它

register_database(FirebirdDatabase, 'firebird')
db = connect('firebird://my-firebird-db')

连接池 ¶

playhouse.pool 模块包含一些 Database 类，它们为 Postgresql、MySQL 和 SQLite 数据库提供连接池。池的工作方式是覆盖 Database 类中打开和关闭后端连接的方法。

在多线程应用程序中，每个线程都有自己的连接；池在任何时候最多维护 max_connections 个打开的连接。在单线程应用程序中，单个连接会被回收。

应用程序只需确保在工作完成后（通常在 HTTP 请求结束时）*关闭*连接。关闭池化连接会将其返回到池中，而不是实际断开连接。

from playhouse.pool import PooledPostgresqlDatabase

db = PooledPostgresqlDatabase(
    'my_app',
    user='postgres',
    max_connections=32,
    stale_timeout=300)

提示

池化数据库实现可以安全地作为其非池化对应物的即插即用替换。

常用池实现

PooledPostgresqlDatabase
PooledMySQLDatabase
PooledSqliteDatabase

其他实现

playhouse.cysqlite_ext - PooledCySqliteDatabase
playhouse.mysql_ext - PooledMariaDBConnectorDatabase
playhouse.mysql_ext - PooledMySQLConnectorDatabase
playhouse.postgres_ext - PooledPostgresqlExtDatabase
playhouse.postgres_ext - PooledPsycopg3Database
playhouse.cockroachdb - PooledCockroachDatabase

注意

使用 Peewee 的 asyncio 集成的应用程序不需要使用特殊的池化数据库 - Async 数据库默认使用连接池。

class PooledDatabase(database, max_connections=20, stale_timeout=None, timeout=None, **kwargs)¶

Mixin 类，混合到上面特定的后端子类中。

参数

database (str) – 数据库或数据库文件的名称。
max_connections (int) – 最大并发连接数。设置为 None 表示无限制。
stale_timeout (int) – 空闲连接被视为过期的秒数，下次重用时将被丢弃。
timeout (int) – 所有连接都在使用时阻塞的秒数。0 表示无限期阻塞；None（默认）表示立即引发。

注意

连接不会在恰好超出其 stale_timeout 时关闭。相反，过期的连接仅在请求新连接时关闭。

注意

如果池已满且未配置 timeout，则会引发 ValueError。

manual_close()¶: 永久关闭当前连接而不将其返回到池。当连接进入不良状态时使用此方法。

close_idle()¶: 关闭所有未使用的池化连接。

close_stale(age=600)¶

参数: age (int) – 连接被视为过期的年龄（秒）。
返回: 已关闭的连接数。

关闭已使用但已超过 age 秒的连接。请谨慎使用。

close_all()¶: 关闭所有连接，包括当前正在使用的连接。请谨慎使用。

class PooledSqliteDatabase(database, max_connections=20, stale_timeout=None, timeout=None, **kwargs)¶: SQLite 数据库的池实现。扩展了 SqliteDatabase。

class PooledPostgresqlDatabase(database, max_connections=20, stale_timeout=None, timeout=None, **kwargs)¶: Postgresql 数据库的池实现。扩展了 PostgresqlDatabase。

class PooledMySQLDatabase(database, max_connections=20, stale_timeout=None, timeout=None, **kwargs)¶: MySQL / MariaDB 数据库的池实现。扩展了 MySQLDatabase。

模式迁移 ¶

playhouse.migrate 模块提供了一个轻量级的 API，用于在不编写原始 SQL 的情况下对现有数据库进行增量模式更改。

peewee 迁移的理念是，依赖于数据库内省、版本控制和自动检测的工具通常是脆弱、易碎且不必要地复杂的。迁移可以写成简单的 python 脚本，并从命令行执行。由于迁移只依赖于应用程序的 Database 对象，迁移脚本不会引入新依赖项。

支持的操作

添加、重命名或删除列。
使列可为空或不可为空。
更改列的类型。
重命名表。
添加或删除索引和约束。
添加或删除列的默认值。

另请参阅

模式管理

from playhouse.migrate import SchemaMigrator, migrate

migrator = SchemaMigrator.from_database(db)

with db.atomic():
    migrate(
        migrator.add_column('tweet', 'is_published', BooleanField(default=True)),
        migrator.add_column('user', 'email', CharField(null=True)),
        migrator.drop_column('user', 'old_bio'),
    )

提示

将迁移包装在 db.atomic() 中，以确保更改不会被部分应用。

操作¶

添加列

# Non-null fields must supply a default value.
migrate(
    migrator.add_column('comment', 'pub_date', DateTimeField(null=True)),
    migrator.add_column('comment', 'body', TextField(default='')),
)

添加外键（列名必须包含 Peewee 默认追加的 _id 后缀）

user_fk = ForeignKeyField(User, field=User.id, null=True)
migrate(
    migrator.add_column('tweet', 'user_id', user_fk),
)

重命名列

migrate(
    migrator.rename_column('story', 'pub_date', 'publish_date'),
    migrator.rename_column('story', 'mod_date', 'modified_date'),
)

删除列

migrate(migrator.drop_column('story', 'old_field'))

可为空 / 不可为空

migrate(
    migrator.drop_not_null('story', 'pub_date'),  # Allow NULLs.
    migrator.add_not_null('story', 'modified_date'),  # Disallow NULLs.
)

更改类型

# Change a VARCHAR(...) to a TEXT field.
migrate(migrator.alter_column_type('person', 'email', TextField()))

重命名表

migrate(migrator.rename_table('story', 'stories'))

添加/删除索引

# Specify table, column(s), and unique/non-unique.
migrate(
    # Create an index on the `pub_date` column.
    migrator.add_index('story', ('pub_date',), False),  # Normal index.

    # Create a unique index on the category and title fields.
    migrator.add_index('story', ('category_id', 'title'), True),  # Unique.

    # Drop the pub-date + status index.
    migrator.drop_index('story', 'story_pub_date_status'),
)

添加/删除约束

from peewee import Check

# Add a CHECK() constraint to enforce the price cannot be negative.
migrate(migrator.add_constraint(
    'products',
    'price_check',
    Check('price >= 0')))

# Remove the price check constraint.
migrate(migrator.drop_constraint('products', 'price_check'))

# Add a UNIQUE constraint on the first and last names.
migrate(migrator.add_unique('person', 'first_name', 'last_name'))

列默认值

# Add a default value:
migrate(migrator.add_column_default('entry', 'status', 'draft'))

# Use a function (not supported in SQLite):
migrate(migrator.add_column_default('entry', 'created_at', fn.NOW()))

# SQLite-compatible function syntax:
migrate(migrator.add_column_default('entry', 'created_at', 'now()'))

# Remove a default:
migrate(migrator.drop_column_default('entry', 'status'))

注意

Postgres 用户在使用非标准 schema 时可能需要设置 search-path。这可以通过以下方式完成

migrator = PostgresqlMigrator(db)
migrate(
    migrator.set_search_path('my_schema'),
    migrator.add_column('table', 'field', TextField(default='')),
)

迁移 API¶

migrate(*operations)¶

执行一个或多个模式更改操作。

用法

migrate(
    migrator.add_column('t', 'col', CharField(default='')),
    migrator.add_index('t', ('col',), False),
)

class SchemaMigrator(database)¶

参数: database – 一个 Database 实例。

SchemaMigrator 负责生成模式更改语句。

classmethod from_database(database)¶

参数: database (Database) – 用于生成迁移的数据库实例。
返回: 适合与提供的数据库一起使用的 SchemaMigrator 实例。

工厂方法，它返回与给定数据库对应的 SchemaMigrator 子类。

add_column(table, column_name, field)¶

参数

table (str) – 要添加列的表的名称。
column_name (str) – 新列的名称。
field (Field) – 一个 Field 实例。

向提供的表添加一个新列。提供的 field 将用于生成适当的列定义。

如果字段不可为空，则必须指定默认值。

注意

对于非空列，将发生以下情况：

列被添加为允许 NULLs
执行 UPDATE 查询以填充默认值
列更改为 NOT NULL

drop_column(table, column_name, cascade=True)¶

参数

table (str) – 要从中删除列的表的名称。
column_name (str) – 要删除的列的名称。
cascade (bool) – 是否应使用 CASCADE 删除该列。

rename_column(table, old_name, new_name)¶

参数

table (str) – 包含要重命名列的表的名称。
old_name (str) – 列的当前名称。
new_name (str) – 列的新名称。

add_not_null(table, column)¶

参数

table (str) – 包含列的表的名称。
column (str) – 要设为不可为空的列的名称。

drop_not_null(table, column)¶

参数

table (str) – 包含列的表的名称。
column (str) – 要设为可为空的列的名称。

add_column_default(table, column, default)¶

参数

table (str) – 包含列的表的名称。
column (str) – 要添加默认值的列的名称。
default – 列的新默认值。请参阅下面的注释。

Peewee 会尝试正确引用默认值（如果它看起来像字符串字面量）。否则，默认值将被视为字面值。Postgres 和 MySQL 支持将默认值指定为 peewee 表达式，例如 fn.NOW()，但 Sqlite 用户需要改用 default='now()'。

drop_column_default(table, column)¶

参数

table (str) – 包含列的表的名称。
column (str) – 要从中删除默认值的列的名称。

alter_column_type(table, column, field, cast=None)¶

参数

table (str) – 表的名称。
column_name (str) – 要修改的列的名称。
field (Field) – 代表新数据类型的 Field 实例。
cast – (仅限 postgres) 如果数据类型不兼容，则指定转换表达式，例如 column_name::int。可以作为字符串或 Cast 实例提供。

更改列的数据类型。应谨慎使用此方法，因为使用不兼容的类型可能无法被数据库很好地支持。

rename_table(old_name, new_name)¶

参数

old_name (str) – 表的当前名称。
new_name (str) – 表的新名称。

add_index(table, columns, unique=False, using=None)¶

参数

table (str) – 要创建索引的表的名称。
columns (list) – 要索引的列的列表。
unique (bool) – 新索引是否应指定唯一约束。
using (str) – 索引类型（如果支持），例如 GiST 或 GIN。

drop_index(table, index_name)¶

参数

table (str) – 包含要删除的索引的表的名称。
index_name (str) – 要删除的索引的名称。

add_constraint(table, name, constraint)¶

参数

table (str) – 要添加约束的表。
name (str) – 用于识别约束的名称。
constraint – Check() 约束，或用于添加任意约束使用 SQL。

drop_constraint(table, name)¶

参数

table (str) – 要从中删除约束的表。
name (str) – 要删除的约束的名称。

add_unique(table, *column_names)¶

参数

table (str) – 要添加约束的表。
column_names (str) – 一个或多个用于 UNIQUE 约束的列。

class PostgresqlMigrator(database)¶

set_search_path(schema_name)¶: 为后续操作设置 Postgres search-path。

class SqliteMigrator(database)¶

SQLite 对 ALTER TABLE 查询的支持有限，因此以下操作目前不支持 SQLite：

add_constraint
drop_constraint
add_unique

class MySQLMigrator(database)¶: MySQL 特定的子类。

内省¶

playhouse.reflection 模块内省现有数据库并从其模式生成 Peewee 模型类。它由 pwiz - 模型生成器和 DataSet 内部使用。

from playhouse.reflection import generate_models

db = PostgresqlDatabase('my_app')
models = generate_models(db)   # Returns {table_name: ModelClass}

# list(models.keys())
# ['account', 'customer', 'order', 'orderitem', 'product']

# Get a reference to a generated model.
Customer = models['customer']

# Or inject into the current namespace:
# globals().update(models)

# Query generated models:
for customer in Customer.select():
    print(customer.name, customer.email)

generate_models(database, schema=None, **options)¶

参数

database (Database) – 要内省的数据库实例。
schema (str) – 可选的 schema 进行内省。
options – 任意选项，请参阅 Introspector.generate_models() 了解详情。

返回

一个 dict，映射表名到模型类。

print_model(model)¶

将模型的字段和索引的可读摘要打印到 stdout。用于交互式探索

>>> print_model(Tweet)
tweet
  id AUTO PK
  user INT FK: User.id
  content TEXT
  timestamp DATETIME

index(es)
  user_id
  timestamp

print_table_sql(model)¶

打印模型类（不带索引或约束）的 CREATE TABLE SQL

>>> print_table_sql(Tweet)
CREATE TABLE IF NOT EXISTS "tweet" (
  "id" INTEGER NOT NULL PRIMARY KEY,
  "user_id" INTEGER NOT NULL,
  "content" TEXT NOT NULL,
  "timestamp" DATETIME NOT NULL,
  FOREIGN KEY ("user_id") REFERENCES "user" ("id")
)

class Introspector(metadata, schema=None)¶

可以通过实例化 Introspector 来从数据库中提取元数据。不建议直接实例化此类，而是建议使用工厂方法 from_database()。

classmethod from_database(database, schema=None)¶

参数

database – 一个 Database 实例。
schema (str) – 可选的 schema（某些数据库支持）。

创建一个适合与给定数据库一起使用的 Introspector 实例。

db = SqliteDatabase('my_app.db')
introspector = Introspector.from_database(db)
models = introspector.generate_models()

# User and Tweet (assumed to exist in the database) are
# peewee Model classes generated from the database schema.
User  = models['user']
Tweet = models['tweet']

generate_models(skip_invalid=False, table_names=None, literal_column_names=False, bare_fields=False, include_views=False)¶

参数

skip_invalid (bool) – 跳过名称不是有效 Python 标识符的表。
table_names (list) – 仅为给定的表生成模型。
literal_column_names (bool) – 使用精确的数据库列名作为字段名（而不是转换为 Python 命名约定）。
bare_fields (bool) – 不尝试检测字段类型；对所有列使用 BareField（*仅限 SQLite*）。
include_views (bool) – 也为视图生成模型。

返回

一个将表名映射到模型类的字典。

内省数据库，读取表、列和外键约束，然后生成一个将每个数据库表映射到动态生成的 Model 类的字典。

pwiz - 模型生成器 ¶

pwiz 是一个命令行工具，它内省数据库并打印可直接使用的 Peewee 模型代码。如果您已经有一个现有的数据库，运行 pwiz 可以节省大量生成初始模型定义的时间。

# Introspect a Postgresql database and write models to a file:
python -m pwiz -e postgresql -u postgres my_db > models.py

# Introspect a SQLite database:
python -m pwiz -e sqlite path/to/my.db

# Introspect a MySQL database (prompts for password):
python -m pwiz -e mysql -u root -P my_db

# Introspect only specific tables:
python -m pwiz -e postgresql my_db -t user,tweet,follow

命令行选项

选项	含义	示例
`-e`	数据库后端	`-e mysql`
`-H`	主机	`-H 10.0.0.1`
`-p`	端口	`-p 5432`
`-u`	用户名	`-u postgres`
`-P`	密码（交互式提示）
`-s`	Schema	`-s public`
`-t`	要包含的表（逗号分隔列表）	`-t user,tweet`
`-v`	包括视图
`-i`	将数据库信息嵌入为注释
`-o`	保留原始列顺序
`-I`	忽略类型未知的字段
`-L`	使用旧版表和列命名

有效的 -e 值：sqlite、mysql、postgresql。

警告

如果需要密码才能访问数据库，系统将提示您使用安全提示输入密码。

密码将包含在输出中。具体来说，在文件顶部将定义一个 Database 以及任何必需的参数 - 包括密码。

SQLite 数据库包含 user 和 tweet 表的示例输出

from peewee import *

database = SqliteDatabase('example.db', **{})

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database

class User(BaseModel):
    username = TextField(unique=True)

    class Meta:
        table_name = 'user'

class Tweet(BaseModel):
    content = TextField()
    timestamp = DateTimeField()
    user = ForeignKeyField(column_name='user_id', field='id', model=User)

    class Meta:
        table_name = 'tweet'

请注意，pwiz 会检测外键、唯一约束，并保留显式表名。

注意

UnknownField 是一个占位符，当模式包含 Peewee 无法映射到字段类的列声明时使用。

测试工具 ¶

playhouse.test_utils 提供了用于测试 peewee 项目的帮助程序。

class count_queries(only_select=False)¶

计算其块内执行的 SQL 查询次数的上下文管理器。

参数: only_select (bool) – 如果为 True，则仅计算 SELECT 查询。

with count_queries() as counter:
    user = User.get(User.username == 'alice')
    tweets = list(user.tweets)   # Triggers a second query.

assert counter.count == 2

count¶: 执行的查询数。

get_queries()¶: 返回每个已执行查询的 (sql, params) 2 元组列表。

assert_query_count(expected, only_select=False)¶

装饰器或上下文管理器，如果执行的查询数与 expected 不匹配，则引发 AssertionError。

作为装饰器

class TestAPI(unittest.TestCase):
    @assert_query_count(1)
    def test_get_user(self):
        user = User.get_by_id(1)

作为上下文管理器

with assert_query_count(3):
    result = my_function_that_should_make_exactly_three_queries()

数据库工具¶

数据库 URL ¶

连接池 ¶

模式迁移 ¶

操作¶

迁移 API¶

内省¶

pwiz - 模型生成器 ¶

测试工具 ¶

peewee

导航

相关主题

数据库工具¶

数据库 URL¶

连接池¶

模式迁移¶

操作¶

迁移 API¶

内省¶

pwiz - 模型生成器¶

测试工具¶

数据库 URL ¶

连接池 ¶

模式迁移 ¶

pwiz - 模型生成器 ¶

测试工具 ¶