Sanic-fast asynchronous web flime

Introduction to Sanic

Sanic — a modern asynchronous web framework for Python, designed for building high‑performance web applications. Built on asyncio, it can efficiently handle thousands of concurrent connections thanks to native async/await support. Sanic is ideal for creating REST APIs, microservices, real‑time applications, and high‑load back‑ends.

Key Features of Sanic

Performance and Architecture

Sanic is engineered with speed in mind. The framework leverages asyncio for asynchronous request handling, achieving performance comparable to Node.js and Go. Built‑in async/await support makes code more readable and efficient.

Functional Capabilities

• Full async/await support out of the box
• Routing with parameters and middleware
• Integrated WebSocket support
• Template engine and static file handling
• CORS, cookie, and session support
• Simple and intuitive project structure
• Excellent scalability for API servers

Installation and Initial Setup

Basic Installation

pip install sanic

Additional Dependencies

pip install sanic[ext]  # Extensions
pip install jinja2 aiofiles python-dotenv

Creating Your First Application

from sanic import Sanic
from sanic.response import text

app = Sanic("MyApp")

@app.get("/")
async def index(request):
    return text("Hello from Sanic")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

Routing and Request Handling

Basic Routes

@app.get("/")
async def get_handler(request):
    return text("GET request")

@app.post("/")
async def post_handler(request):
    return text("POST request")

@app.put("/")
async def put_handler(request):
    return text("PUT request")

@app.delete("/")
async def delete_handler(request):
    return text("DELETE request")

Asynchronous Routes

import asyncio

@app.get("/async")
async def async_route(request):
    await asyncio.sleep(1)  # Simulate async operation
    return text("Asynchronous response")

Working with Parameters and Data

Route Parameters

@app.get("/user/<user_id:int>")
async def get_user(request, user_id):
    return text(f"User ID: {user_id}")

@app.get("/profile/<username:str>")
async def get_profile(request, username):
    return text(f"Profile: {username}")

Query Parameters

@app.get("/search")
async def search(request):
    query = request.args.get("q", "nothing")
    limit = request.args.get("limit", "10")
    return text(f"Search: {query}, limit: {limit}")

JSON and Form Handling

from sanic.response import json

@app.post("/api/data")
async def receive_data(request):
    data = request.json
    return json({"received": data})

@app.post("/form")
async def handle_form(request):
    username = request.form.get("username")
    email = request.form.get("email")
    return json({"username": username, "email": email})

Response Types

Various Response Formats

from sanic.response import text, json, html, redirect, file

@app.get("/text")
async def text_response(request):
    return text("Simple text")

@app.get("/json")
async def json_response(request):
    return json({"message": "JSON response"})

@app.get("/html")
async def html_response(request):
    return html("<h1>HTML page</h1>")

@app.get("/redirect")
async def redirect_response(request):
    return redirect("/new-page")

@app.get("/file")
async def file_response(request):
    return await file("path/to/file.pdf")

Error Handling

Built‑in Exceptions

from sanic.exceptions import NotFound, SanicException, Unauthorized

@app.exception(NotFound)
async def handle_404(request, exception):
    return json({"error": "Page not found"}, status=404)

@app.exception(Exception)
async def handle_generic_error(request, exception):
    return json({"error": "Internal server error"}, status=500)

Custom Exceptions

class CustomException(SanicException):
    status_code = 400
    message = "Custom error"

@app.exception(CustomException)
async def handle_custom_error(request, exception):
    return json({"error": exception.message}, status=exception.status_code)

Middleware and Lifecycle

Request and Response Middleware

@app.middleware("request")
async def before_request(request):
    request.ctx.start_time = time.time()
    print(f"Request to {request.path}")

@app.middleware("response")
async def after_response(request, response):
    response.headers["X-Framework"] = "Sanic"
    response.headers["X-Process-Time"] = str(time.time() - request.ctx.start_time)

Lifecycle Hooks

@app.before_server_start
async def setup_db(app, loop):
    print("Connecting to the database")

@app.after_server_stop
async def close_db(app, loop):
    print("Closing database connection")

Cookies and Sessions

Cookies

@app.get("/set-cookie")
async def set_cookie(request):
    response = text("Cookie set")
    response.cookies["session_id"] = "123456"
    response.cookies["session_id"]["httponly"] = True
    response.cookies["session_id"]["secure"] = True
    return response

@app.get("/get-cookie")
async def get_cookie(request):
    session_id = request.cookies.get("session_id", "not set")
    return text(f"Session ID: {session_id}")

Header Management

@app.get("/headers")
async def handle_headers(request):
    user_agent = request.headers.get("User-Agent")
    auth_header = request.headers.get("Authorization")
    
    response = json({"user_agent": user_agent})
    response.headers["Custom-Header"] = "Sanic-Value"
    return response

File Uploads

File Handling

import os

@app.post("/upload")
async def upload_file(request):
    file = request.files.get("file")
    if file:
        upload_dir = "./uploads"
        os.makedirs(upload_dir, exist_ok=True)
        
        file_path = os.path.join(upload_dir, file.name)
        with open(file_path, "wb") as f:
            f.write(file.body)
        
        return json({"message": "File uploaded", "filename": file.name})
    return json({"error": "File not found"}, status=400)

Templates and Static Files

Jinja2 Integration

from sanic_ext import Extend
from jinja2 import Environment, FileSystemLoader

app = Sanic("TemplateApp")
Extend(app)

@app.get("/template")
async def render_template(request):
    return await app.ext.render("index.html", context={"name": "Sanic"})

Static Files

app.static("/static", "./static")
app.static("/media", "./media", name="media")

WebSocket Support

Basic WebSocket

@app.websocket("/ws")
async def websocket_handler(request, ws):
    while True:
        try:
            message = await ws.recv()
            await ws.send(f"Echo: {message}")
        except:
            break

WebSocket with Authentication

@app.websocket("/ws/auth")
async def auth_websocket(request, ws):
    token = request.headers.get("Authorization")
    if not token:
        await ws.close(code=1008, reason="Unauthorized")
        return
    
    while True:
        message = await ws.recv()
        await ws.send(f"Authenticated echo: {message}")

Database Integration

Using SQLAlchemy

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///./database.db")
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession)

@app.middleware("request")
async def inject_session(request):
    request.ctx.session = SessionLocal()

@app.middleware("response")
async def close_session(request, response):
    await request.ctx.session.close()

Working with Async ORM

@app.get("/users")
async def get_users(request):
    async with request.ctx.session as session:
        # Example ORM query
        users = await session.execute("SELECT * FROM users")
        return json([dict(user) for user in users])

Project Structure

Recommended Organization

project/
├── app.py              # Main application
├── config.py           # Configuration
├── requirements.txt    # Dependencies
├── routes/            # Routes
│   ├── __init__.py
│   ├── api.py
│   └── auth.py
├── models/            # Data models
│   ├── __init__.py
│   └── user.py
├── middleware/        # Middleware
│   ├── __init__.py
│   └── auth.py
├── templates/         # Templates
│   └── index.html
├── static/           # Static assets
│   ├── css/
│   └── js/
└── tests/            # Tests
    ├── __init__.py
    └── test_app.py

Blueprints for Modularity

Creating a Blueprint

from sanic import Blueprint

# routes/users.py
users_bp = Blueprint("users", url_prefix="/users")

@users_bp.get("/")
async def get_users(request):
    return json({"users": []})

@users_bp.get("/<user_id:int>")
async def get_user(request, user_id):
    return json({"user_id": user_id})

# app.py
from routes.users import users_bp
app.blueprint(users_bp)

Testing

Writing Tests

import pytest
from app import app

@pytest.mark.asyncio
async def test_index():
    request, response = await app.asgi_client.get("/")
    assert response.status == 200
    assert "Hello from Sanic" in response.text

@pytest.mark.asyncio
async def test_json_endpoint():
    request, response = await app.asgi_client.get("/api/data")
    assert response.status == 200
    assert response.json["status"] == "ok"

Testing with Fixtures

@pytest.fixture
def test_client():
    return app.asgi_client

@pytest.mark.asyncio
async def test_with_fixture(test_client):
    request, response = await test_client.post("/api/users", json={"name": "Test"})
    assert response.status == 201

Deployment and Production

Production Settings

# config.py
class ProductionConfig:
    DEBUG = False
    ACCESS_LOG = False
    WORKERS = 4
    HOST = "0.0.0.0"
    PORT = 8000

# app.py
from config import ProductionConfig

if __name__ == "__main__":
    app.run(
        host=ProductionConfig.HOST,
        port=ProductionConfig.PORT,
        workers=ProductionConfig.WORKERS,
        debug=ProductionConfig.DEBUG,
        access_log=ProductionConfig.ACCESS_LOG
    )

Running from the Command Line

# Basic run
sanic app.app --host=0.0.0.0 --port=8000

# Production with workers
sanic app.app --host=0.0.0.0 --port=8000 --workers=4

# Additional options
sanic app.app --workers=4 --access-log --debug

Using Gunicorn

pip install gunicorn
gunicorn app:app --bind 0.0.0.0:8000 --worker-class sanic.worker.GunicornWorker

Complete Table of Sanic Methods and Functions

Advanced Features

Custom Decorators

from functools import wraps

def require_auth(f):
    @wraps(f)
    async def decorated_function(request, *args, **kwargs):
        auth = request.headers.get("Authorization")
        if not auth:
            return json({"error": "Unauthorized"}, status=401)
        return await f(request, *args, **kwargs)
    return decorated_function

@app.get("/protected")
@require_auth
async def protected_route(request):
    return json({"message": "Protected content"})

Data Validation

from cerberus import Validator

@app.post("/api/users")
async def create_user(request):
    schema = {
        'name': {'type': 'string', 'required': True},
        'email': {'type': 'string', 'required': True, 'regex': r'^[^@]+@[^@]+\.[^@]+$'},
        'age': {'type': 'integer', 'min': 0, 'max': 150}
    }
    
    v = Validator(schema)
    if not v.validate(request.json):
        return json({"errors": v.errors}, status=400)
    
    # Process validated data
    return json({"message": "User created"}, status=201)

Practical Usage Examples

RESTful API

users_data = []

@app.get("/api/users")
async def get_users(request):
    return json(users_data)

@app.post("/api/users")
async def create_user(request):
    user = request.json
    user['id'] = len(users_data) + 1
    users_data.append(user)
    return json(user, status=201)

@app.get("/api/users/<user_id:int>")
async def get_user(request, user_id):
    user = next((u for u in users_data if u['id'] == user_id), None)
    if not user:
        return json({"error": "User not found"}, status=404)
    return json(user)

WebSocket Chat

connected_clients = set()

@app.websocket("/chat")
async def chat_handler(request, ws):
    connected_clients.add(ws)
    try:
        async for message in ws:
            # Broadcast to all connected clients
            for client in connected_clients.copy():
                try:
                    await client.send(message)
                except:
                    connected_clients.remove(client)
    finally:
        connected_clients.discard(ws)

Performance Optimization

High‑Load Settings

# Optimized configuration
app.config.update({
    "KEEP_ALIVE_TIMEOUT": 30,
    "REQUEST_TIMEOUT": 60,
    "RESPONSE_TIMEOUT": 60,
    "REQUEST_MAX_SIZE": 100_000_000,  # 100 MB
    "ACCESS_LOG": False,
    "AUTO_RELOAD": False,
})

Connection Pooling

import aioredis
from aiohttp import ClientSession

@app.before_server_start
async def setup_connections(app, loop):
    # Redis connection pool
    app.ctx.redis = await aioredis.create_redis_pool(
        'redis://localhost',
        maxsize=10
    )
    
    # HTTP client session
    app.ctx.http_session = ClientSession()

@app.after_server_stop
async def cleanup_connections(app, loop):
    app.ctx.redis.close()
    await app.ctx.redis.wait_closed()
    await app.ctx.http_session.close()

Security

CORS Settings

from sanic_cors import CORS

CORS(app, resources={
    r"/api/*": {
        "origins": ["http://localhost:3000"],
        "methods": ["GET", "POST", "PUT", "DELETE"],
        "allow_headers": ["Content-Type", "Authorization"]
    }
})

Rate Limiting

from collections import defaultdict
import time

request_counts = defaultdict(list)

@app.middleware("request")
async def rate_limit(request):
    client_ip = request.ip
    now = time.time()
    
    # Clean up old requests
    request_counts[client_ip] = [
        req_time for req_time in request_counts[client_ip]
        if now - req_time < 60  # Last minute
    ]
    
    # Enforce limit
    if len(request_counts[client_ip]) >= 100:  # 100 requests per minute
        return json({"error": "Rate limit exceeded"}, status=429)
    
    request_counts[client_ip].append(now)

Frequently Asked Questions

What is Sanic and what is it used for?

Sanic — a high‑performance asynchronous web framework for Python that uses async/await syntax. It is intended for building fast APIs, web services, and microservices capable of handling a large number of concurrent connections.

How does Sanic differ from Flask and Django?

Unlike Flask and Django, Sanic is fully asynchronous, delivering significantly higher throughput under heavy request loads. It also includes built‑in WebSocket support and a modern API‑first development approach.

Can Sanic be used in production?

Yes, Sanic is production‑ready. Many companies run high‑traffic applications with it. For production, it is recommended to use multiple workers and tune performance‑related settings.

Does Sanic support database integration?

Sanic works seamlessly with async ORMs and database drivers such as SQLAlchemy (with asyncio), asyncpg for PostgreSQL, aiomysql for MySQL, and other async‑compatible libraries.

How can I ensure security in Sanic applications?

Sanic supports standard security measures: HTTPS, CORS, JWT authentication, input validation, rate limiting, and middleware for access‑control checks.

What testing tools are available for Sanic?

Sanic provides a built‑in test client, integrates well with pytest for asynchronous tests, and works smoothly with popular Python testing libraries.

$