python 工具函数代码

Posted 2022-03-04 胖虎是只mao

1. 实现一个判断redis 服务器时候存在的装饰器

from functools import wraps
from flask import g

from rmon.common.rest import RestException


class ObjectMustBeExist:
    """该装饰器用于对某一个 Server 实例进行删改查操作，确保操作对象必须存在
    """

    def __init__(self, object_class):
        """
        Args:
            object_class (class): 数据库对象
        """

        self.object_class = object_class

    def __call__(self, func):
        """装饰器实现
        """
        @wraps(func)
        def wrapper(*args, **kwargs):
            """
            Args:
                object_id (int): SQLAlchemy object id
            """

            object_id = kwargs.get('object_id')
            if object_id is None:
                raise RestException(404, 'object not exist')

            obj = self.object_class.query.get(object_id)
            if obj is None:
                raise RestException(404, 'object not exist')

            g.instance = obj
            return func(*args, **kwargs)

        return wrapper

Python 的装饰器一般都是一个函数，是可调用的，但这里实现的 ObjectMustBeExist 却是一个类，但它的实例也可以像普通的装饰器函数一样使用，因为定义了 __call__ 特殊方法，例如可以使用 ObjectMustBeExist(Server)(func) 的语法装饰 func 函数。ObjectMustBeExist 初始化过程中，可以传递一个书局库模型类，类似于前文中定义的 Server Redis 服务器数据库模型，这样整个装饰器就会判断相应的模型对象是否存在。当通过 object_id 找到相应的模型实例后，会将其存储在 flask.g 对象中，它是一个应用上下文变量，每一个 HTTP 请求进来后，获取到的模型对象实例都将是不同的。

有了 ObjectMustBeExist 装饰器， /server/sever_id 对应的 API 实现就比较简单了:

# 省略已完成的代码

class ServerDetail(RestView):
    """ Redis 服务器列表
    """

    # 将装饰器类的实例写入装饰器元组中以备调用
    method_decorators = (ObjectMustBeExist(Server), )

    def get(self, object_id):
        """获取服务器详情
        """
        # schema.dump 返回一个 marshmallow.schema.MarshalResult 类的实例
        # 该实例保存两个字典：Redis 服务器实例字典和错误信息字典
        # 此 get 方法在 RestView 类的 dispatch_request 方法中被调用
        # 调用前已经加了装饰器，如有错误则触发 RestException 异常
        data, _ = ServerSchema().dump(g.instance)
        return data

    def put(self, object_id):
        """更新服务器
        """
        # 这里需要在实例化 schema 载体的时候给 context 属性赋值
        # 也就是将 object_id 对应的 Server 实例存入 self.context 字典里
        schema = ServerSchema(context='instance': g.instance)
        # data 是字典对象，保存着由客户端发来的数据
        data = request.get_json()
        # partial=True 的作用是将 data 字典中的字段更新到 context.instance 中
        # 也就是说返回的 server 就是更新后的 g.instance 对象
        # 这是 ServerSchema 中的 create_or_update 方法实现的
        # 如果此参数的值为 False 则只对 data 这个字典进行反序列化
        # 这样的话 errors 字典里肯定有内容了
        server, errors = schema.load(data, partial=True)
        if errors:
            return errors, 400
        server.save()
        return 'ok': True

    def delete(self, object_id):
        """删除服务器
        """
        g.instance.delete()
        # 状态码 204 的意思是请求执行成功，但不跳转到新页面，也不刷新当前页面
        return 'ok': True, 204

代码最大程度利用了 ObjectMustBeExist 实例装饰器运行时保存的 Server 对象，也就是 g.instance。需要注意的是 put 方法中初始化 ServerSchema 的代码 schema = ServerSchema(context='instance': g.instance) ，由于在后续的 ServerSchema.load 反序列化过程中，需要验证是否有同名 Server 对象，但是为了区分是更新还是创建操作，所以这里通过 context 属性传入了当前更新的 Server 对象。

二、flask-sqlalchemy中Datetime的创建时间、修改时间,default，server_default，onupdate

falsk中如下两个字段

create_time1 = db.Column(db.DateTime, default=datetime.now)

create_time2 = db.Column(db.DateTime, default=datetime.now())

两者的区别:

第一个插入的是期望的, 数据的插入时间，每条数据插入时可自动根据当前时间生成

第二条是一个固定的时间, 程序部署的时间，所有的数据都是这个固定时间

实际上默认值在mysql数据库没有体现, 都是sqlalchemy在插入数据时加的

如果想在生成的table中有默认值使用server_default
SQLAlchemy的默认值和SQL默认值之间存在差异。通常，设置SQLAlchemy默认值就足够了，因为通常不应该直接与服务器交互。如果确实要在服务器端呈现默认值，请使用server_default和server_onupdate参数来Column。

name = db.Column(db.String(45), server_default='hh')

当我们要给布尔值类型指定server_default时，需要用到text

from sqlalchemy import text
is_domain = db.Column(db.Boolean,default=False,server_default=text('0'))

因为mysql的datetime类型的数据不支持函数, 所以没法指定默认值为当前时间

记录每次修改的时间,onupdate

update_time = db.Column(db.DateTime, default=datetime.now,onupdate=datetime.now)

此外，考虑从使用datetime.now更改为datetime.utcnow，因为UTC更容易在幕后使用。只有在向用户显示某些内容时才转换为本地时间。

示例：

class Base(db.Model):
    """ 所有 model 的一个基类，默认添加了时间戳
    """

    __abstract__ = True  # 表示不要把这个类当作 Model 类

    # 设置了 default 和 onupdate 这俩个时间戳都不需要自己去维护
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow,
                        onupdate=datetime.utcnow)


# 改为继承 Base 类
class User(Base):
    __tablename__ = 'user'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique=True, index=True, nullable=False)
    publish_courses = db.relationship('Course')

# 改为继承 Base 类
class Course(Base):
    __tablename__ = 'course'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), unique=True, index=True, nullable=False)
    author_id = db.Column(db.Integer, db.ForeignKey('user.id', ondelete='CASCADE'))
    author= db.relationship('User', uselist=False)

三、flask-sqlalchemy User表的创建

User表：

from werkzeug.security import generate_password_hash, check_password_hash

class User(Base):
    __tablename__ = 'user'

    # 用数值表示角色，方便判断是否有权限，比如说有个操作要角色为员工
    # 及以上的用户才可以做，那么只要判断 user.role >= ROLE_STAFF
    # 就可以了，数值之间设置了 10 的间隔是为了方便以后加入其它角色
    ROLE_USER = 10
    ROLE_STAFF = 20
    ROLE_ADMIN = 30

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(32), unique=True, index=True, nullable=False)
    email = db.Column(db.String(64), unique=True, index=True, nullable=False)
    # 默认情况下，sqlalchemy 会以字段名来定义列名，但这里是 _password, 所以明确指定数据库表列名为 password
    _password = db.Column('password', db.String(256), nullable=False)
    role = db.Column(db.SmallInteger, default=ROLE_USER)
    job = db.Column(db.String(64))
    publish_courses = db.relationship('Course')

    def __repr__(self):
        return '<User:>'.format(self.username)

    @property
    def password(self):
        """ Python 风格的 getter """
        return self._password

    @password.setter
    def password(self, orig_password):
        """ Python 风格的 setter, 这样设置 user.password 就会
        自动为 password 生成哈希值存入 _password 字段
        """
        self._password = generate_password_hash(orig_password)

    def check_password(self, password):
        """ 判断用户输入的密码和存储的 hash 密码是否相等
        """
        return check_password_hash(self._password, password)

    @property
    def is_admin(self):
        return self.role == self.ROLE_ADMIN

    @property
    def is_staff(self):
        return self.role == self.ROLE_STAFF

首先为用户表添加了 email、_password、role、job 字段。job 表示用户的工作，职位。role 是用户的角色，当前我们只创建了三种角色：普通用户（ROLE_USER)，公司的员工（ROLE_STAFF），管理员（ROLE_ADMIN）。用户注册后默认就是普通用户角色。

用户的密码为了数据安全一般不是明文存储的，一般是存储为原始密码的哈希值。Flask 的底层库 werkzeug 提供了用于生成密码哈希的函数 generate_password_hash 和检测密码哈希和密码是否相等的 check_password_hash 函数。

这里将 _password 设为一个私有字段，对外暴露的是 password property。用 Python 风格的 setter 在设置 password 时，使用 generate_password_hash 生成并设置哈希密码。接着接着封装了一个 check_password 函数方法测试用户登录时输入的密码和存储的哈希密码是否相等。然后就是俩个 property 方便判断用户是否是员工或者管理员角色。

遂重温下这个property的使用

在我们定义数据库字段类的时候,往往需要对其中的类属性做一些限制,一般用get和set方法来写,那在python中,我们该怎么做能够少写代码,又能优雅的实现想要的限制,减少错误的发生呢,这时候就需要我们的@property

用代码来举例子更容易理解,比如一个学生成绩表定义成这样:

class Student(object):

    def get_score(self):
        return self._score

    def set_score(self, value):
        if not isinstance(value, int):
            raise ValueError('score must be an integer!')
        if value < 0 or value > 100:
            raise ValueError('score must between 0 ~ 100!')
        self._score = value

我们调用的时候需要这么调用:

>>> s = Student()
>>> s.set_score(60) # ok!
>>> s.get_score()
60
>>> s.set_score(9999)
Traceback (most recent call last):
  ...
ValueError: score must between 0 ~ 100!

但是为了方便,节省时间,我们不想写s.set_score(9999)啊,直接写s.score = 9999不是更快么,加了方法做限制不能让调用的时候变麻烦啊,@property快来帮忙….

class Student(object):

    @property
    def score(self):
        return self._score

    @score.setter
    def score(self,value):
        if not isinstance(value, int):
            raise ValueError('分数必须是整数才行呐')
        if value < 0 or value > 100:
            raise ValueError('分数必须0-100之间')
        self._score = value

看上面代码可知,把get方法变为属性只需要加上@property装饰器即可,此时@property本身又会创建另外一个装饰器@score.setter,负责把set方法变成给属性赋值,这么做完后,我们调用起来既可控又方便

>>> s = Student()
>>> s.score = 60 # OK，实际转化为s.set_score(60)
>>> s.score # OK，实际转化为s.get_score()
60
>>> s.score = 9999
Traceback (most recent call last):
  ...
ValueError: score must between 0 ~ 100!