Reference¶
Data¶
DataView¶
- class openapi.data.view.DataView[source]¶
Utility class for data with a valid Supported Schema
- cleaned(schema, data, *, multiple=False, strict=True, Error=None)[source]¶
Clean data using a given schema
- Parameters
- Return type
- dump(schema, data)[source]¶
Dump data using a given a valid Supported Schema, if the schema is None it returns the same data as the input.
- get_schema(schema=None)[source]¶
Get the Supported Schema. If not found it raises an exception
- raise_validation_error(message='', errors=None)[source]¶
Raise an
aiohttp.web.HTTPUnprocessableEntity
- Return type
- validation_error(message='', errors=None)[source]¶
Create the validation exception used by
raise_validation_error()
- Return type
TypeInfo¶
- class openapi.utils.TypingInfo(element: Any, container: Optional[type] = None)[source]¶
Information about a type annotation
- element: Any¶
Alias for field number 0
- classmethod get(value)[source]¶
Create a
TypingInfo
from a typing annotation or another typing info- Parameters
value (
Any
) – typing annotation- Return type
Data Fields¶
- openapi.data.fields.data_field(required=False, validator=None, dump=None, format=None, description=None, items=None, post_process=None, ops=(), hidden=False, meta=None, **kwargs)[source]¶
Extend a dataclass field with the following metadata
- Parameters
validator (
Optional
[Callable
[[Field
,Any
],Any
]]) – optional callable which accept field and raw value as inputs and return the validated valuerequired (
bool
) – boolean specifying if field is requireddump (
Optional
[Callable
[[Any
],Any
]]) – optional callable which receive the field value and convert to the desired value to serve in requestsformat (
Optional
[str
]) – optional string which represents the JSON schema formatitems (
Optional
[Field
]) – field for items of the current field (only used for List and Dict fields)post_process (
Optional
[Callable
[[Any
],Any
]]) – post processor function executed after validationops (
Tuple
) – optional tuple of strings specifying available operationshidden (
bool
) – when True the field is not added to the Openapi documentationmeta (
Optional
[Dict
[str
,Any
]]) – optional dictionary with additional metadata
- Return type
String field¶
Bool field¶
- openapi.data.fields.bool_field(**kw)[source]¶
Specialized
data_field()
for bool types- Return type
UUID field¶
Numeric field¶
- openapi.data.fields.number_field(min_value=None, max_value=None, precision=None, **kw)[source]¶
A specialized
data_field()
for numeric values
Integer field¶
Email field¶
Enum field¶
- openapi.data.fields.enum_field(EnumClass, **kw)[source]¶
A specialized
data_field()
for enums- Parameters
EnumClass – enum for validation
- Return type
Date field¶
- openapi.data.fields.date_field(**kw)[source]¶
A specialized
data_field()
for dates- Return type
Datetime field¶
- openapi.data.fields.date_time_field(timezone=False, **kw)[source]¶
A specialized
data_field()
for datetimes- Parameters
timezone – timezone for validation
- Return type
JSON field¶
- openapi.data.fields.json_field(**kw)[source]¶
A specialized
data_field()
for JSON data- Return type
Data Validation¶
Validate¶
The entry function to validate input data and return a python representation.
The function accept as input a valid type annotation or a TypingInfo
object.
- openapi.data.validate.validate(schema, data, *, strict=True, multiple=False, raise_on_errors=False, items=None, as_schema=False)[source]¶
Validate data with a given schema
- Parameters
schema (
Any
) – a typing annotation or aTypingInfo
objectdata (
Any
) – a data object to validate against the schemastrict (
bool
) – if True validation is strict, i.e. missing required parameters will cause validation to failsmultiple (
bool
) – allow parameters to have multiple valuesraise_on_errors (
bool
) – when True failure of validation will result in a ValidationErrors error, otherwise aValidatedData
object is returned.items (
Optional
[Field
]) – an optional Field for items in a composite type (List or Dict)as_schema (
bool
) – return the schema object rather than simple data type (dataclass rather than dict for example)
- Return type
Validate Schema¶
Same as the validate()
but returns the validation schema object rather than
simple data types (this is mainly different for dataclasses)
- openapi.data.validate.validated_schema(schema, data, *, multiple=False, strict=True)[source]¶
Validate data with a given schema and return a valid representation of the data as a schema instance
- Parameters
schema (
Any
) – a valid Supported Schema or aTypingInfo
objectdata (
Any
) – a data object to validate against the schemastrict (
bool
) – if True validation is strict, i.e. missing required parameters will cause validation to fails
- Return type
Dataclass from db table¶
- openapi.data.db.dataclass_from_table(name, table, *, exclude=None, include=None, default=False, required=False, ops=None)[source]¶
Create a dataclass from an
sqlalchemy.schema.Table
- Parameters
name (
str
) – dataclass nametable (
Table
) – sqlalchemy tableexclude (
Optional
[Sequence
[str
]]) – fields to exclude from the dataclassinclude (
Optional
[Sequence
[str
]]) – fields to include in the dataclassdefault (
Union
[bool
,Sequence
[str
]]) – use columns defaults in the dataclassrequired (
Union
[bool
,Sequence
[str
]]) – set non nullable columns without a default as required fields in the dataclassops (
Optional
[Dict
[str
,Sequence
[str
]]]) – additional operation for fields
- Return type
Dump data¶
- openapi.data.dump.dump(schema, data)[source]¶
Dump data with a given schema.
- Parameters
schema (
Any
) – a valid Supported Schemadata (
Any
) – data to dump, if dataclasses are part of the schema, the dump metadata function will be used if available (seedata_field()
)
- Return type
Openapi Specification¶
OpenApiInfo¶
OpenApiSpec¶
- class openapi.spec.OpenApiSpec(info=<factory>, default_content_type='application/json', default_responses=<factory>, security=<factory>, servers=<factory>, validate_docs=False, allowed_tags=<factory>, spec_url='/spec', redoc=None)[source]¶
Open API Specification
- info: openapi.spec.spec.OpenApiInfo¶
openapi info object, it provides metadata about the API
- redoc: Optional[openapi.spec.redoc.Redoc] = None¶
Optional object for rendering the specification as an HTML page via redoc
op decorator¶
Decorator for specifying schemas at route/method level. It is used by both the business logic as well the auto-documentation.
Redoc¶
Allow to add redoc redering to your api.
- class openapi.spec.Redoc(path='/docs', favicon_url='https://raw.githubusercontent.com/Redocly/redoc/master/demo/favicon.png', redoc_js_url='https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js', font='family=Montserrat:300,400,700|Roboto:300,400,700')[source]¶
A dataclass for redoc rendering
DB¶
This module provides integration with SqlAlchemy asynchronous engine for postgresql. The connection string supported is of this type only:
postgresql+asyncpg://<db_user>:<db_password>@<db_host>:<db_port>/<db_name>
Database¶
- class openapi.db.container.Database(dsn='', metadata=None)[source]¶
A container for tables in a database and a manager of asynchronous connections to a postgresql database
- Parameters
dsn (
str
) – Data source name used for database connectionsmetadata (
Optional
[MetaData
]) –sqlalchemy.schema.MetaData
containing tables
- property metadata: sqlalchemy.sql.schema.MetaData¶
The
sqlalchemy.schema.MetaData
containing tables- Return type
- property engine: sqlalchemy.ext.asyncio.engine.AsyncEngine¶
The
sqlalchemy.ext.asyncio.AsyncEngine
creating connection and transactions- Return type
- property sync_engine: sqlalchemy.engine.base.Engine¶
The
sqlalchemy.engine.Engine
for synchrouns operations- Return type
- __getattr__(name)[source]¶
Retrive a
sqlalchemy.schema.Table
from metadata tables
- transaction()[source]¶
Context manager for initializing an asynchronous database transaction
- Return type
- ensure_transaction(conn=None)[source]¶
Context manager for ensuring we a connection has initialized a database transaction
- Return type
- ensure_connection(conn=None)¶
Context manager for ensuring we a connection has initialized a database transaction
- Return type
CrudDB¶
Database container with CRUD operations. Used extensively by the SqlApiPath
routing class.
- class openapi.db.dbmodel.CrudDB(dsn='', metadata=None)[source]¶
A
Database
with additional methods for CRUD operations- async db_count(table, filters=None, *, conn=None, consumer=None)[source]¶
Count rows in a table
- Parameters
table (
Table
) – sqlalchemy Tablefilters (
Optional
[Dict
]) – key-value pairs for filtering rowsconn (
Optional
[AsyncConnection
]) – optional db connectionconsumer (
Optional
[Any
]) – optional consumer (seeget_query()
)
- Return type
- async db_delete(table, filters, *, conn=None, consumer=None)[source]¶
Delete rows from a given table
- Parameters
table (
Table
) – sqlalchemy Tablefilters (
Dict
) – key-value pairs for filtering rowsconn (
Optional
[AsyncConnection
]) – optional db connectionconsumer (
Optional
[Any
]) – optional consumer (seeget_query()
)
- Return type
- async db_insert(table, data, *, conn=None)[source]¶
Perform an insert into a table
- Parameters
- Return type
- async db_select(table, filters, *, conn=None, consumer=None)[source]¶
Select rows from a given table
- Parameters
table (
Table
) – sqlalchemy Tablefilters (
Dict
) – key-value pairs for filtering rowsconn (
Optional
[AsyncConnection
]) – optional db connectionconsumer (
Optional
[Any
]) – optional consumer (seeget_query()
)
- Return type
- async db_update(table, filters, data, *, conn=None, consumer=None)[source]¶
Perform an update of rows
- Parameters
table (
Table
) – sqlalchemy Tablefilters (
Dict
) – key-value pairs for filtering rows to updatedata (
Dict
) – key-value pairs for updating columns values of selected rowsconn (
Optional
[AsyncConnection
]) – optional db connectionconsumer (
Optional
[Any
]) – optional consumer (seeget_query()
)
- Return type
- async db_upsert(table, filters, data=None, *, conn=None, consumer=None)[source]¶
Perform an upsert for a single record
- Parameters
table (
Table
) – sqlalchemy Tablefilters (
Dict
) – key-value pairs for filtering rows to updatedata (
Optional
[Dict
]) – key-value pairs for updating columns values of selected rowsconn (
Optional
[AsyncConnection
]) – optional db connectionconsumer (
Optional
[Any
]) – optional consumer (seeget_query()
)
- Return type
- default_filter_field(field, op, value)[source]¶
Applies a filter on a field.
Notes on ‘ne’ op:
Example data: [None, ‘john’, ‘roger’] ne:john would return only roger (i.e. nulls excluded) ne: would return john and roger
Notes on ‘search’ op:
For some reason, SQLAlchemy uses to_tsquery rather than plainto_tsquery for the match operator
to_tsquery uses operators (&, |, ! etc.) while plainto_tsquery tokenises the input string and uses AND between tokens, hence plainto_tsquery is what we want here
For other database back ends, the behaviour of the match operator is completely different - see: http://docs.sqlalchemy.org/en/rel_1_0/core/sqlelement.html
get_db¶
- openapi.db.get_db(app, store_url=None)[source]¶
Create an Open API db handler and set it for use in an aiohttp application
- Parameters
This function 1) adds the database to the aiohttp application at key “db”, 2) add the db command to the command line client (if command is True) and 3) add the close handler on application shutdown
SingleConnDatabase¶
A CrudDB
container for testing database driven Rest APIs.
Routes¶
ApiPath¶
- class openapi.spec.path.ApiPath(request)[source]¶
A
DataView
class for OpenAPI path- insert_data(data, *, multiple=False, strict=True, body_schema='body_schema')[source]¶
Validate data for insertion
if a
path_schema
is given, it validate the request match_info against it and add it to the validated data.- Parameters
data (
Any
) – object to be validated against the body_schema, usually obtained from the request body (JSON)multiple (
bool
) – multiple values for a given key are acceptablestrict (
bool
) – all required attributes in schema must be availablebody_schema (
Union
[str
,List
[Type
],Type
]) – the schema to validate against
- Return type
- get_filters(*, query=None, query_schema='query_schema')[source]¶
Collect a dictionary of filters from the request query string. If
path_schema
is defined, it collects filter data from path variables as well.
SqlApiPath¶
- class openapi.db.path.SqlApiPath(request)[source]¶
An
ApiPath
backed by an SQL model.This class provides utility methods for all CRUD operations.
- property db: openapi.db.dbmodel.CrudDB¶
Database connection pool
- Return type
- property db_table: sqlalchemy.sql.schema.Table¶
Default database table for this route.
Obtained from the
table
attribute.- Return type
- async get_list(*, filters=None, query=None, table=None, query_schema='query_schema', dump_schema='response_schema', conn=None)[source]¶
Get a list of models
- Parameters
filters (
Optional
[Dict
[str
,Any
]]) – dictionary of filters, if not provided it will be created from the query_schemaquery (
Optional
[Dict
[str
,Any
]]) – additional query parameters, only used when filters is not providedtable (
Optional
[Table
]) – sqlalchemy table, if not provided the defaultdb_table
is used insteadconn (
Optional
[AsyncConnection
]) – optional db connection
- Return type
- async create_one(*, data=None, table=None, body_schema='body_schema', dump_schema='response_schema', conn=None)[source]¶
Create a new database model
- async create_list(*, data=None, table=None, body_schema='body_schema', dump_schema='response_schema', conn=None)[source]¶
Create multiple models
- async get_one(*, filters=None, query=None, table=None, query_schema='query_schema', dump_schema='response_schema', conn=None)[source]¶
Get a single model
- Parameters
filters (
Optional
[Dict
[str
,Any
]]) – dictionary of filters, if not provided it will be created from the query_schemaquery (
Optional
[Dict
[str
,Any
]]) – additional query parameters, only used when filters is not providedtable (
Optional
[Table
]) – sqlalchemy table, if not provided the defaultdb_table
is used insteadconn (
Optional
[AsyncConnection
]) – optional db connection
- async update_one(*, data=None, filters=None, query=None, table=None, body_schema='body_schema', query_schema='query_schema', dump_schema='response_schema', conn=None)[source]¶
Update a single model
- async delete_one(*, filters=None, query=None, table=None, query_schema='query_schema', conn=None)[source]¶
Delete a single model
- Return type
Websocket¶
Websocket RPC¶
SocketsManager¶
- class openapi.ws.manager.SocketsManager[source]¶
A base class for websocket managers
- property sockets: Set[openapi.ws.manager.Websocket]¶
Set of connected
Websocket
- property channels: openapi.ws.channels.Channels¶
Pub/sub
Channels
currently active on the running pod
- async publish(channel, event, body)[source]¶
Publish an event to a channel
- Property channel
the channel to publish to
- Property event
the event in the channel
- Property body
the body of the event to broadcast in the channel
This method should raise
CannotPublish
if not possible to publish- Return type
- async subscribe(channel)[source]¶
Subscribe to a channel
This method should raise
CannotSubscribe
if not possible to publish- Return type
- async subscribe_to_event(channel, event)[source]¶
Callback when a subscription to an event is done
- Property channel
the channel to publish to
- Property event
the event in the channel
You can use this callback to perform any backend subscriptions to third-party streaming services if required.
By default it does nothing.
- Return type
Channels¶
Channel¶
WsPathMixin¶
- class openapi.ws.path.WsPathMixin[source]¶
Api Path mixin for Websocket RPC protocol
- SOCKETS_KEY = 'web_sockets'¶
Key in the app where the Web Sockets manager is located
- property sockets: openapi.ws.manager.SocketsManager¶
Connected websockets
- Return type
- property channels: openapi.ws.channels.Channels¶
Channels for pub/sub
- Return type
Subscribe¶
- class openapi.ws.pubsub.Subscribe[source]¶
Mixin which implements the subscribe and unsubscribe RPC methods
Must be used as mixin of
WsPathMixin