aio-openapi 1.8.6

Minimal OpenAPI asynchronous server application

aio-openapi

Asynchronous web middleware for aiohttp and serving Rest APIs with OpenAPI v 3 specification and with optional PostgreSql database bindings.

Table of Contents

• Installation
• Development
• Features
• Web App
• OpenAPI Documentation
• Database Integration
• Websockets
  • RPC protocol
  • Publish/Subscribe
• Environment Variables

Installation

pip install aio-openapi

Development

Clone the repository and create a virtual environment venv.

Install dependencies by running the install script

./dev/install

To run tests

pytest --cov

Features

• 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 main.py)

from openapi.rest import rest

def create_app():
    return rest(
        openapi=dict(
            title='A REST API',
            ...
        ),
        base_path='/v1',
        allowed_tags=[...],
        validate_docs=True,
        setup_app=setup_app,
        commands=[...]
    )


def setup_app(app):
    app.router.add_routes(...)
    return app


if __name__ == '__main__':
    create_app().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 typing import List

from aiohttp import web

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


routes = web.RouteTableDef()


@routes.view('/tasks')
class TasksPath(SqlApiPath):
    """
    ---
    summary: Create and query Tasks
    tags:
        - name: Task
          description: Task tag description
    """
    table = 'tasks'

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

    @op(response_schema=Task, body_schema=TaskAdd)
    async def post(self) -> web.Response:
        """
        ---
        summary: Create a Task
        description: Create a new Task
        responses:
            201:
                description: the task was successfully added
            422:
                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 aiohttp import web

from openapi.db import get_db

def setup_app(app: web.Application) -> None:
    db = get_db(app)
    meta = db.metadata

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

python main.py db --help

The database container is available at the db app key:

app['db']

Websockets

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

from aiohttp import web

from openapi.ws 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": {
        ...
    }
}

Publish/Subscribe

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)