from typing import MutableMapping, Optional, Type
class _BaseMetaClass(type):
subclasses: MutableMapping[str, Type["_ChildContext"]] = {}
def __new__(cls, clsname, bases, attrs):
# produce the method that creates child contexts
def mk_ch_context_meth(child_context, field):
def fn(self, **kwargs):
ch_context = cls.subclasses[child_context](self)
for at, v in kwargs.items():
if not at.startswith("_") and at in dir(ch_context):
meth = getattr(ch_context, at)
if callable(meth):
meth(v)
continue
raise AttributeError(f"Child does not have {at}")
self.document[field] = ch_context.document
return ch_context
return fn
# produce the method that updates attributes
def mk_attr_meth(field):
def fn(self, v):
self.document[field] = v
return self
return fn
if "CHILD_CONTEXTS" in attrs:
for name, (ch_context, field) in attrs["CHILD_CONTEXTS"].items():
# FIXME: Hopefully we want the docstrings for params here
# Which is kind of a problem because the "more root" types
# are defined earlier, and to obtain the possible params
# we need to look at the "less root" class which is yet to be
# defined.
docstring = f"""Set {field}
Create {ch_context} and set {field}
"""
meth = mk_ch_context_meth(ch_context, field)
meth.__doc__ = docstring
meth.__name__ = name
attrs[name] = meth
if "ATTRIBUTE_FIELDS" in attrs:
for name, field in attrs["ATTRIBUTE_FIELDS"].items():
docstring = f"""Set {field} field"""
meth = mk_attr_meth(field)
meth.__doc__ = docstring
meth.__name__ = name
attrs[name] = meth
new_cls = type.__new__(cls, clsname, bases, attrs)
cls.subclasses[clsname] = new_cls
return new_cls
class _BaseContext(metaclass=_BaseMetaClass):
# {'method_name': ('ContextClass', 'fieldName')}
CHILD_CONTEXTS = {}
# {'method_name: 'fieldName'}
ATTRIBUTE_FIELDS = {}
EXTENSION = False
def __init__(self):
self.document = {}
def __getattr__(self, item: str):
if self.EXTENSION and item.startswith("x_") and len(item) > 2:
default_field_name = item.replace("_", "-")
def set_extension_value(__value, field: str = default_field_name):
if not field.startswith("x-") or len(field) < 3:
raise ValueError(f"Invalid extension field '{field}'")
self.document[field] = __value
return self
set_extension_value.__name__ = f"{item}"
set_extension_value.__doc__ = "Set extension"
return set_extension_value
raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{item}'")
class _ChildContext(_BaseContext):
def __init__(self, parent):
super().__init__()
self.parent = parent
def end(self):
"""Explicitly return to the parent context
Returns:
Parent context
"""
return self.parent
def __getattr__(self, attr_name: str):
try:
return super(_ChildContext, self).__getattr__(attr_name)
except AttributeError:
pass # would have returned if it's an extension
multilevel_allow = ["path", "get", "put", "post", "delete", "options", "head", "patch", "trace"]
if not attr_name.startswith("_"):
if attr_name in dir(self.parent) or (attr_name in multilevel_allow and hasattr(self.parent, attr_name)):
return getattr(self.parent, attr_name)
raise AttributeError()
class _HasParameters(_BaseContext):
def parameter(self, name, in_, required):
parameters = self.document.setdefault("parameters", [])
param_context = OpenAPIParameterContext(self, name, in_, required)
parameters.append(param_context.document)
return param_context
class _HasSummary(_BaseContext):
ATTRIBUTE_FIELDS = {"summary": "summary"}
class _HasDescription(_BaseContext):
ATTRIBUTE_FIELDS = {"description": "description"}
class _HasExternalDocs(_BaseContext):
CHILD_CONTEXTS = {
"external_docs": ("OpenAPIExternalDocsContext", "externalDocs"),
}
class _HasTags(_BaseContext):
def tag(self, t):
tags = self.document.setdefault("tags", [])
tags.append(t)
return self
[docs]
class OpenAPIContext(_HasExternalDocs):
CHILD_CONTEXTS = {
"info": ("OpenAPIInfoContext", "info"),
}
EXTENSION = True
def __init__(self):
super(OpenAPIContext, self).__init__()
self.document = {
"openapi": "3.0.3",
"paths": {},
}
[docs]
def server(self, url: str, description: Optional[str] = None):
servers = self.document.setdefault("servers", [])
server = {"url": url}
if description:
server["description"] = description
servers.append(server)
return self
[docs]
def path(self, path: str, summary: Optional[str] = None, description: Optional[str] = None):
path_item = OpenAPIPathItemContext(self)
if summary:
path_item.summary(summary)
if description:
path_item.description(description)
self.document["paths"][path] = path_item.document
return path_item
[docs]
class OpenAPIInfoContext(_ChildContext, _HasDescription):
ATTRIBUTE_FIELDS = {
"title": "title",
"terms_of_service": "termsOfService",
"version": "version",
}
CHILD_CONTEXTS = {
"contact": ("OpenAPIContactContext", "contact"),
"license": ("OpenAPILicenseContext", "license"),
}
EXTENSION = True
[docs]
class OpenAPIContactContext(_ChildContext):
ATTRIBUTE_FIELDS = {
"name": "name",
"url": "url",
"email": "email",
}
EXTENSION = True
[docs]
class OpenAPILicenseContext(_ChildContext):
ATTRIBUTE_FIELDS = {
"name": "name",
"url": "url",
}
EXTENSION = True
[docs]
class OpenAPIPathItemContext(_ChildContext, _HasSummary, _HasDescription, _HasParameters):
CHILD_CONTEXTS = {}
for http_method in ["get", "put", "post", "delete", "options", "head", "patch", "trace"]:
CHILD_CONTEXTS[http_method] = ("OpenAPIOperation", http_method)
EXTENSION = True
[docs]
class OpenAPIOperation(_ChildContext, _HasSummary, _HasExternalDocs, _HasTags, _HasDescription, _HasParameters):
ATTRIBUTE_FIELDS = {"operation_id": "operationId"}
EXTENSION = True
def __init__(self, parent):
super(OpenAPIOperation, self).__init__(parent)
self.document = {
"responses": {
"200": {
"description": "Success",
}
}
}
[docs]
class OpenAPIParameterContext(_ChildContext, _HasDescription):
ATTRIBUTE_FIELDS = {
"deprecated": "deprecated",
"allow_empty": "allowEmptyValue",
"style": "style",
"explode": "explode",
"allow_reserved": "allowReserved",
"schema": "schema",
}
EXTENSION = True
def __init__(self, parent, name: str, in_: str, required: bool):
super(OpenAPIParameterContext, self).__init__(parent)
self.document = {
"name": name,
"in": in_,
"required": required,
}
[docs]
def type(self, typ: str, **kwargs):
schema = {
"type": typ,
}
for k in ["minimum", "maximum", "default"]:
v = kwargs.get(k)
if v is not None:
schema[k] = v
self.schema(schema)
return self
[docs]
class OpenAPIExternalDocsContext(_ChildContext, _HasDescription):
ATTRIBUTE_FIELDS = {
"url": "url",
}
EXTENSION = True
OpenAPIDocumentBuilder = OpenAPIContext