Design and Implementation of a High‑Performance Python Microservice Framework Based on Sanic

The article explains why Python web development often struggles with performance (the C10K problem) and how modern asynchronous frameworks such as Sanic, built on top of asyncio, can provide Flask‑like simplicity while delivering high throughput.

Key features include the use of Sanic with uvloop as the event‑loop engine, asyncpg for fast PostgreSQL access, aiohttp as an HTTP client, Peewee for lightweight model definition, OpenTracing for distributed tracing, Swagger for API documentation, and unittest with mock for unit testing.

Server‑side initialization creates a queue for tracing spans, registers an OpenTracing tracer, and builds a database connection pool:

@app.listener('before_server_start')
async def before_server_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)

Middleware wraps requests and responses, extracts JSON payloads for POST/PUT, creates tracing spans, and normalises the response format to a JSON object containing a code field and optional data or pagination fields:

@app.middleware('request')
async def request_middleware(request):
    if request.method in ('POST', 'PUT'):
        request['data'] = request.json
    span = before_request(request)
    request['span'] = span

@app.middleware('response')
async def response_middleware(request, response):
    span = request.get('span')
    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')
        span.set_tag('component', request.app.name)
        span.finish()
    return response

Asynchronous request handling demonstrates parallel execution of I/O‑bound calls using asyncio.gather :

async def async_request(datas):
    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('/
')
@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

Model design & ORM uses Peewee solely for model definition and migration, while actual queries are performed with asyncpg for maximum performance:

# models.py
class Users(Model):
    id = PrimaryKeyField()
    create_time = DateTimeField(default=datetime.datetime.utcnow)
    name = CharField(max_length=128)
    age = IntegerField()
    sex = CharField(max_length=32)
    city_id = IntegerField()
    role_id = IntegerField()
    class Meta:
        db_table = 'users'

# migrations.py
from sanic_ms.migrations import MigrationModel, info, db
class UserMigration(MigrationModel):
    _model = Users

def migrations():
    try:
        um = UserMigration()
        with db.transaction():
            um.auto_migrate()
            print('Success Migration')
    except Exception as e:
        raise e

if __name__ == '__main__':
    migrations()

Database operations illustrate direct use of asyncpg for fast queries, showing both non‑transactional acquire() for reads and transactional transaction() for writes, and note a TODO for read/write splitting.

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)

Client side wraps aiohttp in a reusable client instance to avoid per‑request session creation, enabling keep‑alive connections across microservices:

@app.listener('before_server_start')
async def before_server_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(f'/roles/{id}') as res:
        return await res.json()

@app.listener('before_server_stop')
async def before_server_stop(app, loop):
    app.client.close()

Logging & Distributed Tracing defines a @logger decorator to attach metadata (type, category, detail, description, tracing flag, log level) to each method, and explains how OpenTracing records a span for every request, later converted to Zipkin format for aggregation.

@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)
    ...

API definition uses the doc helper to generate Swagger‑compatible specifications directly from route functions, automatically inferring request/response schemas from Peewee models.

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) RETURNING id",
            data['name'], data['age'], data['city_id'], data['role_id']
        )
        return {'id': record['id']}

Response handling advises returning plain data objects; the middleware will wrap them into a unified JSON response format.

Unit testing employs unittest with a custom MockClient to stub external service calls, demonstrating how to set up test data and assert responses.

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)

Code coverage shows typical commands to generate XML and HTML reports:

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

Exception handling replaces the default error handler with a custom one that raises a ServerError containing an error code, message, and HTTP status.

from sanic_ms.exception import ServerError
@visit_bp.delete('/users/
')
async def del_user(request, id):
    raise ServerError(error='内部错误', code=10500, message='msg')

The article concludes with author attribution and a note that the content originates from a CSDN blog post.