Plugins
To assist dealing with differences in the description document and the protocol, aiopenapi3 provides a capable plugin interface which allows mangling the description document and the messages sent/received to match.
Init
Init
plugins are run at certain stages during the initialization of the OpenAPI object.
Available callbacks:
Examples
initialized
The following examples modifies specific pydantic models to allow unknown properties.
class DellInit(aiopenapi3.plugin.Init):
def initialized(self, ctx):
"""
Resource_Oem & Attributes are objects with additionalProperties
the default will ignore unknown properties
"""
def schemas(name, fn):
for doc in self.api._documents.values():
schema = doc.components.schemas.get(name, None)
if not schema:
continue
fn(doc, schema)
schemas("Resource_Oem", lambda doc, schema:
setattr(schema.get_type().Config, "extra", pydantic.Extra.allow))
schemas("DellAttributes_v1_0_0_DellAttributes",
lambda doc, schema: setattr(
doc.components.schemas["DellAttributes_v1_0_0_Attributes"].get_type().Config,
"extra",
pydantic.Extra.allow)
)
return ctx
reducing initialization processing time
Large -multi megabyte- APIs have a significant processing time.
Reducing this processing time during development is possible by limiting the number of objects initialized.
The schemas callback limits initialization to the schemas returned and their dependencies. The paths callback removes all PathItems.
"""
to speed up things, we use some aiopenapi3 plugins to limit the loading process to the schemas required
removing all paths
"""
from aiopenapi3.plugin import Init, Document
class SchemaSelector(Init):
"""
remove the schemas we do not need models for
"""
def __init__(self, *schemas):
super().__init__()
self._schemas = schemas
def schemas(self, ctx: "Init.Context") -> "Init.Context":
ctx.schemas = {k: ctx.schemas[k] for k in (set(self._schemas) & set(ctx.schemas.keys()))}
return ctx
class RemovePaths(Document):
def parsed(self, ctx: "Document.Context") -> "Document.Context":
"""
emtpy the paths - not needed
"""
ctx.document["paths"] = {}
return ctx
selector = SchemaSelector(*(list(names) + [f"{name}Request" for name in names]))
api = OpenAPI.load_file(..., plugins=[selector, RemovePaths()])
Document
Document
plugins allow modification of the description document.
Available callbacks:
Examples
As an example, due to a bug #21997 the response repoGetArchive operation of gitea does not match the content type of the description document:
Using a Document plugin to modify the parsed description document to state the content type “application/octet-stream” for the repoGetArchive operation.
import tarfile
import io
class ContentType(aiopenapi3.plugin.Document):
def parsed(self, ctx):
try:
ctx.document["paths"]["/repos/{owner}/{repo}/archive/{archive}"]["get"]["produces"] = ["application/octet-stream"]
except Exception as e:
print(e)
return ctx
api = OpenAPI.load_sync("https://try.gitea.io/swagger.v1.json", plugins=[ContentType()])
Message
Message
plugins allow modification of the messages sent/received.
For messages sent, the available callbacks are:
Avaiable callbacks for messages received:
Examples
Signing the Body
This example signs a message body by providing a HMAC512 signature in the http headers:
class XHookSignature(aiopenapi3.plugin.Message):
def sending(self, ctx: "Message.Context") -> "Message.Context":
ctx.headers["X-Hook-Signature"] = sign(ctx.sending)
return ctx
api = await aiopenapi3.OpenAPI.load_async(url, plugins=[XHookSignature()])
Correct an invalid Responses
Aliasing a property
This treatment is about a bug #22048 in the
repoGetPullRequestCommits operation.
The returned parameter X-Total is not set, X-Total-Count is set instead. To mitigate we provide a message plugin
which copies the value to X-Total in the aiopenapi3.plugin.Message.received()
callback.
class repoGetPullRequestCommitsMessage(aiopenapi3.plugin.Message):
def received(self, ctx: "Message.Context") -> "Message.Context":
if ctx.operationId != "repoGetPullRequestCommits":
return ctx
try:
if ctx.headers.get("X-Total", None) is None:
ctx.headers["X-Total"] = ctx.headers.get("X-Total-Count", 0)
except Exception as e:
print(e)
return ctx
api = OpenAPI.load_sync("https://try.gitea.io/swagger.v1.json", plugins=[repoGetPullRequestCommitsMessage()])
Correcting Spelling for an enum
… the ConnectorType is an enum value the services does not honor:
class SerialInterface(Message):
def parsed(self, ctx):
if "ConnectorType" in ctx.expected_type.get_type().__fields__ and ctx.parsed.get("ConnectorType", None) == 'DB9 Male.':
ctx.parsed["ConnectorType"] = "DB9 Male"
return ctx
Adding missing properties
… the service is missing required Fields:
class IdMissingMessage(Message):
def parsed(self, ctx):
rq = set(map(lambda x: x.alias, filter(lambda x: x.required == True, ctx.expected_type.get_type().__fields__.values())))
av = set(ctx.parsed.keys())
m = rq - av
if m:
print(f"missing {m} at {ctx.parsed}")
for k in m:
ctx.parsed[k] = ""
Correcting an invalid datetime value
… the service uses invalid datetime values (month & day == 0):
class DateError(Message):
def __init__(self, key):
super().__init__()
self.key = key
def parsed(self, ctx):
if self.key in ctx.expected_type.get_type().__fields__:
v = ctx.parsed.get(self.key, None)
if v in ['0000-00-00T00:00:00+00:00',"00:00:00Z"]:
# '0000-00-00T00:00:00+00:00'
del ctx.parsed[self.key]
return ctx