sanic 实践（0.6.0）

Overview

Sanic is a Python 3.7+ web server and web framework that’s written to go fast. It allows the usage of the async/await syntax added in Python 3.5, which makes your code non-blocking and speedy.

Example

下面是一个简单的例子，使用到如下库：

sanic (sanic is a flask-like python3.5+ web server)
peewee (simple and small ORM)
PyMySQL (mysql database driver)
marshmallow (serialize and deserialize models)
AoikLiveReload (automatic reload app in development)

目录结构

启动服务

(sanic) [blazehu@MacBook ~]$ python3 app.py
2018-12-21 10:53:50 - (sanic)[INFO]: Goin' Fast @ http://0.0.0.0:8000
2018-12-21 10:53:50 - (sanic)[INFO]: Starting worker [44832]

访问测试

create a new employee

list employees

logs

(sanic) [blazehu@MacBook ~]$ python3 app.py
2018-12-21 10:53:50 - (sanic)[INFO]: Goin' Fast @ http://0.0.0.0:8000
2018-12-21 10:53:50 - (sanic)[INFO]: Starting worker [44832]
2018-12-21 10:59:54 - (network)[INFO][127.0.0.1:55817]: GET http://127.0.0.1:8000/employee/  200 267
2018-12-21 11:00:03 - (network)[INFO][127.0.0.1:55865]: GET http://127.0.0.1:8000/employee/  200 267
2018-12-21 11:01:05 - (network)[INFO][127.0.0.1:56135]: GET http://127.0.0.1:8000/employee/  200 38
2018-12-21 11:01:44 - (network)[INFO][127.0.0.1:56325]: POST http://127.0.0.1:8000/employee/  200 28
2018-12-21 11:02:38 - (network)[INFO][127.0.0.1:56562]: GET http://127.0.0.1:8000/employee/  200 95

具体实现

models.py

# -*- coding: utf-8 -*-
from peewee import *
from playhouse.pool import MySQLDatabase
from playhouse.shortcuts import RetryOperationalError

import config


class RetryMysqlDatabase(RetryOperationalError, MySQLDatabase):
    def __init__(self, database, **kwargs):
        super(MySQLDatabase, self).__init__(database, **kwargs)

    def sequence_exists(self, seq):
        pass


db = RetryMysqlDatabase(
    database=config.DB_NAME, host=config.DB_HOST,
    user=config.DB_USER, passwd=config.DB_PASSWORD, port=config.DB_PORT,
)


class BaseModel(Model):
    """A base model that will use our MySQL database"""
    is_deleted = BooleanField(u'是否删除', default=False)

    class Meta:
        database = db


class Employee(BaseModel):
    # base info
    number = CharField(verbose_name=u'编号', unique=True)
    name = CharField(verbose_name=u'姓名', null=True)
    email = CharField(verbose_name=u'邮箱', null=True)

serialize.py

# -*- coding: utf-8 -*-
from marshmallow import Schema, fields, post_load
import models


class EmployeeSchema(Schema):
    # 基本信息
    number = fields.String()
    name = fields.String()
    email = fields.Email()

    @post_load
    def make_employee(self, data):
        return models.Employee(**data)

views.py

# -*- coding: utf-8 -*-
from sanic.views import HTTPMethodView
from sanic import Blueprint, response

from models import Employee
import serialize

employee_bp = Blueprint("employee_bp")


class EmployeeView(HTTPMethodView):
    async def get(self, request):
        employees = Employee.select().where(Employee.is_deleted == 0)
        schema = serialize.EmployeeSchema()
        data = schema.dump(employees, many=True).data
        return response.json({"code": 200, "msg": "success", "data": data})

    async def post(self, request):
        data = request["POST"]
        employee_schema = serialize.EmployeeSchema()
        _, error = employee_schema.load(data)
        if error:
            msg = ""
            for error_item in error:
                error_detail = error.get(error_item)
                if isinstance(error_detail, list):
                    error_detail = ','.join(error_detail)
                else:
                    error_detail = str(error_detail)
                msg += "{0}: {1}".format(error_item, error_detail)
            return response.json({"code": 500, "msg": msg})
        else:
            number = data.get("number")
            try:
                Employee.get(Employee.number == number)
                return response.json({"code": 500, "msg": "the number is uniq"})
            except Employee.DoesNotExist:
                employee = Employee()
                for key in data:
                    value = data.get(key)
                    if hasattr(employee, key):
                        setattr(employee, key, value)
                employee.save()
        return response.json({"code": 201, "msg": "success"})


employee_bp.add_route(EmployeeView.as_view(), "/")

app.py

# -*- coding: utf-8 -*-
from sanic import Sanic
from sanic_cors import CORS
from aoiklivereload import LiveReloader

from views import employee_bp
from models import db, Employee
import config

app = Sanic(__name__)
app.blueprint(employee_bp, url_prefix='/employee')
CORS(app, automatic_options=True)


@app.middleware('request')
async def transform_data_request(request):
    try:
        request['POST'] = request.json if request.json else {}
        request['GET'] = request.args if request.args else {}
    except Exception as e:
        print(repr(e))


@app.middleware('response')
async def close_db(request, response):
    if not db.is_closed():
        db.close()


# application config
app.config.from_object(config)

# init the database
db.create_tables([Employee, ], safe=True)

if __name__ == '__main__':
    # reload the app
    reloader = LiveReloader()
    reloader.start_watcher_thread()
    # run server debug
    app.run(host='0.0.0.0', port=8000, debug=True)

config.py

# -*- coding: utf-8 -*-
# 关闭长连接
KEEP_ALIVE = False
# 关闭启动logo
LOGO = None

# Mysql Database for dev
DB_HOST = '127.0.0.1'
DB_PORT = 3306
DB_NAME = 'sanic'
DB_USER = 'root'
DB_PASSWORD = '123456'

requirements.txt

# python 3.7.2
PyMySQL==0.7.11
AoikLiveReload==0.1.0
peewee==2.8.5
marshmallow==2.13.6
sanic-crud==0.2.4
Sanic-Cors==0.6.0.0
sanic==0.6.0

start.sh

# dev, demo debug 
python3 app.py

# deploy
gunicorn app:app --bind 0.0.0.0:8000 --worker-class sanic.worker.GunicornWorker