Flask常用扩展用法 | Patrick's Space

Flask-SQLAlchemy

强大的 ORM 工具，让对数据库的操作变得简单。

初始化

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
db.init_app(app)

配置

# 根据实际的数据库类型来确定 URI
SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'dev.sqlite')
SQLALCHEMT_ECHO = True  # 可选
SQLALCHEMY_TRACK_MODIFICATIONS = True  # 需要显式指明

创建模型

class Role(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), unique=True)
    description = db.Column(db.String(255))

    def __init__(self, name='default'):
        self.name = name

    def __repr__(self):
        return '<Role {}>'.format(self.name)

操作数据库

db.create_all()
db.session.query()
db.session.add()
db.session.commit()
db.session.flush()
db.drop_all()

Flask-Login

好用的登录控制扩展，功能全面并且强大。

初始化

from flask-login import LoginManger

login_manger.login_view = 'main.login'
login_manger.session_protection = 'strong'
login_manger.login_message = '请登录以访问该页面'
login_manger.login_message_category = 'info'

login_manger.init_app(app)

配置

需要在用户类里面实现特定的方法，或直接继承 UserMixin，注意如果用户的主键不是 id 则要自己实现 user_loader() 方法。

from flask_login import UserMixin, AnonymousUserMixin

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(255))
    password = db.Column(db.String(255))
    

class AnonymousUser(AnonymousUserMixin):
    pass

作为函数的装饰器

@main_blueprint.route('/logout', methods=['GET', 'POST'])
@login_required
def logout():
    logout_user()
    flash('登出成功！', category='success')
    return redirect(url_for('blog.home'))

登入登出

login_user(user, remember=form.remember_me.data)
logout_user()

获得当前用户对象

from flask-login import current_user

在 Jinja2 模板中可以直接调用

{{ current_user.name }}

Flask-Bcrypt

用来生成和检查加密字符串，常用作密码管理。

初始化

from flask-bcrypt import Bcrypt

bcrypt = Bcrypt()
bcrypt.init_app(app)

生成与检验密码

# property 装饰器将方法封装成了属性
@property
def passwd(self):
    raise AttributeError('密码不可读！')

# 通过下面的方法来实现直接对“属性”赋值
# self.passwd = 'aaa'
@passwd.setter
def passwd(self, password):
    # 生成密码
    self.password = bcrypt.generate_password_hash(password)

def verify_password(self, password):
    # 检验密码
    return bcrypt.check_password_hash(self.password, password)

Flask-Principal

权限管理扩展。

基本概念介绍

Identity

Identity 用来表示身份，而 Need 是 Identity 的一部分，用来表示具体的权限。

Identity 一般在身份状态改变的时候创建，此时也会发出 identity_loaded 的信号，例如：

identity_changed.send(current_app._get_current_object(),
                      identity=Identity(user.id))

Identity 使用用户的 id 来表示用户以及创建对象。

Need

Need 如上面所说是用来表示具体的权限，是用 namedtuple 来实现的。

namedtuple 是 Python collections 模块中的一个数据结构，实现类似 C 语言中 struct 的作用。

>>> from collections import namedtuple
>>> MyNT = namedtuple('tuple_name', ['attr1', 'attr2'])
>>> MyNT.__name__
'tuple_name'
>>> nt = MyNT('this is the first attr', 'hello')
>>> nt.attr1
'this is the first attr'

Need 的定义如下：

Need = namedtuple('Need', ['method', 'value'])

Need 一般是 tuple，之后在 Permission 初始化的时候会被转换为 set，set 的基本元素是 Need。

UserNeed 和 RoleNeed 是 Need 的两种典型的表现形式，前者用 user id 来表示 method（个人感觉和 Identity 有些重叠），后者用 role 表示 method 作为权限。它们使用 Python 的 functools 中的 partial 来固定了 Need 的 method。

partial 一般用来固定函数的某个参数的值，产生一个偏函数，例如：

>>> pow(2, 3)
8
>>> from functools import partial
>>> my_pow = partial(pow, 2)  # 固定了 2 作为了 pow 的第一个参数
>>> my_pow(5)
32

RoleNeed 和 UserNeed 就是这样产生的：

RoleNeed = partial(Need, 'role')
UserNeed = partial(Need, 'id')

Permission

Permission 用来表示权限，和具体的 Identity 无关，但是可以用来判断当前的 Identity 是否具有相应的权限。

Permission 使用 Need 来初始化：

admin_permission = Permission(RoleNeed('admin'))
poster_permission = Permission(RoleNeed('poster'))
default_permission = Permission(RoleNeed('default'))

使用 can() 方法来检测当前用户是否具有相应的权限。

if not admin_permission.can():
    abort(403)

关联

使用 Permission 表示权限，Identity 表示用户，用户所拥有的权限用 Need 来表示，通过 Permission 来检测 Identity 里面是否具有相应的 Need 来判断这个用户是否有相应的权限。

初始化

from flask_principal import Principal, Permission, RoleNeed, identity_loaded()

principals = Principal()
# 定义了三种权限
admin_permission = Permission(RoleNeed('admin'))
poster_permission = Permission(RoleNeed('poster'))
default_permission = Permission(RoleNeed('default'))

principals.init_app(app)

# 当接收到 identity_loaded 信号时会被调用
# 此时会将 Need 添加到相应的身份对象中
@identity_loaded.connect_via(app)
def on_identity_loaded(sender, identity):
    identity.user = current_user

    if hasattr(current_user, 'id'):
        identity.provides.add(UserNeed(current_user.id))

    if hasattr(current_user, 'roles'):
        for role in current_user.roles:
            identity.provides.add(RoleNeed(role.name))

用法

作为装饰器或者直接调用 can() 方法来判断。

@blog_print.route('/edit/<int:post_id>', methods=['GET', 'POST'])
# 没有 poster 权限无法访问
@poster_permission.require(http_exception=403) 
@login_required
def edit_post(post_id):

    post = Post.query.get_or_404(post_id)
    # 这个权限表示只有创建者才能访问
    permission = Permission(UserNeed(post.user_id))
    # 如果不是作者而是管理员则也可以访问
    if not (permission.can() or admin_permission.can()):
        abort(403)

    ...

源码分析

Identity

class Identity(object):
    """Represent the user's identity.

    :param id: The user id
    :param auth_type: The authentication type used to confirm the user's
                      identity.

    The identity is used to represent the user's identity in the system. This
    object is created on login, or on the start of the request as loaded from
    the user's session.

    Once loaded it is sent using the `identity-loaded` signal, and should be
    populated with additional required information.

    Needs that are provided by this identity should be added to the `provides`
    set after loading.
    """
    def __init__(self, id, auth_type=None):
        self.id = id
        self.auth_type = auth_type
        self.provides = set()

    def can(self, permission):
        """Whether the identity has access to the permission.

        :param permission: The permission to test provision for.
        """
        return permission.allows(self)


class AnonymousIdentity(Identity):
    """An anonymous identity"""

    def __init__(self):
        Identity.__init__(self, None)

Identity 接受 user id 作为参数，provides 属性是和 Permission 类似都是一个用来储存 Need 的 set，其中 provides 的内容在切换身份的时候添加。

Identity 可以使用 can() 方法来判断该 Identity 是否具有相应的权限，这个方法接受一个 Permission 来作为参数，实际上直接调用 Permission 的方法来检查权限。

而匿名 Identity 则是 id 为空的特殊 Identity。

IdentityContext

class IdentityContext(object):
    """The context of an identity for a permission.

    .. note:: The principal is usually created by the flaskext.Permission.require method
              call for normal use-cases.

    The principal behaves as either a context manager or a decorator. The
    permission is checked for provision in the identity, and if available the
    flow is continued (context manager) or the function is executed (decorator).
    """

    def __init__(self, permission, http_exception=None):
        self.permission = permission
        self.http_exception = http_exception
        """The permission of this principal
        """

    @property
    def identity(self):
        """The identity of this principal
        """
        return g.identity

    def can(self):
        """Whether the identity has access to the permission
        """
        return self.identity.can(self.permission)

IdentityContext 通常通过 Permission.require() 方法来创建，它接受 Permission 作为参数，绑定的 Identity 是当前用户的 Identity（通过 g 来实现）。

Permission

class Permission(object):
    """Represents needs, any of which must be present to access a resource

    :param needs: The needs for this permission
    """
    def __init__(self, *needs):
        """A set of needs, any of which must be present in an identity to have
        access.
        """

        self.needs = set(needs)
        self.excludes = set()

    def require(self, http_exception=None):
        """Create a principal for this permission.

        The principal may be used as a context manager, or a decroator.

        If ``http_exception`` is passed then ``abort()`` will be called
        with the HTTP exception code. Otherwise a ``PermissionDenied``
        exception will be raised if the identity does not meet the
        requirements.

        :param http_exception: the HTTP exception code (403, 401 etc)
        """
        return IdentityContext(self, http_exception)

    def allows(self, identity):
        """Whether the identity can access this permission.

        :param identity: The identity
        """
        if self.needs and not self.needs.intersection(identity.provides):
            return False

        if self.excludes and self.excludes.intersection(identity.provides):
            return False

        return True

    def can(self):
        """Whether the required context for this permission has access

        This creates an identity context and tests whether it can access this
        permission
        """
        return self.require().can()

Permission 可以说是最关键的部分了，它接受 Need 的 list 作为参数，并将其转换为 set 来储存所有的权限。为什么是 set 呢？因为 set 不可重复，并且做集合运算很容易直到是否存在某个 Need。

它的 require() 方法是使用得最多的方法之一，在调用这个方法的时候会创建一个 IdentityContext 实例，它会调用 IdentityContext.__enter__() 方法来检查权限，而这个方法里面又要调用 IdentityContext.can() 方法，这个方法又调用了相应的 Identity 对应的 Identity.can() 方法，并且传入当前 Identity 含有的权限作为参数。Identity 又调用 Permission.allows() 方法来检查权限，而这个方法则把 Permission 所具有的 Need 的 set 和 Identity.provides 这个 set 来做集合交运算来判断是否有相应的权限。（调用了一大圈，汗）

而 Permission.can() 的用法和上面的几乎一样。不过不同的是 require() 方法一般被用作装饰器，而 can() 方法一般用来直接进行判断。

Principal 总结

不得不说整个扩展设计得还是非常不错的，各个模块的逻辑性很强，只是太难理解了……我前前后后看了两遍……这次才终于搞清楚前前后后的逻辑了……

果然还是要看源代码，话说源代码写得还真不错，不仅文档详实，代码真的超级规范！