Blog

Looking for a Python web framework that’s tiny enough to fit in a single file yet powerful enough to power production‑grade APIs? Meet Bottle, the lightweight, zero‑dependency micro‑framework that lets you spin up web applications in minutes. Whether you’re a seasoned developer building micro‑services, a hobbyist prototyping a RESTful API, or a student learning the ropes of web development, Bottle offers a clean, intuitive API without the overhead of heavyweight stacks. In this guide we’ll explore what makes Bottle unique, walk through a hands‑on example, and reveal best practices to keep your code lean, fast, and SEO‑friendly.

What Is Bottle?

Bottle is a single‑file Python web framework designed for simplicity. Created by Marcel Hellkamp in 2009, it follows the “micro‑framework” philosophy: core functionality for routing, request/response handling, and templating, with optional plugins for database access, authentication, and more. Because Bottle has no external dependencies, you can drop the bottle.py file into any project and start serving HTTP traffic instantly.

Key Features of the Bottle Lightweight Web Framework

Zero‑Dependency Design

• Runs on any Python 2.7+ or 3.x interpreter without installing extra packages.
• Ideal for embedded devices, serverless functions, or constrained environments.

Built‑in Development Server

• Provides a simple WSGI‑compatible server for quick testing.
• Can be swapped for production‑grade servers like Gunicorn or uWSGI.

Powerful Routing System

• Supports dynamic URL parameters, regex matching, and HTTP method constraints.
• Allows nested routes and route prefixes for clean API design.

Flexible Templating

• Includes a fast built‑in template engine (SimpleTemplate) and supports Jinja2, Mako, etc., via plugins.
• Templates are plain text files that can be rendered with a single function call.

Extensible Plugin Architecture

• Over 30 community‑maintained plugins for databases, authentication, caching, and more.
• Plugins can be added or removed without touching core code.

Installing and Setting Up Bottle

Getting started with Bottle is as easy as copying a file or installing via pip. Choose the method that fits your workflow:

# Option 1: Install with pip (recommended for virtualenv)
pip install bottle

# Option 2: Download the single file
curl -O https://bottlepy.org/bottle.py

Once you have bottle.py in your project directory, you can import it like any other module:

from bottle import route, run, template

Creating Your First Bottle App

Below is a minimal “Hello, World!” application that demonstrates the core concepts of routing, request handling, and template rendering.

from bottle import route, run, template

@route('/hello/<name>')
def greet(name):
    # The name variable is captured from the URL
    return template('Hello {{name}}!', name=name)

# Run the built‑in development server on port 8080
run(host='localhost', port=8080, debug=True)

Save this script as app.py and launch it with python app.py. Navigate to http://localhost:8080/hello/World to see the greeting.

Routing and Request Handling in Depth

Bottle’s routing syntax is concise yet expressive. Here are some common patterns:

• @route('/users') – Handles a static path.
• @route('/users/<id:int>') – Captures an integer id parameter.
• @route('/search', method='POST') – Restricts the route to POST requests.
• @route('/files/<filename:path>') – Captures the rest of the URL as a file path.

Inside a route function you can access request data via the request object:

from bottle import request

@route('/login', method='POST')
def login():
    username = request.forms.get('username')
    password = request.forms.get('password')
    # Perform authentication logic here
    return f'Logged in as {username}'

Templates and Static Files

While Bottle’s built‑in SimpleTemplate engine is sufficient for many projects, you can also integrate Jinja2 for more complex rendering. Here’s a quick example using SimpleTemplate:

# file: views/greeting.tpl
Hello {{name}}! Welcome to the {{framework}} tutorial.

from bottle import template

@route('/welcome/<name>')
def welcome(name):
    return template('views/greeting.tpl', name=name, framework='Bottle')

To serve static assets (CSS, JavaScript, images), map a static route:

@route('/static/<filename:path>')
def static_files(filename):
    return static_file(filename, root='static')

Testing and Debugging Your Bottle Application

Because Bottle follows the WSGI standard, you can test routes with any WSGI‑compatible test client. The built‑in TestApp class simplifies unit testing:

from bottle import Bottle, response
from bottle import TestApp
import unittest

app = Bottle()

@app.route('/add/<a:int>/<b:int>')
def add(a, b):
    return str(a + b)

class BottleTest(unittest.TestCase):
    def setUp(self):
        self.test_app = TestApp(app)

    def test_addition(self):
        resp = self.test_app.get('/add/3/4')
        self.assertEqual(resp.body.decode(), '7')
        self.assertEqual(resp.status_int, 200)

if __name__ == '__main__':
    unittest.main()

Enable debug=True in the run() call to get detailed error pages during development. For production, switch to a robust server like Gunicorn:

gunicorn -w 4 -b 0.0.0.0:8000 myapp:app

When to Choose Bottle Over Other Frameworks

While Flask, Django, and FastAPI dominate the Python web landscape, Bottle shines in specific scenarios:

• Micro‑services or serverless functions where cold‑start time and binary size matter.
• Embedded devices (Raspberry Pi, IoT) that lack the resources for larger frameworks.
• Educational projects where simplicity helps students focus on HTTP fundamentals.
• Quick prototypes that need a single file without a full dependency tree.

If you require built‑in ORM, admin interface, or extensive authentication modules out of the box, a larger framework may be more appropriate. However, Bottle’s extensibility via plugins means you can add only the features you need, keeping your deployment lean.

Performance and Scalability Tips

Even though Bottle is tiny, you can still achieve high throughput by following best practices:

1. Use a production‑grade WSGI server (Gunicorn, uWSGI, or Waitress) instead of the built‑in server.
2. Enable keep‑alive connections to reduce handshake overhead.
3. Leverage asynchronous workers (e.g., uvicorn with asgi compatibility via bottle-asgi plugin) for I/O‑bound workloads.
4. Cache static assets with a reverse proxy like Nginx or Cloudflare.
5. Profile and monitor using tools such as cProfile or APM services to identify bottlenecks.

SEO‑Optimized Content Strategy for Bottle Articles

To ensure your Bottle tutorial ranks well on search engines, incorporate the following SEO tactics:

• Use the primary keyword “Python Bottle lightweight web framework” in the title, first paragraph, and at least one subheading.
• Include related long‑tail keywords like “Bottle micro framework tutorial”, “Bottle vs Flask”, and “Bottle API example”.
• Structure content with clear <h2> and <h3> tags for hierarchy, helping crawlers understand topic relevance.
• Add internal links to other Python web development posts and external links to the official
  

Welcome to the ultimate Python Tornado async framework guide! Whether you’re building real‑time chat apps, streaming APIs, or high‑traffic micro‑services, Tornado’s non‑blocking I/O engine can give you the performance edge you need. In this article we’ll explore Tornado’s core concepts, walk through a step‑by‑step async setup, and share best practices that keep your code clean, scalable, and SEO‑friendly. By the end, you’ll be ready to harness Tornado’s power and deliver lightning‑fast Python web applications.

What Makes Tornado Different?

Tornado is often described as a “Python async framework” because it was designed from the ground up to handle thousands of simultaneous connections without spawning a new thread for each request. Its key differentiators include:

• Non‑blocking network I/O – Powered by an event loop that monitors sockets and callbacks.
• Built‑in WebSocket support – Ideal for real‑time dashboards and chat applications.
• Scalable HTTP server – Handles static files, reverse proxies, and long‑polling out of the box.
• Compatibility with asyncio – Since version 5.0, Tornado can run on Python’s native asyncio loop, making integration with modern async libraries seamless.

Installing Tornado and Setting Up a Project

Step 1: Create a virtual environment

python -m venv venv
source venv/bin/activate   # On Windows: venv\Scripts\activate

Step 2: Install Tornado

pip install tornado

Step 3: Project layout

• app.py – Main entry point.
• handlers/ – Directory for request handlers.
• templates/ – HTML templates (optional).
• static/ – CSS, JavaScript, images.

Writing Your First Async Handler

All Tornado request handlers inherit from tornado.web.RequestHandler. To make a handler asynchronous, you can either use the legacy @tornado.web.asynchronous decorator or, in modern code, define an async def method. Below is a minimal example that demonstrates both approaches.

Legacy callback style

import tornado.ioloop
import tornado.web

class LegacyHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
        # Simulate a long‑running operation
        tornado.ioloop.IOLoop.current().add_timeout(
            tornado.ioloop.timedelta(seconds=2),
            self.on_complete
        )

    def on_complete(self):
        self.write("Response after 2 seconds")
        self.finish()

application = tornado.web.Application([
    (r"/legacy", LegacyHandler),
])

if __name__ == "__main__":
    application.listen(8888)
    tornado.ioloop.IOLoop.current().start()

Modern async/await style

import tornado.ioloop
import tornado.web
import asyncio

class AsyncHandler(tornado.web.RequestHandler):
    async def get(self):
        await asyncio.sleep(2)   # Non‑blocking sleep
        self.write("Async response after 2 seconds")

app = tornado.web.Application([
    (r"/async", AsyncHandler),
])

if __name__ == "__main__":
    app.listen(8888)
    tornado.platform.asyncio.AsyncIOMainLoop().install()
    asyncio.get_event_loop().run_forever()

Notice how the async version reads like synchronous code while still remaining non‑blocking. This is the recommended pattern for new projects.

Integrating WebSockets for Real‑Time Communication

WebSockets are a natural fit for Tornado’s event‑driven model. Below is a concise guide to creating a chat server that broadcasts messages to all connected clients.

WebSocket handler example

import tornado.web
import tornado.websocket
import tornado.ioloop

class ChatSocket(tornado.websocket.WebSocketHandler):
    clients = set()

    def open(self):
        self.clients.add(self)
        self.write_message("Welcome to the Tornado chat!")

    def on_message(self, message):
        for client in self.clients:
            if client != self:
                client.write_message(message)

    def on_close(self):
        self.clients.remove(self)

app = tornado.web.Application([
    (r"/ws", ChatSocket),
])

if __name__ == "__main__":
    app.listen(8889)
    tornado.ioloop.IOLoop.current().start()

By storing active connections in a class‑level set, each incoming message can be broadcast instantly, demonstrating Tornado’s ability to handle thousands of concurrent WebSocket connections with minimal overhead.

Best Practices for Production‑Ready Tornado Apps

• Run behind a reverse proxy (e.g., Nginx) to handle TLS termination, static file caching, and request buffering.
• Use multiple processes – Leverage tornado.process.fork_processes or a process manager like gunicorn with the tornado worker class to utilize all CPU cores.
• Graceful shutdown – Implement signal handlers to close open sockets and finish pending requests before exiting.
• Monitoring and logging – Integrate with structlog or loguru for structured logs; expose Prometheus metrics via a dedicated endpoint.
• Security hardening – Enable xsrf_cookies, set secure_cookie flags, and validate user input rigorously.

Testing Async Tornado Code

Testing asynchronous code can feel tricky, but Tornado provides a convenient AsyncTestCase base class. Here’s a quick example using unittest:

import unittest
import tornado.testing
import tornado.web
import asyncio

class TestAsyncHandler(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/async", AsyncHandler),   # Assume AsyncHandler from earlier
        ])

    @tornado.testing.gen_test
    async def test_response(self):
        response = await self.http_client.fetch(self.get_url("/async"))
        self.assertEqual(response.code, 200)
        self.assertIn(b"Async response", response.body)

if __name__ == "__main__":
    unittest.main()

The @gen_test decorator (or async def with await in newer Tornado versions) ensures the test runner waits for the asynchronous request to complete before asserting results.

Performance Tuning Tips

1. Adjust the I/O loop’s max body size – Use max_body_size in the application settings for large uploads.
2. Enable HTTP/2 – When running behind a supporting proxy, HTTP/2 can reduce latency for many small requests.
3. Cache static assets – Serve CSS/JS via the reverse proxy with long‑term cache headers to offload work from Tornado.
4. Profile with py-spy or cProfile – Identify blocking calls and replace them with async equivalents.
5. Use async database drivers – Libraries like aiopg (PostgreSQL) or motor (MongoDB) keep the event loop free.

When to Choose Tornado Over Other Async Frameworks

Python offers several async web frameworks such as FastAPI, Sanic, and Quart. Tornado shines in scenarios where you need:

• Long‑lived connections (WebSockets, Server‑Sent Events).
• Fine‑grained control over the I/O loop and low‑level networking.
• Compatibility with legacy code that already uses Tornado’s callback style.
• A single, battle‑tested library that bundles HTTP server, WebSocket support, and async primitives.

If your project heavily relies on micro‑services, JSON APIs, and modern async/await syntax, FastAPI might be a better fit. However, for real‑time, high‑concurrency workloads, Tornado remains a top contender.

Conclusion

With its robust non‑blocking architecture, native asyncio integration, and built‑in WebSocket capabilities, Tornado continues to be a powerful choice for developers who demand speed and scalability. This Python Tornado async framework guide covered installation, async handler patterns, real‑time communication, testing, and production best practices. By applying the techniques and tips shared here, you’ll be able to build responsive, high‑throughput web applications that stand up to modern traffic demands. Happy coding, and let Tornado’s storm of performance propel your next Python project forward!

In the world of distributed systems, Python microservices are gaining traction for their simplicity and rapid development cycle. Yet, as services multiply, the need for a fast, type‑safe, and language‑agnostic communication protocol becomes critical. Enter gRPC – Google’s high‑performance, open‑source RPC framework that leverages Protocol Buffers to serialize data efficiently. This tutorial walks you through building a complete Python microservice using gRPC, from environment setup to Docker deployment, giving you the practical knowledge to scale your architecture with confidence.

What Is gRPC and Why It Matters for Python Microservices

gRPC (gRPC Remote Procedure Call) is a modern, HTTP/2‑based framework that enables client and server applications to communicate transparently, as if they were local function calls. Its key advantages for Python microservices include:

• Performance: Binary serialization with Protocol Buffers is far smaller and faster than JSON over HTTP/1.1.
• Strong typing: The .proto contract enforces data schemas at compile time, reducing runtime errors.
• Bi‑directional streaming: Supports real‑time data pipelines, a boon for IoT, chat, or video services.
• Multi‑language support: A single .proto file can generate client/server stubs for Python, Go, Java, Node.js, and more.
• Built‑in load balancing & authentication: gRPC integrates with modern service meshes like Istio.

Prerequisites

Before diving in, make sure you have the following installed on your development machine:

1. Python 3.9+ (python --version)
2. pip package manager
3. Docker (optional but recommended for containerization)
4. Basic familiarity with Protocol Buffers syntax

Step 1: Set Up the Python Project

1.1 Create a virtual environment

python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`

1.2 Install gRPC and protobuf packages

pip install grpcio grpcio-tools

Step 2: Define the Service with Protocol Buffers

All gRPC contracts live in a .proto file. Create a directory called proto and add calculator.proto:

syntax = "proto3";

package calculator;

// The Calculator service definition.
service Calculator {
  // Unary RPC to add two numbers.
  rpc Add (AddRequest) returns (AddResponse);
  // Server‑streaming RPC to generate a Fibonacci series.
  rpc Fibonacci (FibRequest) returns (stream FibResponse);
}

// Request message for Add.
message AddRequest {
  double a = 1;
  double b = 2;
}

// Response message for Add.
message AddResponse {
  double result = 1;
}

// Request message for Fibonacci.
message FibRequest {
  int32 limit = 1; // Number of terms to generate.
}

// Response message for Fibonacci.
message FibResponse {
  int64 number = 1;
}

This file defines a simple Calculator service with a unary Add method and a server‑streaming Fibonacci method.

Step 3: Generate Python Stubs from the .proto File

Run the grpcio-tools compiler to produce the client and server code:

python -m grpc_tools.protoc -I./proto --python_out=./generated --grpc_python_out=./generated ./proto/calculator.proto

The command creates two files in generated/:

• calculator_pb2.py – data structures (messages).
• calculator_pb2_grpc.py – service base classes and client stubs.

Step 4: Implement the gRPC Server

Create a new file server.py and import the generated modules:

import grpc
from concurrent import futures
import time

from generated import calculator_pb2
from generated import calculator_pb2_grpc

class CalculatorServicer(calculator_pb2_grpc.CalculatorServicer):
    def Add(self, request, context):
        result = request.a + request.b
        return calculator_pb2.AddResponse(result=result)

    def Fibonacci(self, request, context):
        a, b = 0, 1
        for _ in range(request.limit):
            yield calculator_pb2.FibResponse(number=a)
            a, b = b, a + b

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    calculator_pb2_grpc.add_CalculatorServicer_to_server(CalculatorServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC server listening on port 50051...")
    try:
        while True:
            time.sleep(86400)  # keep server alive
    except KeyboardInterrupt:
        server.stop(0)

if __name__ == '__main__':
    serve()

Key points:

• We inherit from CalculatorServicer generated by the compiler.
• Add implements a simple unary RPC.
• Fibonacci uses a generator to stream results back to the client.

Step 5: Build the gRPC Client

Create client.py to consume the service:

import grpc
from generated import calculator_pb2
from generated import calculator_pb2_grpc

def run():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = calculator_pb2_grpc.CalculatorStub(channel)

        # Unary call
        add_req = calculator_pb2.AddRequest(a=12.5, b=7.3)
        add_res = stub.Add(add_req)
        print(f"Add result: {add_res.result}")

        # Server‑streaming call
        fib_req = calculator_pb2.FibRequest(limit=10)
        print("Fibonacci series:")
        for fib_res in stub.Fibonacci(fib_req):
            print(fib_res.number, end=' ')
        print()

if __name__ == '__main__':
    run()

Running client.py after starting server.py should display the sum and the first ten Fibonacci numbers.

Step 6: Dockerize the Microservice

6.1 Create a Dockerfile

# Use the official lightweight Python image.
FROM python:3.11-slim

# Set a working directory.
WORKDIR /app

# Copy only requirements first for caching.
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy the source code.
COPY . .

# Expose the gRPC port.
EXPOSE 50051

# Command to run the server.
CMD ["python", "server.py"]

6.2 Prepare requirements.txt

grpcio==1.62.0
grpcio-tools==1.62.0

6.3 Build and Run the Container

# Build the image
docker build -t python-grpc-calculator .

# Run the container
docker run -d -p 50051:50051 --name calc-service python-grpc-calculator

The microservice is now isolated, reproducible, and ready to be orchestrated with Kubernetes or Docker Compose.

Step 7: Testing the gRPC Service

Automated testing ensures reliability across releases. Use pytest with the grpcio-testing package for in‑process server testing.

pip install pytest grpcio-testing

Example test file test_calculator.py:

import pytest
import grpc
from grpc_testing import server_from_dictionary, strict_real_time

from generated import calculator_pb2
from generated import calculator_pb2_grpc
from server import CalculatorServicer

@pytest.fixture
def grpc_server():
    servicers = {
        calculator_pb2_grpc.CalculatorServicer: CalculatorServicer()
    }
    return server_from_dictionary(servicers, strict_real_time())

def test_add(grpc_server):
    stub = calculator_pb2_grpc.CalculatorStub(grpc_server.channel())
    response = stub.Add(calculator_pb2.AddRequest(a=5, b=7))
    assert response.result == 12

def test_fibonacci(grpc_server):
    stub = calculator_pb2_grpc.CalculatorStub(grpc_server.channel())
    responses = list(stub.Fibonacci(calculator_pb2.FibRequest(limit=5)))
    assert [r.number for r in responses] == [0, 1, 1, 2, 3]

Run the suite with pytest -q. Passing tests confirm that both unary and streaming RPCs behave as expected.

Best Practices for Production‑Ready Python gRPC Microservices

• Use TLS encryption: Replace insecure_channel with secure_channel and provide server certificates.
• Implement interceptors: Add logging, metrics (Prometheus), and authentication at the interceptor layer.
• Keep .proto files versioned: Store them in a dedicated repository or submodule to guarantee contract stability.
• Leverage health checking: Implement
  

Generating clean, up‑to‑date API documentation can be a headache, especially when your Python service evolves fast. Fortunately, the OpenAPI specification (formerly Swagger) paired with modern Python tools lets you automate the entire documentation pipeline—no manual markdown files required. In this guide we’ll walk through the most popular libraries, show you how to set up auto‑generation, customize the output, and integrate the docs into your CI/CD workflow, so you can keep developers, testers, and stakeholders on the same page.

Why Choose OpenAPI for Python Projects?

• Standardized format: OpenAPI is language‑agnostic, making your API description reusable across tools, languages, and platforms.
• Rich ecosystem: From interactive Swagger UI to client SDK generators, the spec unlocks a suite of productivity boosters.
• Automation ready: Most Python web frameworks (FastAPI, Flask, Django, Starlette) can introspect routes and generate a JSON/YAML spec automatically.
• Improved developer experience: Auto‑generated docs stay in sync with code, reducing mismatches that cause bugs and wasted time.

Key Python Libraries for OpenAPI Auto Documentation

FastAPI

FastAPI was built around OpenAPI from day one. By simply defining path operations with type hints, it produces a complete OpenAPI schema and an interactive Swagger UI at /docs and ReDoc at /redoc.

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}", tags=["Items"])
def read_item(item_id: int, q: str | None = None):
    """
    Retrieve an item by its ID.
    - **item_id**: Unique identifier of the item.
    - **q**: Optional query string for filtering.
    """
    return {"item_id": item_id, "q": q}

Flask with Flask‑RESTX

Flask doesn’t have built‑in OpenAPI support, but Flask‑RESTX (a maintained fork of Flask‑RESTPlus) adds a declarative @api.doc decorator and a Swagger UI.

from flask import Flask
from flask_restx import Api, Resource, fields

app = Flask(__name__)
api = Api(app, version="1.0", title="My API", description="A simple Flask API")

ns = api.namespace('items', description='Item operations')
item_model = api.model('Item', {
    'id': fields.Integer(required=True, description='Item identifier'),
    'name': fields.String(required=True, description='Item name')
})

@ns.route('/<int:id>')
class Item(Resource):
    @api.doc('get_item')
    @api.marshal_with(item_model)
    def get(self, id):
        """Fetch an item given its identifier"""
        return {'id': id, 'name': f'Item {id}'}

Django with drf‑spectacular

If you’re using Django REST Framework (DRF), drf‑spectacular automatically generates an OpenAPI 3 schema from serializers, viewsets, and routers.

# settings.py
INSTALLED_APPS = [
    # …
    "drf_spectacular",
]

REST_FRAMEWORK = {
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
}

# urls.py
from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView

urlpatterns = [
    # …
    path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
    path('api/docs/', SpectacularSwaggerView.as_view(url_name='schema')),
]

Step‑by‑Step: Auto‑Generating Documentation with FastAPI

1. Install FastAPI and an ASGI server

pip install fastapi uvicorn

2. Define your API with type hints and docstrings

FastAPI reads the function signatures, Pydantic models, and docstrings to build the schema. Use Response models for richer documentation.

3. Run the application

uvicorn main:app --reload

Visit http://127.0.0.1:8000/docs for Swagger UI or /redoc for ReDoc.

4. Export the OpenAPI spec

You can download the JSON spec directly:

curl -s http://127.0.0.1:8000/openapi.json -o openapi.json

5. Generate client SDKs (optional)

Use OpenAPI Generator to produce Python, JavaScript, or TypeScript clients:

openapi-generator-cli generate \
  -i openapi.json \
  -g python \
  -o ./generated-client

Customizing the Generated Documentation

Adding Metadata

Supply a title, description, version, and terms_of_service when creating the FastAPI instance.

app = FastAPI(
    title="Acme Inventory API",
    description="API for managing inventory items, orders, and suppliers.",
    version="2.3.1",
    terms_of_service="https://acme.com/terms/",
    contact={"name": "API Support", "email": "support@acme.com"},
    license_info={"name": "MIT", "url": "https://opensource.org/licenses/MIT"},
)

Tagging Endpoints

Tags group related operations in the UI. Use the tags parameter on route decorators.

@app.get("/orders", tags=["Orders"])
def list_orders():
    ...

Adding Security Schemes

Declare authentication methods (API key, OAuth2, JWT) in the app’s openapi_schema or via the dependencies argument.

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me", dependencies=[Depends(oauth2_scheme)], tags=["Users"])
def read_current_user():
    ...

Extending the Schema with openapi_extra

If you need to add custom fields not covered by FastAPI, use the openapi_extra argument on Pydantic models.

class Item(BaseModel):
    id: int
    name: str

    class Config:
        schema_extra = {
            "example": {"id": 1, "name": "Sample Item"},
            "x-custom-attribute": "Custom value"
        }

Integrating Documentation into CI/CD

• Step 1 – Generate the spec as an artifact: Add a script in your pipeline that starts the app, calls /openapi.json, and saves the file.
• Step 2 – Validate the spec: Use swagger-cli validate openapi.json or spectral lint openapi.json to catch schema errors early.
• Step 3 – Publish the docs: Deploy the generated openapi.json to a static site (GitHub Pages, Netlify) and embed Swagger UI via a simple HTML wrapper.
• Step 4 – Version control: Tag the spec with the same version as your API release; this makes rollback and diffing straightforward.

Sample CI script (GitHub Actions)

name: Generate OpenAPI Spec

on:
  push:
    branches: [main]

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Start API in background
        run: |
          uvicorn main:app --port 8000 &
          echo $! > uvicorn.pid
          sleep 5
      - name: Export OpenAPI JSON
        run: |
          curl -s http://127.0.0.1:8000/openapi.json -o openapi.json
          kill $(cat uvicorn.pid)
      - name: Validate spec
        run: |
          npm install -g @apidevtools/swagger-cli
          swagger-cli validate openapi.json
      - name: Upload artifact
        uses: actions/upload-artifact@v3
        with:
          name: openapi-spec
          path: openapi.json

Common Pitfalls and How to Avoid Them

• Missing type hints: Without explicit types FastAPI falls back to any, resulting in vague schema definitions. Always annotate parameters and return models.
• Out‑of‑date docstrings: Keep inline documentation synchronized with code changes; automated tests that compare the spec against expected examples can catch drift.
• Circular imports: When models reference each other, use from __future__ import annotations or define schemas lazily with ForwardRef.
• Large monolithic specs: Split your API into multiple tags or even separate OpenAPI files and use components to reference shared definitions.

SEO Tips for Your Documentation Page

Even though the primary audience is developers, search engines still rank documentation pages. Follow these quick SEO best practices:

If you’ve ever felt limited by traditional REST endpoints, you know the excitement that comes with GraphQL’s flexible data fetching. In the Python ecosystem, consuming a GraphQL API has become remarkably straightforward thanks to modern libraries like gql and requests. This guide walks you through everything you need to start querying, mutating, and handling GraphQL responses in Python—complete with code snippets, best‑practice tips, and SEO‑friendly keywords to boost your site’s visibility.

Why Consume a GraphQL API with Python?

Python developers choose GraphQL for several compelling reasons:

• Fine‑grained data selection: Retrieve exactly what you need, reducing over‑fetching.
• Single endpoint simplicity: One URL handles queries, mutations, and even subscriptions.
• Strong typing via schema introspection: Tools can auto‑generate type hints, improving IDE support.
• Efficient pagination: Cursor‑based pagination scales better than offset‑based approaches.

When paired with Python’s readability, you get a powerful stack for building data‑driven applications, micro‑services, or data‑science pipelines.

Setting Up Your Python Environment

Install the essential packages

Begin with a clean virtual environment and install the two most popular GraphQL clients:

python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`
pip install gql[all] requests

The gql library provides a high‑level API with built‑in async support, while requests offers a lightweight fallback for quick scripts.

Project structure recommendation

• config.py – store endpoint URLs and authentication tokens.
• client.py – centralize the GraphQL client creation.
• queries/ – keep .graphql files for each query or mutation.
• main.py – orchestrate calls and handle results.

Making Your First Query with gql

Define the query

Save the following as queries/get_user.graphql:

query GetUser($id: ID!) {
  user(id: $id) {
    id
    name
    email
    createdAt
  }
}

Python code to execute the query

from gql import gql, Client
from gql.transport.requests import RequestsHTTPTransport
import os

# Load endpoint and token from environment variables
transport = RequestsHTTPTransport(
    url=os.getenv("GRAPHQL_ENDPOINT"),
    headers={"Authorization": f"Bearer {os.getenv('API_TOKEN')}"},
    verify=True,
    retries=3,
)

client = Client(transport=transport, fetch_schema_from_transport=True)

query = gql(open("queries/get_user.graphql").read())

variables = {"id": "12345"}

result = client.execute(query, variable_values=variables)
print(result["user"])

This snippet demonstrates the core workflow: load the query, inject variables, and print the parsed JSON response. The fetch_schema_from_transport=True flag introspects the schema once, enabling auto‑completion in IDEs.

Working with Mutations and Variables

Mutation example: creating a new post

Save queries/create_post.graphql:

mutation CreatePost($input: CreatePostInput!) {
  createPost(input: $input) {
    post {
      id
      title
      publishedAt
    }
    errors {
      field
      message
    }
  }
}

Execute the mutation

mutation = gql(open("queries/create_post.graphql").read())

variables = {
    "input": {
        "title": "GraphQL with Python",
        "content": "A comprehensive guide on consuming GraphQL APIs using Python.",
        "authorId": "12345"
    }
}

response = client.execute(mutation, variable_values=variables)

if response["createPost"]["errors"]:
    print("❌ Mutation failed:", response["createPost"]["errors"])
else:
    print("✅ New post ID:", response["createPost"]["post"]["id"])

Notice the nested error handling – a best practice for production‑grade GraphQL consumption.

Authentication, Headers, and Secure Requests

Most GraphQL APIs protect data with JWTs, API keys, or OAuth2. Below are two common patterns.

Static token in headers

transport = RequestsHTTPTransport(
    url="https://api.example.com/graphql",
    headers={"Authorization": f"Bearer {API_TOKEN}"},
)

Dynamic token refresh (OAuth2)

import requests
from gql.transport.requests import RequestsHTTPTransport

def get_access_token():
    resp = requests.post(
        "https://auth.example.com/oauth/token",
        data={"grant_type": "client_credentials"},
        auth=("client_id", "client_secret")
    )
    return resp.json()["access_token"]

def create_transport():
    token = get_access_token()
    return RequestsHTTPTransport(
        url="https://api.example.com/graphql",
        headers={"Authorization": f"Bearer {token}"},
    )

client = Client(transport=create_transport(), fetch_schema_from_transport=True)

Refresh the token before each request or use a background thread for long‑running scripts.

Pagination Strategies for Large Result Sets

GraphQL commonly uses cursor‑based pagination. Here’s how to iterate through all items safely.

query = gql("""
query ListArticles($first: Int!, $after: String) {
  articles(first: $first, after: $after) {
    edges {
      node {
        id
        title
      }
      cursor
    }
    pageInfo {
      hasNextPage
      endCursor
    }
  }
}
""")

def fetch_all_articles(limit=100):
    after = None
    all_articles = []

    while True:
        vars = {"first": limit, "after": after}
        resp = client.execute(query, variable_values=vars)
        edges = resp["articles"]["edges"]
        all_articles.extend([e["node"] for e in edges])

        if not resp["articles"]["pageInfo"]["hasNextPage"]:
            break
        after = resp["articles"]["pageInfo"]["endCursor"]

    return all_articles

articles = fetch_all_articles()
print(f"Fetched {len(articles)} articles")

By checking pageInfo.hasNextPage, you avoid infinite loops and respect server limits.

Error Handling and Debugging Tips

• GraphQL errors vs. HTTP errors: A 200 status can still contain a errors array in the JSON payload.
• Validate queries locally: Use graphql-core to parse and validate before sending.
• Enable logging: Set gql.logging.enable() to capture request/response traces.
• Retry strategy: Combine requests.adapters.HTTPAdapter with exponential backoff for transient network issues.

Advanced Tips: Caching, Subscriptions, and Async Usage

Caching responses

For read‑heavy workloads, wrap the client with a simple in‑memory cache:

from functools import lru_cache

@lru_cache(maxsize=256)
def cached_execute(query_str, variables):
    query = gql(query_str)
    return client.execute(query, variable_values=variables)

Subscriptions (real‑time updates)

If your GraphQL server supports WebSocket subscriptions, switch to the WebsocketsTransport:

from gql.transport.websockets import WebsocketsTransport

ws_transport = WebsocketsTransport(url="wss://api.example.com/graphql")
async with Client(transport=ws_transport, fetch_schema_from_transport=True) as async_client:
    async for result in async_client.subscribe(gql("subscription { newMessage { id text } }")):
        print("🔔 New message:", result["newMessage"])

Async/await for high‑throughput scripts

When calling many queries concurrently, leverage Python’s asyncio loop:

import asyncio
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport

async def fetch_user(user_id):
    async with Client(transport=AIOHTTPTransport(url=ENDPOINT), fetch_schema_from_transport=True) as session:
        query = gql("query($id: ID!){ user(id:$id){ name } }")
        result = await session.execute(query, variable_values={"id": user_id})
        return result["user"]["name"]

async def main():
    ids = ["1","2","3","4","5"]
    names = await asyncio.gather(*(fetch_user(uid) for uid in ids))
    print(names)

asyncio.run(main())

Testing Your GraphQL Client

Write unit tests with responses or