Skip to main content
Donate to the Python Software Foundation or Purchase a PyCharm License to Benefit the PSF! Donate Now

Minimal OpenAPI asynchronous server application

Project description


PyPI version Python versions CircleCI codecov

This library is an asynchronous web middleware for aiohttp for serving Rest APIs with OpenAPI v 3 specification and with optional PostgreSql database.

Table of Contents


pip install aio-openapi


Clone the repository and create a virtual environment venv.

Install dependencies by running the install script


To run tests

pytest --cov


  • Asynchronous web routes with aiohttp
  • Data validation, serialization and unserialization with python dataclasses
  • OpenApi v 3 auto documentation
  • SqlAlchemy expression language
  • Asynchronous DB interaction with asyncpg
  • Migrations with alembic
  • SqlAlchemy tables as python dataclasses
  • Support click command line interface
  • Optional sentry middleware

Web App

To create an openapi RESTful application follow this schema (lets call the file

from import rest

def create_app():
    return rest(
            title='A REST API',

def setup_app(app):
    return app

if __name__ == '__main__':

The create_app function creates the aiohttp server application by invoking the rest function. This function adds the click command in the cli mapping entry and add documentation for routes which support OpenAPI docs. The setup_app function is used to actually setup the custom application, usually by adding middleware, routes, shutdown callbacks, database integration and so forth.

OpenAPI Documentation

The library provide tools for creating OpenAPI v 3 compliant endpoints and auto-document them.

An example from test tests/example directory

from aiohttp import web

from openapi.db.path import SqlApiPath
from openapi.spec import op

routes = web.RouteTableDef()

class TasksPath(SqlApiPath):
    summary: Create and query Tasks
        - Task
    table = 'tasks'

    @op(query_schema=TaskOrderableQuery, response_schema=[Task])
    async def get(self):
        summary: Retrieve Tasks
        description: Retrieve a list of Tasks
                description: Authenticated tasks
        paginated = await self.get_list()
        return paginated.json_response()

    @op(response_schema=Task, body_schema=TaskAdd)
    async def post(self):
        summary: Create a Task
        description: Create a new Task
                description: the task was successfully added
                description: Failed validation
        data = await self.create_one()
        return self.json_response(data, status=201)

Database Integration

This library provides integration with asyncpg, an high performant asynchronous connector with PostgreSql database. To add the database extension simply use the get_db function in the applicatiuon setup_app function:

from openapi.db import get_db

def setup_app(app):
    db = get_db(app)
    meta = db.metadata

This will enable database connection and command line tools (most of them from alembic):

python db --help

The database container is available at the db app key:



This library provides a simple distributed websocket utility for creating websocket remote procedure calls (RPC) and pub/sub.

from aiohttp import web

from import Sockets

app = web.Application()
app['web_sockets'] = Sockets(app)

RPC protocol

The RPC protocol has the following structure for incoming messages

    "id": "abc",
    "method": "rpc_method_name",
    "payload": {

The id is used by clients to link the request with the corresponding response. The response for an RPC call is eitrher a success

    "id": "abc",
    "method": "rpc_method_name",
    "result": {

or error

    "id": "abc",
    "method": "rpc_method_name":
    "error": {


To subscribe to messages, one need to use the Subscribe mixin with the subscribe RPC handler. Messages take the form:

    "channel": "channel_name",
    "event": "event_name",
    "data": {

Environment Variables

Several environment variables are used by the library to support testing and deployment.

  • DATASTORE: PostgreSql connection string (same as SqlAlchemy syntax)
  • DBPOOL_MIN_SIZE: minimum size of database connection pool (default is 10)
  • DBPOOL_MAX_SIZE: maximum size of database connection pool (default is 10)

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date
aio_openapi-1.3.6-py3-none-any.whl (46.7 kB) Copy SHA256 hash SHA256 Wheel py3
aio-openapi-1.3.6.tar.gz (53.5 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page