Welcome to aio-openapi¶
Asynchronous web middleware for aiohttp and serving Rest APIs with OpenAPI v 3 specification and with optional PostgreSql database bindings.
Current version is 2.1.2.
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
Redoc document rendering (like https://api.metablock.io/v1/docs)
Optional sentry middleware
Getting Started¶
The main implementation:
import uuid
from aiohttp import web
from openapi import sentry
from openapi.db import get_db
from openapi.middleware import json_error
from openapi.spec import Redoc
from openapi.rest import rest
from .db import meta
from .endpoints import routes
from .ws import ws_routes
def create_app():
return rest(redoc=Redoc(), setup_app=setup_app)
def setup_app(app: web.Application) -> None:
db = get_db(app)
meta(db.metadata)
app.middlewares.append(json_error())
app.middlewares.append(
sentry.middleware(app, f"https://{uuid.uuid4().hex}@sentry.io/1234567", "test")
)
app.router.add_routes(routes)
if __name__ == "__main__":
create_app().main()
The endpoint implementation:
from typing import List
from aiohttp import web
from sqlalchemy.sql.expression import null
from openapi.db.path import SqlApiPath
from openapi.spec import op
from .models import (
Task,
TaskAdd,
TaskOrderableQuery,
TaskPathSchema,
TaskQuery,
TaskUpdate
)
routes = web.RouteTableDef()
@routes.view("/tasks")
class TasksPath(SqlApiPath):
"""
---
summary: Create and query Tasks
tags:
- Task
"""
table = "tasks"
def filter_done(self, op, value):
done = self.db_table.c.done
return done != null() if value else done == null()
@op(query_schema=TaskOrderableQuery, response_schema=List[Task])
async def get(self):
"""
---
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):
"""
---
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)
@op(query_schema=TaskQuery)
async def delete(self):
"""
---
summary: Delete Tasks
description: Delete a group of Tasks
responses:
204:
description: Tasks successfully deleted
"""
await self.delete_list(query=dict(self.request.query))
return web.Response(status=204)
@routes.view("/tasks/{id}")
class TaskPath(SqlApiPath):
"""
---
summary: Create and query tasks
tags:
- name: Task
description: Simple description
- name: Random
description: Random description
"""
table = "tasks"
path_schema = TaskPathSchema
@op(response_schema=Task)
async def get(self):
"""
---
summary: Retrieve a Task
description: Retrieve a Task by ID
responses:
200:
description: the task
"""
data = await self.get_one()
return self.json_response(data)
@op(response_schema=Task, body_schema=TaskUpdate)
async def patch(self):
"""
---
summary: Update a Task
description: Update an existing Task by ID
responses:
200:
description: the updated task
"""
data = await self.update_one()
return self.json_response(data)
@op()
async def delete(self):
"""
---
summary: Delete a Task
description: Delete an existing task
responses:
204:
description: Task successfully deleted
"""
await self.delete_one()
return web.Response(status=204)
The data model implementation
