wtforms的调用验证方式优化

通常, 在使用flask验证器的过程中视图会有类似下面的语句:

data = request.json
form = ClientForm(data=data)
if form.validate():
    #...
else:
    #...
  • 重写的目的:
  1. 无需每次在视图中使用data=reqeust.data, form = XXForm(data=data)获取和传入请求数据.
  2. 无需在视图中使用ifelse判断, 请求数据有问题时候, 直接向客户端返回异常.
  • 分析:

为了省去视图函数向验证器每次传数据的过程, 重写构造器的时候可以直接使用request.data获取;
为了直接返回异常, 需要重写validate, 此处我们不重写, 直接定义一个类似的方法, 去调用validate.


from flask import request from wtforms import Form # 这是之前定义的参数错误异常处理类 from app.libs.error_code import ParameterException class BaseForm(Form): def __init__(self): data = request.json super(BaseForm, self).__init__(data=data) def validate_for_api(self): valid = super(BaseForm, self).validate() if not valid: # 这里如果出错将直接跑出异常, 视图中将免除了判断并使用ifelse的麻烦 raise ParameterException(msg=self.errors) return self

这时候, 继承这个类的验证表单如果验证参数有问题, 将会’自动向客户端返回一个异常json数据, 而无需在后续使用raise, 因为在验证器已经将错误raise出去了.

在form定义的表单可以直接继承自定义的基础表单, 不在继承wtfforms的原始表单, 如:

class ClientForm(BaseForm):
    #...
    secret = StringField(validators=[...])  

之后在视图中如果需要验证变方便的多, 如: 

@api.route('/register', methods=['POST'])
def create_client():
    form = ClientForm().validate_for_api()
    return Success()

视图中不在需要使用if判断参数是否合法, 也不需要将request数据当做参数传到form对象中, 后续将大大减少代码量.

搭建自己的sentry 服务(docker+ubuntu)

Sentry 是一个开源实时错误的报告工具,支持 web 前后端、移动应用以及游戏,支持 Python、OC、Java、Go、Node.js、Django、RoR 等主流编程语言和框架 ,还提供了 GitHub、Slack、Trello 等常见开发工具的集成。它可以实时收集程序内部发生的异常,并友好展示在后台。

一. 安装docker-compose

二. 部署Sentry

三. 启动


一. 安装docker-compose

安装docker

sudo apt install docker.io

安装docker-compose, 在某个目录下执行:

sudo curl -L https://github.com/docker/compose/releases/download/1.21.2/docker-compose-$(uname -s)-$(uname -m) -o /usr/local/bin/docker-compose
sudo chmod 755 /usr/local/bin/docker-compose

二. 部署

克隆

git clone https://github.com/getsentry/onpremise
cd onpremise

生成一个SENTRY_SECRET_KEY, 运行后进入漫长等待:

docker-compose run --rm web config generate-secret-key

打开目录下docker-compose.yml, 修改SENTRY_SECRET_KEY

  base:
    restart: unless-stopped
    build: .
    environment:
      # Run `docker-compose run web config generate-secret-key`
      # to get the SENTRY_SECRET_KEY value.
      SENTRY_SECRET_KEY: '0cw5he02=baa2p2)o2c*(ludb1y8taomopy(8o%xpfu7qbpy7-))'  # 就是这里
      SENTRY_MEMCACHED_HOST: memcached
      SENTRY_REDIS_HOST: redis
      SENTRY_POSTGRES_HOST: postgres
      SENTRY_EMAIL_HOST: smtp
    volumes:
      - ./data/sentry:/var/lib/sentry/files

生成数据库, 并创建用户:

docker-compose run --rm web upgrade

三. 启动

docker-compose up -d

打开:http://url:9000登录并添加项目

python多线程高级锁condition

多线程编程中如果使用Condition对象代替lock, 能够实现在某个事件触发后才处理数据, condition中含有的方法:
– wait:线程挂起,收到notify通知后继续运行
– notify:通知其他线程, 解除其它线程的wai状态
– notifyAll(): 通知所有线程
– acquire和release: 获得锁和解除锁, 与lock类似,
– enter和exit使得对象支持上下文操作:

    def __enter__(self):
        return self._lock.__enter__()

    def __exit__(self, *args):
        return self._lock.__exit__(*args)

代码:

import threading
from threading import Condition
# condition


class XiaoAi(threading.Thread):
    def __init__(self, cond):
        self.cond = cond
        super().__init__(name="xiaoai")

    def run(self):
        self.cond.acquire()
        self.cond.wait()
        print('{}:ennn. '.format(self.name))
        self.cond.notify()
        self.cond.wait()

        print('{}:好嗒. '.format(self.name))
        self.cond.release()

class TianMao(threading.Thread):
    def __init__(self, cond):
        super().__init__(name="tiaomao")
        self.cond = cond

    def run(self):
        self.cond.acquire()
        print('{}:hello ~ xiaoai. '.format(self.name))
        self.cond.notify()
        self.cond.wait()

        print('{}:我们来念一首诗吧! . '.format(self.name))
        self.cond.notify()
        self.cond.release()

if __name__ == '__main__':
    condition = Condition()
    xiaoai = XiaoAi(condition)
    tianmao = TianMao(condition)
    # 启动顺序很重要
    xiaoai.start()
    tianmao.start()

打印结果:

tiaomao:hello ~ xiaoai. 
xiaoai:ennn. 
tiaomao:我们来念一首诗吧! . 
xiaoai:好嗒

总结:

这个比较鸡肋

jsonify的改进

一. json和jsonify

二. 改进的jsonify

三. 使用


一. json和jsonify

json.dumps需要指定返回类型:

    t = {
      #...
    }
    return Response(json.dumps(t), mimetype='application/json')

jsonify默认将返回的类型指定成了application/json :

    t = {
      #...
    }
    return jsonify(t), 200

但是在使用默认情况下, jsonify对于很多类型的数据是无法序列化的(当然json.dumps也不行), 比如时间类型的字典, model的实例对象.

二. 改进的jsonify

如果jsonify不知道如何序列化指定的对象, 程序便会调用JSONEncodedefualt方法:

# 源码部分
class JSONEncoder(_json.JSONEncoder):
    def default(self, o):
        # ...
        if isinstance(o, datetime):
            return http_date(o.utctimetuple())
        if isinstance(o, date):
            return http_date(o.timetuple())
        if isinstance(o, uuid.UUID):
            return str(o)
        if hasattr(o, '__html__'):
            return text_type(o.__html__())
        return _json.JSONEncoder.default(self, o)
    #...

其中参数o便是需要序列化的对象.
因此可以重新定义这个default方法, 使jsonify支持对sqlalchemy实例和datetime类型数据的序列化:

在合适的位置引入并重写:

from flask.json import JSONEncoder as _JSONEncoder
class JSONEncoder(_JSONEncoder):
    """重写对象序列化, 当默认jsonify无法序列化对象的时候将调用这里的default"""
    def default(self, o):
        if hasattr(o, 'keys') and hasattr(o, '__getitem__'):
            return dict(o)
        if isinstance(o, date):
            # 也可以序列化时间类型的对象
            return o.strftime('%Y-%m-%d')
        raise ServerError()
  • : dict(o)用到了对象转字典, 其中ServerError也是自定义的异常.

重写后必须指定flask的json_encoder:

from flask import Flask as _Flask
class Flask(_Flask):
    json_encoder = JSONEncoder

实例化flask对象的时候, 不要使用原版的Flask类, 而是使用刚刚改进后的class Flask(_Flask):., 直接app = Flask(__name__)即可.

三. 使用

改进后的jsonify已经支持对时间和model对象的序列化

from flask import jsonify

#...
    books = Book.query.all()
    return jsonify(books)

上下文管理器类和上下文管理器装饰器

一. 什么是上下文管理器

二. 自定义一个上下文管理器类:

三. 使用contextmanager

四. 一个例子, sqlalchemy: 数据库的自动提交和回滚


一. 什么是上下文管理器

上下文管理器是在Python2.5之后加入的功能,可以在方便的需要的时候比较精确地分配和释放资源, with便是上下文管理器的最广泛的应用, 比如:

with open("test/test.txt","w") as f:
  f.write("hello")

这上会比使用try:...finally:f.close方便的多.

二. 自定义一个上下文管理器类:

class MyResource:
    # __enter__ 返回的对象会被with语句中as后的变量接受
    def __enter__(self):
        print('connect to resource')
        return self

    def __exit__(self, exc_type, exc_value, tb):
        print('close resource conection')

    def query(self):
        print('query data')

类中有两个特殊的魔术方法:
__enter__: with语句中的代码块执行前, 会执行__enter__, 返回的值将赋值给with句中as后的变量.
__exit__: with语句中的代码块执行结束或出错, 会执行_exit__

比如以下代码:

with Myresource() as r:
    r.query()

的打印结果为:

connect to resource
query data
close resource conection

那么有没有一个简化定义的方法呢, python提供了一个装饰器contextmanager

三. 使用contextmanager

from contextlib import contextmanager
class MyResource:
    def query(self):
        print('query data')

@contextmanager
def make_myresource():
    print('start to connect')
    yield MyResource()
    print('end connect')
    pass

被装饰器装饰的函数分为三部分:
with语句中的代码块执行前执行函数中yield之前代码
yield返回的内容复制给as之后的变量
with代码块执行完毕后执行函数中yield之后的代码

比如下方代码:

with make_myresource() as r:
     r.query()

的结果为:


start to connect query data end connect

四. 一个例子, sqlalchemy: 数据库的自动提交和回滚

在编程中如果频繁的修改数据库, 一味的使用类似try:... except..: rollback() raise e其实是不太好的.

比如某一段的代码的是这样的:

   try:
        gift = Gift()
        gift.isbn = isbn
        ... 
        db.session.add(gift)
        db.session.commit()
    except Exception as e:
        db.session.rollback()
        raise e

为了达到使用with语句的目的, 我们可以重写db所属的类:

from flask_sqlalchemy import SQLAlchemy as _SQLALchemy
class SQLAlchemy(_SQLALchemy):
    @contextmanager
    def auto_commit(self):
        try:
            yield
            self.session.commit()
        except Exception as e:
            db.session.rollback()
            raise e

这时候, 在执行数据的修改的时候便可以:

 with db.auto_commit():
        gift = Gift()
        gift.isbn = isbndb.session.add(gift)
        db.session.add(gift)

with db.auto_commit():
    user = User()
    user.set_attrs(form.data)
    db.session.add(user)