API

General

class aiopenapi3.OpenAPI(url, document, session_factory=<class 'httpx.AsyncClient'>, loader=None, plugins=None, use_operation_tags=True)

property _: OperationIndex: the sad smiley interface access operations by operationId

authenticate(*args, **kwargs)

authenticate, multiple authentication schemes can be used simultaneously serving “or” or “and” authentication schemes

Parameters:

args – None to remove all credentials / reset the authorizations
kwargs – scheme=value

static cache_load(path, plugins=None, session_factory=None)

read a pickle api object from path and init the schema types

Parameters:: path (Path) – cache path
Return type:: OpenAPI

cache_store(path)

write the pickled api object to Path to dismiss potentially local defined objects loader, plugins and the session_factory are dropped

Parameters:: path (Path) – cache path
Return type:: None

clone(baseurl=None)

shallwo copy the api object optional set a base url

Parameters:: baseurl (Optional[URL])
Return type:: OpenAPI

createRequest(operationId)

create a Request

lookup the Operation by operationId or path,method

the type of Request returned depends on the session_factory of the OpenAPI object and OpenAPI/Swagger version

Parameters:: operationId (str | tuple[str, HTTPMethodType]) – the operationId or tuple(path,method)
Return type:: Union[Request, Request]
Returns:: the returned Request is either aiopenapi3.request.RequestBase or - in case of a httpx.AsyncClient session_factory - aiopenapi3.request.AsyncRequestBase

async classmethod load_async(url, session_factory=<class 'httpx.AsyncClient'>, loader=None, plugins=None, use_operation_tags=False)

Create an asynchronous OpenAPI object from a description document.

Parameters:

url (str) – the url of the description document
session_factory (Callable[..., AsyncClient]) – used to create the session for http/s io
loader (Optional[Loader]) – the backend to access referenced description documents
plugins (Optional[list[Plugin]]) – potions to cure defects in the description document or requests/responses
use_operation_tags (bool) – honor tags

Return type:

OpenAPI

classmethod load_file(url, path, session_factory=<class 'httpx.AsyncClient'>, loader=None, plugins=None, use_operation_tags=False)

Create an OpenAPI object from a description document file.

from pathlib import Path
import aiopenapi3
import httpx

loader = aiopenapi3.FileSystemLoader(pathlib.Path("<root>"))
api = aiopenapi3.OpenAPI.load_file(
    url="<live-url>",
    path=pathlib.Path("<path-relative-to-root>"),
    loader=loader,
    session_factory=httpx.Client
    )

Parameters:

url (str) – the fictive url of the description document
path (str | Path | URL) – description document location
session_factory (Callable[..., AsyncClient | Client]) – used to create the session for http/s io, defaults to use an AsyncClient
loader (Optional[Loader]) – the backend to access referenced description documents
plugins (Optional[list[Plugin]]) – potions to cure defects in the description document or requests/responses
use_operation_tags (bool) – honor tags

Return type:

OpenAPI

classmethod load_sync(url, session_factory=<class 'httpx.Client'>, loader=None, plugins=None, use_operation_tags=False)

Create a synchronous OpenAPI object from a description document.

Parameters:

url – the url of the description document
session_factory (Callable[..., Client]) – used to create the session for http/s io
loader (Optional[Loader]) – the backend to access referenced description documents
plugins (Optional[list[Plugin]]) – potions to cure defects in the description document or requests/responses
use_operation_tags (bool) – honor tags

Return type:

OpenAPI

classmethod loads(url, data, session_factory=<class 'httpx.AsyncClient'>, loader=None, plugins=None, use_operation_tags=False)

Parameters:

url (str) – the url of the description document
data (str) – description document
session_factory (Callable[..., AsyncClient | Client]) – used to create the session for http/s io, defaults to use an AsyncClient
loader (Optional[Loader]) – the backend to access referenced description documents
plugins (Optional[list[Plugin]]) – potions to cure defects in the description document or requests/responses
use_operation_tags (bool) – honor tags

Return type:

OpenAPI

raise_on_http_status: list[tuple[type[Exception], tuple[int, int]]]: Raise for http status code

Requests

Inheritance diagram of aiopenapi3.v20.glue.Request, aiopenapi3.v30.glue.Request, aiopenapi3.v20.glue.AsyncRequest, aiopenapi3.v30.glue.AsyncRequest

Requests encapsulate the required information to call an operation. They

compile the actual HTTP request to be sent, including authentication information, path, headers, parameters and the body.

send it

receive the result

process it

and return the model of the data

class aiopenapi3.request.RequestBase(api, method, path, operation, servers)

__call__(*args, return_headers=False, context=None, **kwargs)

Parameters:

args
return_headers (bool) – if set return a tuple (header, body)
kwargs

Return type:

Union[dict[str, JSON], list[JSON], str, int, float, bool, None, tuple[ResponseHeadersType, JSON]]

Returns:

body or (header, body)

abstract property data: Schema | Schema | Schema | None

Returns:: the Schema for the body

operation: Union[Operation, Operation, Operation]: associated OpenAPI Operation

abstract property parameters: list[ParameterType]

Returns:: list of aiopenapi3.base.ParameterBase which can be used to inspect the required/optional parameters of the requested Operation

request(data=None, parameters=None, context=None)

Sends an HTTP request as described by this Path

Parameters:

data (any, should match content/type) – The request body to send.
parameters (dict{str: str}) – The path/header/query/cookie parameters required for the operation
context (Any) – The request context for use in aiopenapi3.plugin.Message

Return type:

Response

Returns:

headers, data, response

root: API document root

stream(data=None, parameters=None, context=None)

Sends an HTTP request as described by this Path - but do not process the result

returns a tuple of Schema, httpx.Client, httpx.Response
requires closing the Client when done processing the response
requires manual processing of the data
intended for use with of large results
httpx response streaming via Response.iter_bytes()
combine with ijson coroutines

Parameters:

data (any, should match content/type) – The request body to send.
parameters (dict{str: str}) – The path/header/query/cookie parameters required for the operation

Return type:

StreamResponse

Returns:

schema, session, response

class aiopenapi3.request.AsyncRequestBase(api, method, path, operation, servers)

async __call__(*args, return_headers=False, context=None, **kwargs)

Parameters:

args
return_headers (bool) – if set return a tuple (header, body)
kwargs

Return type:

Union[dict[str, JSON], list[JSON], str, int, float, bool, None, tuple[dict[str, str], JSON]]

Returns:

body or (header, body)

abstract property data: Schema | Schema | Schema | None

Returns:: the Schema for the body

operation: Union[Operation, Operation, Operation]: associated OpenAPI Operation

abstract property parameters: list[ParameterType]

Returns:: list of aiopenapi3.base.ParameterBase which can be used to inspect the required/optional parameters of the requested Operation

async request(data=None, parameters=None, context=None)

Sends an HTTP request as described by this Path

Parameters:

data (any, should match content/type) – The request body to send.
parameters (dict{str: str}) – The path/header/query/cookie parameters required for the operation
context (Any) – The request context for use in aiopenapi3.plugin.Message

Return type:

Response

Returns:

headers, data, response

root: API document root

async stream(data=None, parameters=None, context=None)

Sends an HTTP request as described by this Path - but do not process the result

returns a tuple of Schema, httpx.Client, httpx.Response
requires closing the Client when done processing the response
requires manual processing of the data
intended for use with of large results
httpx response streaming via Response.iter_bytes()
combine with ijson coroutines

Parameters:

data (any, should match content/type) – The request body to send.
parameters (dict{str: str}) – The path/header/query/cookie parameters required for the operation

Return type:

StreamResponse

Returns:

schema, session, response

The different major versions of the OpenAPI protocol require their own Request/AsyncRequest.

class aiopenapi3.v20.glue.Request(api, method, path, operation, servers)

class aiopenapi3.v20.glue.AsyncRequest(api, method, path, operation, servers)

class aiopenapi3.v30.glue.Request(api, method, path, operation, servers)

class aiopenapi3.v30.glue.AsyncRequest(api, method, path, operation, servers)

OperationIndex

class aiopenapi3.request.OperationIndex(api, use_operation_tags)

__getattr__(item)

the sad smiley interface

Parameters:: item (str) – the operationId
Return type:: Union[Request, Request]
Returns:

__getitem__(item)

index operator interface access operations by operationId or (path, method)

Parameters:: item (str | tuple[str, HTTPMethodType]) – operationId or tuple of path & method
Return type:: Union[Request, Request]
Returns:

Parameters

Parameters are part of the operation specification and can be in

path e.g. /users/{name}
query e.g. /users?limit=50
header
cookie

Used to compile a aiopenapi3.request.RequestBase, not meant to be dealt with besides debugging.

Parameter

class aiopenapi3.base.ParameterBase

class aiopenapi3.v20.parameter.Parameter(**data)

Describes a single operation parameter.

allowEmptyValue: bool | None

collectionFormat: str | None

default: Any

description: str | None

enum: Optional[Any]

exclusiveMaximum: bool | None

exclusiveMinimum: bool | None

extensions: Any | None

format: str | None

in_: _In

items: Items | Empty | None

maxItems: int | None

maxLength: int | None

maximum: int | None

minItems: int | None

minLength: int | None

minimum: int | None

multipleOf: int | None

name: str

pattern: str | None

required: bool | None

schema_: Schema | Reference | None

type: str | None

uniqueItems: bool | None

class aiopenapi3.v30.parameter.Parameter(**data)

allowEmptyValue: bool | None

allowReserved: bool | None

content: dict[str, 'MediaType']

deprecated: bool | None

description: str | None

example: Any | None

examples: dict[str, Union['Example', Reference]]

explode: bool | None

extensions: Any | None

in_: _In

name: str

required: bool | None

schema_: Schema | Reference | None

style: str | None

class aiopenapi3.v31.parameter.Parameter(**data)

allowEmptyValue: bool | None

allowReserved: bool | None

content: dict[str, 'MediaType'] | None

deprecated: bool | None

description: str | None

example: Any | None

examples: dict[str, Union['Example', Reference]]

explode: bool | None

extensions: Any | None

in_: _In

name: str

required: bool | None

schema_: Schema | None

style: str | None

Header

class aiopenapi3.v30.parameter.Header(**data)

allowEmptyValue: bool | None

allowReserved: bool | None

content: dict[str, 'MediaType']

deprecated: bool | None

description: str | None

example: Any | None

examples: dict[str, Union['Example', Reference]]

explode: bool | None

extensions: Any | None

required: bool | None

schema_: Schema | Reference | None

style: str | None

class aiopenapi3.v31.parameter.Header(**data)

allowEmptyValue: bool | None

allowReserved: bool | None

content: dict[str, 'MediaType'] | None

deprecated: bool | None

description: str | None

example: Any | None

examples: dict[str, Union['Example', Reference]]

explode: bool | None

extensions: Any | None

required: bool | None

schema_: Schema | None

style: str | None

Parameter Encoding

Each of those Parameters has a different encoding strategy for different argument types. e.g. encoding a List[str] as query parameter or object in a header. Additionally Swagger 2.0 has a different encoding strategy to OpenAPI 3.x.

class aiopenapi3.v20.parameter._ParameterCodec

_decode(value)

_encode(name, value)

class aiopenapi3.v30.parameter._ParameterCodec

_decode(value)

_encode(name, value)

Plugin Interface

Inheritance diagram of aiopenapi3.plugin.Init, aiopenapi3.plugin.Document, aiopenapi3.plugin.Message

Init Plugins

Init plugins are used during initialization, they can be used to modify PathItems or Schemas before generating the OperationIndex/Models.

Examples

class Init.Context(initialized=None, schemas=None, resolved=None, paths=None)

initialized: Optional[OpenAPI] = None: available in initialized()

paths: dict[str, PathItemBase] | None = None: available in paths()

schemas: dict[str, SchemaBase] | None = None: available in schemas()

class aiopenapi3.plugin.Init

initialized(ctx)

it is initialized

Return type:: Context

paths(ctx)

modify the paths/PathItems before initializing the Operations

Return type:: Context

schemas(ctx)

modify the Schema before creating Models

Return type:: Context

Document Plugins

Document plugins are used to mangle description documents.

Examples

class Document.Context(url, document)

document: dict[str, Any]: available in loaded() parsed()

url: URL: available in loaded() parsed()

class aiopenapi3.plugin.Document

loaded(text) -> parsed(dict)

loaded(ctx)

modify the text before parsing

Return type:: Context

parsed(ctx)

modify the parsed dict before …

Return type:: Context

Message Plugins

Message plugins are used to mangle message.

Examples

class Message.Context(request, operationId, marshalled=None, sending=None, received=None, headers=None, cookies=None, status_code=None, content_type=None, parsed=None, expected_type=None, unmarshalled=None)

content_type: str | None = None: available received()

expected_type: type | None = None: available parsed()

headers: Headers = None: available sending() received()

marshalled: dict[str, Any] | None = None: available marshalled()

operationId: str: available marshalled() sending() received() parsed() unmarshalled()

parsed: dict[str, Any] | None = None: available parsed()

received: bytes | None = None: available received()

sending: str | None = None: available sending()

status_code: str | None = None: available received()

unmarshalled: BaseModel | None = None: available unmarshalled()

class aiopenapi3.plugin.Message

sending: marshalled(dict)-> sending(str)

receiving: received -> parsed -> unmarshalled

marshalled(ctx)

modify the dict before sending

Return type:: Context

parsed(ctx)

modify the parsed dict structure

Return type:: Context

received(ctx)

modify the received text

Return type:: Context

sending(ctx)

modify the text before sending

Return type:: Context

unmarshalled(ctx)

modify the object

Return type:: Context

Loader

The loader is used to access description documents.

aiopenapi3.loader.Loader is the base class, providing flexibility to load description documents.

Inheritance diagram of aiopenapi3.loader.FileSystemLoader, aiopenapi3.loader.WebLoader, aiopenapi3.loader.ChainLoader, aiopenapi3.loader.RedirectLoader, aiopenapi3.loader.ChainLoader

loading operation

The order of operation for the loader is:

aiopenapi3.loader.Loader.get()
- aiopenapi3.loader.Loader.load()
  - aiopenapi3.loader.Loader.decode()
    - aiopenapi3.plugin.Document.loaded()
- aiopenapi3.loader.Loader.parse()
  - yaml?
    - yaml.load(yload)
  - json?
    - json.loads
- aiopenapi3.plugin.Document.parsed()

Loaders

class aiopenapi3.loader.Loader(yload=<class 'aiopenapi3.loader.YAML12Loader'>)

Loaders are used to ‘get’ description documents:

load

decode

parse

classmethod decode(data, codec)

decode bytes to ascii or utf-8

Parameters:

data (bytes)
codec (str | None)

Return type:

str

Returns:

get(plugins, url): load & parse the description document :type plugins: Plugins :param plugins: collection of aiopenapi3.plugin.Document plugins :type url: URL :param url: location of the description document :return:

abstract load(plugins, url, codec=None)

load and decode description document

Parameters:

plugins (Plugins) – collection of aiopenapi3.plugin.Document plugins
url (URL) – location of the description document
codec (Optional[str])

Returns:

decoded data

parse(plugins, url, data)

parse the downloaded document as json or yaml

Parameters:

plugins (Plugins) – collection of aiopenapi3.plugin.Document plugins
url (URL) – location of the description document
data (str) – decoded data of the description document

Returns:

class aiopenapi3.loader.FileSystemLoader(base, yload=<class 'aiopenapi3.loader.YAML12Loader'>): Loader to use the local filesystem

class aiopenapi3.loader.WebLoader(baseurl, session_factory=<class 'httpx.Client'>, yload=<class 'aiopenapi3.loader.YAML12Loader'>): Loader downloads data via http/s using the supplied session_factory

class aiopenapi3.loader.ChainLoader(*loaders, yload=<class 'aiopenapi3.loader.YAML12Loader'>): Loader to chain different Loaders: succeed or raise trying

The ChainLoader is useful when using multiple locations with description documents. As an example, try to lookup the referenced description documents locally or from the web.

description_documents = Path("/data/description_documents")

loader = ChainLoader(
            RedirectLoader(description_documents / "dell"),
            WebLoader(yarl.URL("https://redfish.dmtf.org/schemas/v1/")),
            WebLoader(yarl.URL("http://redfish.dmtf.org/schemas/swordfish/v1/")),
)

api = OpenAPI.load_file(
    target, yarl.URL("openapi.yaml"), loader=loader
)

class aiopenapi3.loader.RedirectLoader(base, yload=<class 'aiopenapi3.loader.YAML12Loader'>): Loader to redirect web-requests to a local directory everything but the “name” is stripped of the url

The RedirectLoader allows redirecting to local resources. A description documents URI is stripped to the file name of the document, and loaded relative to the basedir of the RedirectLoader.

RedirectLoader("description_documents/dell")

Exceptions

There is different types of Exceptions used depending on the subsystem/failure.

Inheritance diagram of aiopenapi3.errors.SpecError, aiopenapi3.errors.ReferenceResolutionError, aiopenapi3.errors.OperationParameterValidationError, aiopenapi3.errors.ParameterFormatError, aiopenapi3.errors.HTTPError, aiopenapi3.errors.RequestError, aiopenapi3.errors.ResponseError, aiopenapi3.errors.ContentTypeError, aiopenapi3.errors.HTTPStatusError, aiopenapi3.errors.ResponseDecodingError, aiopenapi3.errors.ResponseSchemaError, aiopenapi3.errors.ContentLengthExceededError, aiopenapi3.errors.HeadersMissingError, aiopenapi3.errors.HTTPStatusIndicatedError, aiopenapi3.errors.HTTPClientError, aiopenapi3.errors.HTTPServerError

Description Document Validation

class aiopenapi3.errors.SpecError(message, element=None): This error class is used when an invalid format is found while parsing an object in the spec.

class aiopenapi3.errors.ReferenceResolutionError(message, element=None): This error class is used when resolving a reference fails, usually because of a malformed path in the reference.

class aiopenapi3.errors.ParameterFormatError(message, element=None): The specified parameter encoding is invalid for the parameter family

class aiopenapi3.errors.OperationParameterValidationError(path, method, operationid, message)

The operations parameters do not match the path parameters

message: str

method: str

operationid: str

path: str

Message

exception aiopenapi3.errors.HTTPError

HTTPError is the base class for all request/response related errors.

Inheritance diagram of aiopenapi3.errors.RequestError, aiopenapi3.errors.ResponseError

exception aiopenapi3.errors.RequestError(operation, request, data, parameters)

data: Union[dict[str, JSON], list[JSON], str, int, float, bool, None, BaseModel, Sequence[tuple[str, Union[IO[bytes], bytes, str, Tuple[Optional[str], Union[IO[bytes], bytes, str]], Tuple[Optional[str], Union[IO[bytes], bytes, str], Optional[str]], Tuple[Optional[str], Union[IO[bytes], bytes, str], Optional[str], Mapping[str, str]]]]]]

operation: Union[Operation, Operation, Operation, None]

parameters: Optional[dict[str, Union[str, BaseModel]]]

request: Union[Request, Request, None]

A RequestError typically wraps an error of the underlying httpx library.

exception aiopenapi3.errors.ResponseError: the response can not be processed accordingly

ResponseErrors indicate the response does not match the expectation/definition in the description document. Most ResponseErrors can be mitigated around using plugins to match the protocol to the description document.

Inheritance diagram of aiopenapi3.errors.ContentTypeError, aiopenapi3.errors.ContentLengthExceededError, aiopenapi3.errors.HTTPStatusError, aiopenapi3.errors.ResponseDecodingError, aiopenapi3.errors.ResponseSchemaError, aiopenapi3.errors.HeadersMissingError

exception aiopenapi3.errors.ContentTypeError(operation, content_type, message, response)

The content-type is unexpected

content_type: str | None

message: str

operation: Union[Operation, Operation, Operation]

response: Response

exception aiopenapi3.errors.ContentLengthExceededError(operation, content_length, message, response)

The Content-Length exceeds our Limits

content_length: int

message: str

operation: Union[Operation, Operation, Operation]

response: Response

exception aiopenapi3.errors.HTTPStatusError(operation, http_status, message, response)

The HTTP Status is unexpected

http_status: int

message: str

operation: Union[Operation, Operation, Operation]

response: Response

exception aiopenapi3.errors.ResponseDecodingError(operation, data, response)

the json decoder failed

data: str

operation: Union[Operation, Operation, Operation]

response: Response

exception aiopenapi3.errors.ResponseSchemaError(operation, expectation, schema, response, exception)

the response data does not match the schema

exception: Exception | None

expectation: Union[Response, MediaType, MediaType]

operation: Union[Operation, Operation, Operation]

response: Response

schema: Union[Schema, Schema, Schema, None]

exception aiopenapi3.errors.HeadersMissingError(operation, missing, response)

the response is missing required header/s

missing: dict[str, HeaderType]

operation: Union[Operation, Operation, Operation]

response: Response

HTTP Status

exception aiopenapi3.errors.HTTPStatusIndicatedError(status_code, headers, data)

The HTTP Status is 4xx or 5xx

data: BaseModel

headers: dict[str, str]

status_code: int

exception aiopenapi3.errors.HTTPClientError(status_code, headers, data): response code 4xx

exception aiopenapi3.errors.HTTPServerError(status_code, headers, data): response code 5xx

Extra

Cull & Reduce

Reduce & Cull are Plugins limiting the models built to the minimum required to match the requirements of the supplied Operations Code below will eleminate all schemas not required to serve the operations identified by the pattern/string match and http methods associated.

class aiopenapi3.extra.Reduce(*operations)

__init__(*operations)

Parameters:: operations (list[Union[tuple[PathMatchType, list[HTTPMethodMatchType] | None], Pattern, str]]) – the operation selector

class aiopenapi3.extra.Cull(*operations): The Cull plugin removes everything not required as early as possible

api = OpenAPI.load_sync(
    "http://127.0.0.1/api.yaml",
    plugins=[
        Cull(
            "getPetById",
            re.compile(r".*Pet.*"),
            ("/logout", ["get"]),
            (re.compile(r"^/user.*"), ["post"]),
        )
    ],
)

Inheritance diagram of aiopenapi3.plugin.Init, aiopenapi3.plugin.Document, aiopenapi3.extra.Cull, aiopenapi3.extra.Reduce

Cookies

This plugin deals with cookies from responses and adds in requests based on a policy.

class aiopenapi3.extra.Cookies(cookiejar=None, policy='jar')

__init__(cookiejar=None, policy='jar')

Parameters:

cookiejar (Optional[CookieJar]) – a cookiejar.CookieJar instance
policy (Literal['jar', 'securitySchemes']) –
There are two policies:
- securitySchemes - only cookies whose name is in the securitySchemes are used. To match authentication requirements credentials are set via OpenAPI.authenticate(name=value)
- jar - all cookies are used, basically like a browser would do. Cookies not mentioned in securitySchemes are set besides OpenAPI.authenticate() to allow using them without adjusting the description document.

api = OpenAPI.load_sync(
    "http://127.0.0.1/api.yaml",
    plugins=[
        Cookies(policy="jar")
    ],
)

Inheritance diagram of aiopenapi3.plugin.Plugin, aiopenapi3.plugin.Init, aiopenapi3.plugin.Message, aiopenapi3.extra.cookies.Cookies