How to make your Python code fast and asynchronous with Sanic

In anticipation of the start of the " Python Developer. Professional " course, we have prepared a traditional translation of a useful article.

Hello everyone, in this article I will talk about creating simple asynchronous projects on the Sanic framework.

Introduction

Sanic – Flask - - Python 10 , . async/await, Python 3.5, .

Sanic , .

– HTTP-, , .

, , , .

: github. . 

• Python3.6+

• pipenv ( )

• PostgreSQL ( , MySQL SQLite)

• secure – , cookie - Python.

• environs – Python . , .

• sanic-envconfig – , Sanic.

• databases – Python, SQLAlchemy Core PostgreSQL, MySQL SQLite.

Pipfile.

pipenv  -- python python3.6

, pipenv.

pipenv install sanic secure environs sanic-envconfig

:

pipenv install databases[postgresql]

postgresql, mysql, sqlite.

, .

├── .env
├── Pipfile
├── Pipfile.lock
├── setup.py
└── project
    ├── __init__.py
    ├── __main__.py
    ├── main.py
    ├── middlewares.py
    ├── routes.py
    ├── settings.py
    └── tables.py

setup.py, project .  

from setuptools import setup

setup(
    name='project',
)

…

pipenv install -e .

.env , . 

_main_.py , project .

pipenv run python -m project

main.py

from project.main import init

init()

. init main.py.

from sanic import Sanic

app = Sanic(__name__)

def init():
    app.run(host='0.0.0.0', port=8000, debug=True)

app Sanic, , , debug.

…

pipenv run python -m project

Sanic . http://0.0.0.0:8000, :

Error: Requested URL / not found

, . .

. .env, Sanic.

.env

DEBUG=True
HOST=0.0.0.0
POST=8000

…

from sanic import Sanic
from environs import Env
from project.settings import Settings
app = Sanic(__name__)
def init():
    env = Env()
    env.read_env()
    
    app.config.from_object(Settings)
    app.run(
        host=app.config.HOST, 
        port=app.config.PORT, 
        debug=app.config.DEBUG,
        auto_reload=app.config.DEBUG,    
    )

settings.py

from sanic_envconfig import EnvConfig


class Settings(EnvConfig):
    DEBUG: bool = True
    HOST: str = '0.0.0.0'
    PORT: int = 8000

, auto_reload, .

.

, . 

databases asyncpg, PostgreSQL. . .

Sanic, . , :

• afterserverstart

• afterserverstop

main.py

from sanic import Sanic
from databases import Database
from environs import Env
from project.settings import Settings
app = Sanic(__name__)
def setup_database():
    app.db = Database(app.config.DB_URL)

    @app.listener('after_server_start')
    async def connect_to_db(*args, **kwargs):
        await app.db.connect()

    @app.listener('after_server_stop')
    async def disconnect_from_db(*args, **kwargs):
        await app.db.disconnect()
def init():
    env = Env()
    env.read_env()
    
    app.config.from_object(Settings)
    setup_database()
    app.run(
        host=app.config.HOST, 
        port=app.config.PORT, 
        debug=app.config.DEBUG,
        auto_reload=app.config.DEBUG,    
    )

-. DB_URL .

.env

DEBUG=True
HOST=0.0.0.0
POST=8000
DB_URL=postgresql://postgres:postgres@localhost/postgres

settings.py:

from sanic_envconfig import EnvConfig


class Settings(EnvConfig):
    DEBUG: bool = True
    HOST: str = '0.0.0.0'
    PORT: int = 8000
    DB_URL: str = ''

, DB_URL . app.db. .

, SQL-.

tables.py SQLAlchemy.

import sqlalchemy


metadata = sqlalchemy.MetaData()

books = sqlalchemy.Table(
    'books',
    metadata,
    sqlalchemy.Column('id', sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column('title', sqlalchemy.String(length=100)),
    sqlalchemy.Column('author', sqlalchemy.String(length=60)),
)

, books . Alembic – , SQLAlchemy Python.

SQLAlchemy. .

# Executing many
query = books.insert()
values = [
    {"title": "No Highway", "author": "Nevil Shute"},
    {"title": "The Daffodil", "author": "SkyH. E. Bates"},
]
await app.db.execute_many(query, values)

# Fetching multiple rows
query = books.select()
rows = await app.db.fetch_all(query)

# Fetch single row
query = books.select()
row = await app.db.fetch_one(query)

. routes.py books.

from sanic.response import json
from project.tables import books

def setup_routes(app):
    @app.route("/books")
    async def book_list(request):
        query = books.select()
        rows = await request.app.db.fetch_all(query)
        return json({
            'books': [{row['title']: row['author']} for row in rows]
        })

, , init setup_routes.

from project.routes import setup_routes
app = Sanic(__name__)
def init():
    ...
    app.config.from_object(Settings)
    setup_database()
    setup_routes(app)
    ...

…

$ curl localhost:8000/books
{"books":[{"No Highway":"Nevil Shute"},{"The Daffodil":"SkyH. E. Bates"}]}

Middleware

, -?

$ curl -I localhost:8000
Connection: keep-alive
Keep-Alive: 5
Content-Length: 32
Content-Type: text/plain; charset=utf-8

, . , X-XSS-Protection, Strict-Transport-Security .. secure.

middlewares.py

from secure import SecureHeaders

secure_headers = SecureHeaders()

def setup_middlewares(app):
    @app.middleware('response')
    async def set_secure_headers(request, response):
        secure_headers.sanic(response)

middleware main.py:

from project.middlewares import setup_middlewares
app = Sanic(__name__)
def init():
    ...
    app.config.from_object(Settings)
    setup_database()
    setup_routes(app)
    setup_middlewares(app)
    ...

:

$ curl -I localhost:8000/books
Connection: keep-alive
Keep-Alive: 5
Strict-Transport-Security: max-age=63072000; includeSubdomains
X-Frame-Options: SAMEORIGIN
X-XSS-Protection: 1; mode=block
X-Content-Type-Options: nosniff
Referrer-Policy: no-referrer, strict-origin-when-cross-origin
Pragma: no-cache
Expires: 0
Cache-control: no-cache, no-store, must-revalidate, max-age=0
Content-Length: 32
Content-Type: text/plain; charset=utf-8

, github. , Sanic. Sanic , .

. " Python C: Python " .