Skip to content
Open API utilities
Branch: master
Clone or download
Latest commit bb7dbf5 Apr 19, 2019
Type Name Latest commit message Commit time
Failed to load latest commit information.
.circleci 3.7 test passage Nov 4, 2018
.vscode log tests Feb 14, 2019
openapi added post process Apr 15, 2019
.gitignore regresison for 3.6 Nov 4, 2018
LICENSE released to pypi May 27, 2018 released to pypi May 27, 2018
Makefile regresison for 3.6 Nov 4, 2018 response Apr 19, 2019
setup.cfg added post process Apr 15, 2019


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",
    "response": {

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)
You can’t perform that action at this time.