基于Sanic的微服务基础架构

sanic-ms:`https://github.com/songcser/sanic-ms`

Example:`https://github.com/songcser/sanic-ms/tree/master/examples`

@app.listener('before_server_start')
async def before_srver_start(app, loop):
    queue = asyncio.Queue()
    app.queue = queue
    loop.create_task(consume(queue, app.config.ZIPKIN_SERVER))
    reporter = AioReporter(queue=queue)
    tracer = BasicTracer(recorder=reporter)
    tracer.register_required_propagators()
    opentracing.tracer = tracer
    app.db = await ConnectionPool(loop=loop).init(DB_CONFIG)

@app.middleware('request')
async def cros(request):
    if request.method == 'POST' or request.method == 'PUT':
        request['data'] = request.json
    span = before_request(request)
    request['span'] = span
@app.middleware('response')
async def cors_res(request, response):
    span = request['span'] if 'span' in request else None
    if response is None:
        return response
    result = {'code': 0}
    if not isinstance(response, HTTPResponse):
        if isinstance(response, tuple) and len(response) == 2:
            result.update({
                'data': response[0],
                'pagination': response[1]
            })
        else:
            result.update({'data': response})
        response = json(result)
        if span:
            span.set_tag('http.status_code', "200")
    if span:
        span.set_tag('component', request.app.name)
        span.finish()
    return response

async def async_request(datas):
    # async handler request
    results = await asyncio.gather(*[data[2] for data in datas])
    for index, obj in enumerate(results):
        data = datas[index]
        data[0][data[1]] = results[index]
@user_bp.get('/<id:int>')
@doc.summary("get user info")
@doc.description("get user info by id")
@doc.produces(Users)
async def get_users_list(request, id):
    async with request.app.db.acquire(request) as cur:
        record = await cur.fetch(
            """ SELECT * FROM users WHERE id = $1 """, id)
        datas = [
            [record, 'city_id', get_city_by_id(request, record['city_id'])]
            [record, 'role_id', get_role_by_id(request, record['role_id'])]
        ]
        await async_request(datas)
        return record

Peewee is a simple and small ORM. It has few (but expressive) concepts, making it easy to learn and intuitive to use。

# models.py
class Users(Model):
    id = PrimaryKeyField()
    create_time = DateTimeField(verbose_name='create time',
                                default=datetime.datetime.utcnow)
    name = CharField(max_length=128, verbose_name="user's name")
    age = IntegerField(null=False, verbose_name="user's age")
    sex = CharField(max_length=32, verbose_name="user's sex")
    city_id = IntegerField(verbose_name='city for user', help_text=CityApi)
    role_id = IntegerField(verbose_name='role for user', help_text=RoleApi)
    class Meta:
        db_table = 'users'
# migrations.py
from sanic_ms.migrations import MigrationModel, info, db
class UserMigration(MigrationModel):
    _model = Users
    # @info(version="v1")
    # def migrate_v1(self):
    #     migrate(self.add_column('sex'))
def migrations():
    try:
        um = UserMigration()
        with db.transaction():
            um.auto_migrate()
            print("Success Migration")
    except Exception as e:
        raise e
if __name__ == '__main__':
    migrations()

peewee：http://docs.peewee-orm.com/en/latest/

asyncpg is the fastest driver among common Python, NodeJS and Go implementations

sql = "SELECT * FROM users WHERE name=$1"
name = "test"
async with request.app.db.acquire(request) as cur:
    data = await cur.fetchrow(sql, name)
async with request.app.db.transaction(request) as cur:
    data = await cur.fetchrow(sql, name)

asyncpg:https://github.com/MagicStack/asyncpg

benchmarks:https://magic.io/blog/asyncpg-1m-rows-from-postgres-to-python/

Don’t create a session per request. Most likely you need a session per application which performs all requests altogether.
A session contains a connection pool inside, connection reusage and keep-alives (both are on by default) may speed up total performance.

@app.listener('before_server_start')
async def before_srver_start(app, loop):
    app.client =  Client(loop, url='http://host:port')
async def get_role_by_id(request, id):
    cli = request.app.client.cli(request)
    async with cli.get('/cities/{}'.format(id)) as res:
        return await res.json()
@app.listener('before_server_stop')
async def before_server_stop(app, loop):
    app.client.close()

aiohttp:http://aiohttp.readthedocs.io/en/stable/client.html

Enter OpenTracing: by offering consistent, expressive, vendor-neutral APIs for popular platforms, OpenTracing makes it easy for developers to add (or switch) tracing implementations with an O(1) configuration change. OpenTracing also offers a lingua franca for OSS instrumentation and platform-specific tracing helper libraries. Please refer to the Semantic Specification.

@logger(type='method', category='test', detail='detail', description="des", tracing=True, level=logging.INFO)
async def get_city_by_id(request, id):
    cli = request.app.client.cli(request)

opentracing:https://github.com/opentracing/opentracing-python

zipkin:https://github.com/openzipkin/zipkin

jaeger:https://uber.github.io/jaeger/

from sanic_ms import doc
@user_bp.post('/')
@doc.summary('create user')
@doc.description('create user info')
@doc.consumes(Users)
@doc.produces({'id': int})
async def create_user(request):
    data = request['data']
    async with request.app.db.transaction(request) as cur:
        record = await cur.fetchrow(
            """ INSERT INTO users(name, age, city_id, role_id)
                VALUES($1, $2, $3, $4, $5)
                RETURNING id
            """, data['name'], data['age'], data['city_id'], data['role_id']
        )
        return {'id': record['id']}

swagger:https://swagger.io/

from sanic_ms.tests import APITestCase
from server import app
class TestCase(APITestCase):
    _app = app
    _blueprint = 'visit'
    def setUp(self):
        super(TestCase, self).setUp()
        self._mock.get('/cities/1',
                       payload={'id': 1, 'name': 'shanghai'})
        self._mock.get('/roles/1',
                       payload={'id': 1, 'name': 'shanghai'})
    def test_create_user(self):
        data = {
            'name': 'test',
            'age': 2,
            'city_id': 1,
            'role_id': 1,
        }
        res = self.client.create_user(data=data)
        body = ujson.loads(res.text)
        self.assertEqual(res.status, 200)

coverage erase
coverage run --source . -m sanic_ms tests
coverage xml -o reports/coverage.xml
coverage2clover -i reports/coverage.xml -o reports/clover.xml
coverage html -d reports

unittest:https://docs.python.org/3/library/unittest.html
coverage:https://coverage.readthedocs.io/en/coverage-4.4.1/

from sanic_ms.exception import ServerError
@visit_bp.delete('/users/<id:int>')
async def del_user(request, id):
    raise ServerError(error='内部错误',code=10500, message="msg")
code: 错误码，无异常时为0，其余值都为异常
message: 状态码信息
error: 自定义错误信息
status_code: http状态码，使用标准的http状态码