示例应用程序

我们将构建一个类似于 twitter 的简单网站。示例的源代码可以在 examples/twitter 目录中找到。您还可以在 github 上浏览源代码。如果更喜欢的话，还有一个示例博客应用程序，但本指南中未涵盖此应用程序。

示例应用程序使用 flask Web 框架，该框架非常容易上手。如果您还没有 flask，则需要安装它才能运行示例

pip install flask

运行示例

确保已安装 flask 后，cd 到 twitter 示例目录并执行 run_example.py 脚本

python run_example.py

示例应用程序可以在 http://localhost:5000/ 访问

深入了解代码

为简单起见，所有示例代码都包含在一个模块 examples/twitter/app.py 中。有关使用 peewee 构建较大型 Flask 应用程序的指南，请查看构建 Flask 应用程序。

模型

秉承流行的 Web 框架 Django 的精神，peewee 使用声明式模型定义。如果您不熟悉 Django，其理念是为每个表声明一个模型类。然后，模型类定义一个或多个字段属性，这些属性对应于表的列。对于 twitter 克隆，只有三个模型

用户:: 表示用户帐户并存储用户名和密码、用于使用 gravatar 生成头像的电子邮件地址，以及指示该帐户创建时间的日期时间字段。
关系:: 这是一个实用模型，其中包含两个外键，指向用户模型，并存储哪些用户关注彼此。
消息:: 类似于推文。消息模型存储推文的文本内容、创建推文的时间以及发布推文的人（外键，指向用户）。

如果您喜欢 UML，这些是表和关系

为了创建这些模型，我们需要实例化一个 SqliteDatabase 对象。然后，我们定义我们的模型类，将列指定为类上的 Field 实例。

# create a peewee database instance -- our models will use this database to
# persist information
database = SqliteDatabase(DATABASE)

# model definitions -- the standard "pattern" is to define a base model class
# that specifies which database to use.  then, any subclasses will automatically
# use the correct storage.
class BaseModel(Model):
    class Meta:
        database = database

# the user model specifies its fields (or columns) declaratively, like django
class User(BaseModel):
    username = CharField(unique=True)
    password = CharField()
    email = CharField()
    join_date = DateTimeField()

# this model contains two foreign keys to user -- it essentially allows us to
# model a "many-to-many" relationship between users.  by querying and joining
# on different columns we can expose who a user is "related to" and who is
# "related to" a given user
class Relationship(BaseModel):
    from_user = ForeignKeyField(User, backref='relationships')
    to_user = ForeignKeyField(User, backref='related_to')

    class Meta:
        # `indexes` is a tuple of 2-tuples, where the 2-tuples are
        # a tuple of column names to index and a boolean indicating
        # whether the index is unique or not.
        indexes = (
            # Specify a unique multi-column index on from/to-user.
            (('from_user', 'to_user'), True),
        )

# a dead simple one-to-many relationship: one user has 0..n messages, exposed by
# the foreign key. a users messages will be accessible as a special attribute,
# User.messages.
class Message(BaseModel):
    user = ForeignKeyField(User, backref='messages')
    content = TextField()
    pub_date = DateTimeField()

注意

请注意，我们创建了一个 BaseModel 类，该类简单地定义了我们想要使用的数据库。然后，所有其他模型都扩展此类，并且还将使用正确的数据库连接。

Peewee 支持许多不同的字段类型，这些字段类型映射到数据库引擎通常支持的不同列类型。Python 类型与数据库中使用的类型之间的转换是透明处理的，允许你在应用程序中使用以下内容

字符串（unicode 或其他）
整数、浮点数和 Decimal 数字。
布尔值
日期、时间和日期时间
None (NULL)
二进制数据

创建表

为了开始使用模型，必须创建表。这是一个一次性操作，可以使用交互式解释器快速完成。我们可以创建一个小型帮助器函数来完成此操作

def create_tables():
    with database:
        database.create_tables([User, Relationship, Message])

在示例应用程序旁边的目录中打开 Python shell 并执行以下操作

>>> from app import *
>>> create_tables()

注意

如果你遇到 ImportError，则表示未找到 flask 或 peewee，并且可能未正确安装。查看安装和测试文档，了解有关安装 peewee 的说明。

每个模型都有一个 create_table() 类方法，该方法在数据库中运行 SQL CREATE TABLE 语句。此方法将创建表，包括所有列、外键约束、索引和序列。通常，这只是在添加新模型时才会执行一次的操作。

Peewee 提供了一个帮助器方法 Database.create_tables()，该方法将解决模型间依赖关系，并对每个模型调用 create_table()，确保按顺序创建表。

注意

在创建表后添加字段需要你删除表并重新创建它，或使用 ALTER TABLE 查询手动添加列。

或者，你可以使用模式迁移扩展，使用 Python 更改数据库模式。

建立数据库连接

你可能已经注意到在上面的模型代码中，在名为 Meta 的基本模型上定义了一个类，该类设置了 database 属性。Peewee 允许每个模型指定它使用的数据库。你可以指定许多 Meta 选项，这些选项控制模型的行为。

这是一个 peewee 惯用法

DATABASE = 'tweepee.db'

# Create a database instance that will manage the connection and
# execute queries
database = SqliteDatabase(DATABASE)

# Create a base-class all our models will inherit, which defines
# the database we'll be using.
class BaseModel(Model):
    class Meta:
        database = database

在开发 Web 应用程序时，通常会在请求开始时打开连接，并在返回响应时关闭连接。您应始终明确管理连接。例如，如果您正在使用连接池，只有在您调用 connect() 和 close() 时，连接才会正确回收。

我们将告诉 Flask，在请求/响应周期中，我们需要创建与数据库的连接。Flask 提供了一些便捷的装饰器，可以轻松实现此目的

@app.before_request
def before_request():
    database.connect()

@app.after_request
def after_request(response):
    database.close()
    return response

注意

Peewee 使用线程局部存储来管理连接状态，因此此模式可与多线程 WSGI 服务器配合使用。

执行查询

在 User 模型中有一些实例方法，它们封装了一些特定于用户的功能

following()：此用户关注了谁？
followers()：谁关注了此用户？

这些方法在实现上类似，但在 SQL JOIN 和 WHERE 子句中存在重要差异

def following(self):
    # query other users through the "relationship" table
    return (User
            .select()
            .join(Relationship, on=Relationship.to_user)
            .where(Relationship.from_user == self)
            .order_by(User.username))

def followers(self):
    return (User
            .select()
            .join(Relationship, on=Relationship.from_user)
            .where(Relationship.to_user == self)
            .order_by(User.username))

创建新对象

当新用户想要加入网站时，我们需要确保用户名可用，如果可用，则创建一个新的 User 记录。查看 join() 视图，我们可以看到我们的应用程序尝试使用 Model.create() 创建用户。我们使用唯一约束定义了 User.username 字段，因此如果用户名已被占用，数据库将引发 IntegrityError。

try:
    with database.atomic():
        # Attempt to create the user. If the username is taken, due to the
        # unique constraint, the database will raise an IntegrityError.
        user = User.create(
            username=request.form['username'],
            password=md5(request.form['password']).hexdigest(),
            email=request.form['email'],
            join_date=datetime.datetime.now())

    # mark the user as being 'authenticated' by setting the session vars
    auth_user(user)
    return redirect(url_for('homepage'))

except IntegrityError:
    flash('That username is already taken')

当用户希望关注某人时，我们将使用类似的方法。为了表示关注关系，我们在 Relationship 表中创建一行，从一个用户指向另一个用户。由于 from_user 和 to_user 上的唯一索引，我们将确保不会出现重复行

user = get_object_or_404(User, username=username)
try:
    with database.atomic():
        Relationship.create(
            from_user=get_current_user(),
            to_user=user)
except IntegrityError:
    pass

执行子查询

如果您已登录并访问 Twitter 主页，您将看到您关注的用户发布的推文。为了干净地实现此功能，我们可以使用子查询

注意

子查询 user.following() 通常默认选择 User 模型上的所有列。由于我们将其用作子查询，因此 peewee 将仅选择主键。

# python code
user = get_current_user()
messages = (Message
            .select()
            .where(Message.user.in_(user.following()))
            .order_by(Message.pub_date.desc()))

此代码对应于以下 SQL 查询

SELECT t1."id", t1."user_id", t1."content", t1."pub_date"
FROM "message" AS t1
WHERE t1."user_id" IN (
    SELECT t2."id"
    FROM "user" AS t2
    INNER JOIN "relationship" AS t3
        ON t2."id" = t3."to_user_id"
    WHERE t3."from_user_id" = ?
)

其他感兴趣的主题

示例应用程序中还有一些其他值得简要提及的巧妙之处。

对结果列表进行分页的支持在一个名为 object_list 的简单函数中实现（在 Django 中是它的推论）。所有返回对象列表的视图都使用此函数。

def object_list(template_name, qr, var_name='object_list', **kwargs):
    kwargs.update(
        page=int(request.args.get('page', 1)),
        pages=qr.count() / 20 + 1)
    kwargs[var_name] = qr.paginate(kwargs['page'])
    return render_template(template_name, **kwargs)

带有 login_required 装饰器的简单身份验证系统。第一个函数在用户成功登录时，只需将用户数据添加到当前会话中。装饰器 login_required 可用于包装视图函数，检查会话是否已验证，如果没有则重定向到登录页面。

def auth_user(user):
    session['logged_in'] = True
    session['user'] = user
    session['username'] = user.username
    flash('You are logged in as %s' % (user.username))

def login_required(f):
    @wraps(f)
    def inner(*args, **kwargs):
        if not session.get('logged_in'):
            return redirect(url_for('login'))
        return f(*args, **kwargs)
    return inner

在数据库中找不到对象时，返回 404 响应，而不是引发异常。

def get_object_or_404(model, *expressions):
    try:
        return model.get(*expressions)
    except model.DoesNotExist:
        abort(404)

注意

为了避免频繁复制/粘贴 object_list() 或 get_object_or_404()，这些函数作为 playhouse flask 扩展模块的一部分包含在内。

from playhouse.flask_utils import get_object_or_404, object_list