nikto_b/turbosloth
1
1
Fork 0
Code Issues 7 Pull Requests Actions Packages Releases Wiki Activity

Add openapi request basic schema generation, fix SerializedResponse headers, HTTP codes

Browse Source
This commit is contained in:
Артур Шевченко
2025-08-26 17:33:23 +03:00
parent c40bdca9e4
commit f823d2df5a
10 changed files with 887 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
src/turbosloth/__main__.py
View File

@@ -1,19 +1,21 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Optional from typing import Optional
import uvicorn import uvicorn
from turbosloth import SlothApp from turbosloth import SlothApp
from turbosloth.doc.openapi_app import OpenAPIApp
from turbosloth.doc.openapi_models import Info
from turbosloth.interfaces.base import BasicResponse
from turbosloth.interfaces.serialize_selector import SerializeSelector from turbosloth.interfaces.serialize_selector import SerializeSelector
from turbosloth.interfaces.serialized import SerializedResponse, SerializedRequest from turbosloth.interfaces.serialized import SerializedResponse
from turbosloth.internal_types import QTYPE, BTYPE, PTYPE, HTYPE from turbosloth.schema import RequestBody, HeaderParam, QueryParam
from turbosloth.req_schema import UnwrappedRequest
from turbosloth.schema import RequestBody, HeaderParam, PathParam
app = SlothApp(di_autodoc_prefix='/didoc', app = SlothApp(di_autodoc_prefix='/didoc',
serialize_selector=SerializeSelector(default_content_type='application/json')) serialize_selector=SerializeSelector(default_content_type='application/json'),
openapi_app=OpenAPIApp(Info('asdf', '1.0.0')))
# @app.get("/") # @app.get("/")
@@ -117,7 +119,41 @@ async def test_body(r: RequestBody(UserPostSchema),
'user': user.__dict__, 'user': user.__dict__,
'q2': q2 'q2': q2
} }
return SerializedResponse(200, {}, resp) return SerializedResponse(resp)
@app.get('/openapi.json')
async def openapi_schema(a: SlothApp, version: QueryParam(Optional[str], 'v') = None) -> SerializedResponse:
dat = a.openapi_app.export_as_dict()
if version is not None:
dat['openapi'] = version
return SerializedResponse(dat)
@app.get('/stoplight')
async def stoplight() -> BasicResponse:
ret = '''<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no">
<title>Elements in HTML</title>
<!-- Embed elements Elements via Web Component -->
<script src="https://unpkg.com/@stoplight/elements/web-components.min.js"></script>
<link rel="stylesheet" href="https://unpkg.com/@stoplight/elements/styles.min.css">
<style>.sl-elements {height: 100vh}</style>
</head>
<body>
<elements-api
apiDescriptionUrl="/openapi.json"
router="hash"
layout="sidebar"
/>
</body>
</html>'''
return BasicResponse(200, {}, ret.encode())
if __name__ == '__main__': if __name__ == '__main__':

197
src/turbosloth/app.py
View File

@@ -2,8 +2,9 @@ from __future__ import annotations
import html import html
import typing import typing
from dataclasses import dataclass from dataclasses import dataclass
from types import NoneType
from typing import Optional, Callable, Awaitable, Protocol, get_type_hints, get_origin, get_args, Any, Annotated, Tuple, \ from typing import Optional, Callable, Awaitable, Protocol, get_type_hints, get_origin, get_args, Any, Annotated, Tuple, \
Iterable Iterable, TypeAliasType, Union
import breakshaft.util_mermaid import breakshaft.util_mermaid
import megasniff.exceptions import megasniff.exceptions
@@ -11,6 +12,8 @@ from breakshaft.models import ConversionPoint, Callgraph
from megasniff import SchemaInflatorGenerator from megasniff import SchemaInflatorGenerator
from .didoc import create_di_autodoc_handler from .didoc import create_di_autodoc_handler
from .doc.openapi_app import OpenAPIApp
from .doc.openapi_models import RequestBody as OpenApiRequestBody, MediaType, Schema, Reference, Parameter, Response
from .exceptions import HTTPException from .exceptions import HTTPException
from .interfaces.base import BasicRequest, BasicResponse from .interfaces.base import BasicRequest, BasicResponse
from .interfaces.serialize_selector import SerializeSelector from .interfaces.serialize_selector import SerializeSelector
@@ -125,9 +128,9 @@ class HTTPApp(ASGIApp):
return return
except (megasniff.exceptions.FieldValidationException, megasniff.exceptions.MissingFieldException) as e: except (megasniff.exceptions.FieldValidationException, megasniff.exceptions.MissingFieldException) as e:
print(e) print(e)
sresp = SerializedResponse(400, {}, 'Schema error') sresp = SerializedResponse('Schema error', code=400)
except HTTPException as e: except HTTPException as e:
sresp = SerializedResponse(e.code, {}, str(e)) sresp = SerializedResponse(str(e), code=e.code)
try: try:
ct = self.extract_content_type(req) ct = self.extract_content_type(req)
@@ -207,13 +210,15 @@ class MethodRoutersApp(ASGIApp):
class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp): class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
di_autodoc_prefix: Optional[str] = None di_autodoc_prefix: Optional[str]
openapi_app: Optional[OpenAPIApp]
def __init__(self, def __init__(self,
on_startup: Optional[Callable[[], Awaitable[None]]] = None, on_startup: Optional[Callable[[], Awaitable[None]]] = None,
on_shutdown: Optional[Callable[[], Awaitable[None]]] = None, on_shutdown: Optional[Callable[[], Awaitable[None]]] = None,
di_autodoc_prefix: Optional[str] = None, di_autodoc_prefix: Optional[str] = None,
serialize_selector: Optional[SerializeSelector] = None, serialize_selector: Optional[SerializeSelector] = None,
openapi_app: Optional[OpenAPIApp] = None
): ):
self.router = Router() self.router = Router()
self._on_startup = on_startup self._on_startup = on_startup
@@ -223,11 +228,12 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
self.serialize_selector = serialize_selector self.serialize_selector = serialize_selector
self.infl_generator = SchemaInflatorGenerator(strict_mode=True, store_sources=True) self.infl_generator = SchemaInflatorGenerator(strict_mode=True, store_sources=True)
self.di_autodoc_prefix = di_autodoc_prefix self.di_autodoc_prefix = di_autodoc_prefix
self.openapi_app = openapi_app
if di_autodoc_prefix is not None: if di_autodoc_prefix is not None:
self.inj_repo = ConvRepo(store_sources=True, store_callseq=True) self.inj_repo = ConvRepo(store_sources=True, store_callseq=True)
else: else:
self.inj_repo = ConvRepo() self.inj_repo = ConvRepo(store_callseq=True)
@self.inj_repo.mark_injector() @self.inj_repo.mark_injector()
def extract_query(req: BasicRequest | SerializedRequest) -> QTYPE: def extract_query(req: BasicRequest | SerializedRequest) -> QTYPE:
@@ -245,6 +251,10 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
def extract_body(req: SerializedRequest) -> BTYPE: def extract_body(req: SerializedRequest) -> BTYPE:
return req.body return req.body
@self.inj_repo.mark_injector()
def provide_sloth_app() -> SlothApp:
return self
self.inj_repo.add_injector(self.extract_content_type) self.inj_repo.add_injector(self.extract_content_type)
self.inj_repo.add_injector(self.extract_accept_type) self.inj_repo.add_injector(self.extract_accept_type)
self.inj_repo.add_injector(self.serialize_request) self.inj_repo.add_injector(self.serialize_request)
@@ -274,6 +284,164 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
return infl, set(ConversionPoint.from_fn(infl, type_remap=type_remap)) return infl, set(ConversionPoint.from_fn(infl, type_remap=type_remap))
return None, None return None, None
@staticmethod
def schema_from_type(tp) -> Schema | Reference:
if isinstance(tp, TypeAliasType):
return SlothApp.schema_from_type(tp.__value__)
origin = get_origin(tp)
args = get_args(tp)
if origin is Annotated:
return SlothApp.schema_from_type(args[0])
if origin is None:
# базовый тип
if tp == str:
return Schema(type='string')
elif tp == int:
return Schema(type='integer')
elif tp == float:
return Schema(type='number', format='float')
elif tp == bool:
return Schema(type='boolean')
elif tp == type(None):
return Schema(type='null')
else:
# кастомный класс → $ref
return Reference(ref=f'#/components/schemas/{tp.__name__}')
elif origin is list:
item_type = args[0] if args else Any
return Schema(type='array', items=SlothApp.schema_from_type(item_type))
elif origin is dict:
key_type, value_type = args if args else (str, Any)
if key_type != str:
raise ValueError('OpenAPI dict keys must be strings')
return Schema(
type="object",
additionalProperties=SlothApp.schema_from_type(value_type)
)
elif origin is typing.Union:
schemas = [SlothApp.schema_from_type(a) for a in args]
return Schema(oneOf=schemas)
else:
raise NotImplementedError(f"Тип {tp} не поддержан")
def register_endpoint_components(self, handler: Callable):
callseq = getattr(handler, '__breakshaft_callseq__', [])
if len(callseq) == 0:
return []
ret = []
for call in callseq:
call: ConversionPoint
config: EndpointConfig | None = getattr(call.fn, '__turbosloth_config__', None)
if config is None:
continue
for k, v in config.header_schemas.items():
t = v.replacement_type
s = self.schema_from_type(t)
if isinstance(s, Reference):
self.openapi_app.register_component(t)
for k, v in config.query_schemas.items():
t = v.replacement_type
s = self.schema_from_type(t)
if isinstance(s, Reference):
self.openapi_app.register_component(t)
for k, v in config.path_schemas.items():
t = v.replacement_type
s = self.schema_from_type(t)
if isinstance(s, Reference):
self.openapi_app.register_component(t)
if config.body_schema is not None:
t = config.body_schema.replacement_type
s = self.schema_from_type(t)
if isinstance(s, Reference):
self.openapi_app.register_component(t)
@staticmethod
def construct_req_body(handler: Callable) -> Schema | Reference | None:
callseq = getattr(handler, '__breakshaft_callseq__', [])
if len(callseq) == 0:
return None
ret = {}
for call in callseq:
call: ConversionPoint
if BTYPE in call.requires:
return SlothApp.schema_from_type(call.injects)
return None
@staticmethod
def construct_req_params(handler: Callable) -> list[Parameter]:
callseq = getattr(handler, '__breakshaft_callseq__', [])
if len(callseq) == 0:
return []
ret = []
for call in callseq:
call: ConversionPoint
config: EndpointConfig | None = getattr(call.fn, '__turbosloth_config__', None)
if config is None:
continue
for k, v in config.query_schemas.items():
v: ParamSchema
allow_empty = v.has_default
t = v.schema
if isinstance(t, TypeAliasType):
t = t.__value__
origin = get_origin(t)
if origin is Union:
args = get_args(t)
if None in args or NoneType in args:
allow_empty = True
ret.append(Parameter(
name=v.key_name,
in_='query',
required=not v.has_default,
allowEmptyValue=allow_empty,
schema=SlothApp.schema_from_type(v.schema)
))
for k, v in config.path_schemas.items():
v: ParamSchema
ret.append(Parameter(
name=v.key_name,
in_='path',
required=not v.has_default,
schema=SlothApp.schema_from_type(v.schema)
))
for k, v in config.header_schemas.items():
v: ParamSchema
ret.append(Parameter(
name=v.key_name,
in_='header',
required=not v.has_default,
schema=SlothApp.schema_from_type(v.schema)
))
return ret
def route(self, method: MethodType, path_pattern: str): def route(self, method: MethodType, path_pattern: str):
def decorator(fn: HandlerType): def decorator(fn: HandlerType):
path_substs = self.router.find_pattern_substs(path_pattern) path_substs = self.router.find_pattern_substs(path_pattern)
@@ -306,6 +474,22 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
self.route('GET', self.di_autodoc_prefix + '/' + method + path_pattern)( self.route('GET', self.di_autodoc_prefix + '/' + method + path_pattern)(
create_di_autodoc_handler(method, path_pattern, p, depgraph)) create_di_autodoc_handler(method, path_pattern, p, depgraph))
if self.openapi_app is not None:
self.openapi_app.register_endpoint(
method,
path_pattern,
self.construct_req_params(p),
{
'200': Response('desc', content={'application/json': MediaType(schema=Schema(type='boolean'))})
},
request_body=OpenApiRequestBody(
{
'application/json': MediaType(self.construct_req_body(p))
}
)
)
self.register_endpoint_components(p)
return fn return fn
return decorator return decorator
@@ -379,6 +563,9 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp):
setattr(func, '__turbosloth_DI_name__', breakshaft.util.universal_qualname(main_injects)) setattr(func, '__turbosloth_DI_name__', breakshaft.util.universal_qualname(main_injects))
fork_with |= set(ConversionPoint.from_fn(func, type_remap=fn_type_hints)) fork_with |= set(ConversionPoint.from_fn(func, type_remap=fn_type_hints))
setattr(func, '__turbosloth_config__', config)
return fork_with, fn_type_hints return fork_with, fn_type_hints
def add_injector(self, func: Callable): def add_injector(self, func: Callable):

276
src/turbosloth/doc/openapi_app.py Normal file
View File

@@ -0,0 +1,276 @@
import re
from dataclasses import is_dataclass, asdict, fields, MISSING
from typing import Sequence, Optional, Any, get_type_hints, get_origin, Annotated, get_args, Union, TypeAliasType
from mypy.checkpattern import defaultdict
from turbosloth.doc.openapi_models import Info, Server, PathItem, Components, SecurityRequirement, Tag, \
ExternalDocumentation, Parameter, RequestBody, Operation, Response, OpenAPI, Schema, Reference, SchemaType
from turbosloth.internal_types import MethodType
def resolve_type(tp):
if isinstance(tp, TypeAliasType):
return resolve_type(tp.__value__)
origin = get_origin(tp)
if origin is Annotated:
base_type, *metadata = get_args(tp)
return resolve_type(base_type)
return tp
def type_to_schema(tp, components: Components, *,
_visited_component_names: Optional[list[str]] = None) -> Schema | Reference:
if _visited_component_names is None:
_visited_component_names = []
tp = resolve_type(tp)
origin = get_origin(tp)
args = get_args(tp)
if hasattr(tp, '__name__'):
name = tp.__name__
else:
name = str(tp)
if origin is None:
# базовый тип
if tp == str:
return Schema(type='string')
elif tp == int:
return Schema(type='integer')
elif tp == float:
return Schema(type='number', format='float')
elif tp == bool:
return Schema(type='boolean')
elif tp is type(None):
return Schema(type='null')
# Optional / Union
if origin is Union:
schemas = []
nullable = False
for a in args:
if a is type(None):
nullable = True
else:
schemas.append(type_to_schema(a, components, _visited_component_names=_visited_component_names))
if len(schemas) == 1:
schema = schemas[0]
else:
schema = Schema(oneOf=schemas)
if nullable:
schema = Schema(oneOf=[schema, Schema(type='null')])
return schema
# Базовые типы
if tp in (str, int, float, bool):
mapping: dict[type, SchemaType] = {
str: 'string',
int: 'integer',
float: 'number',
bool: 'boolean',
None: 'null'
}
schema = Schema(type=mapping[tp])
if tp is float:
schema.format = 'float'
return schema
# List
if origin is list:
item_type = args[0] if args else Any
return Schema(type='array',
items=type_to_schema(item_type, components, _visited_component_names=_visited_component_names))
# Tuple
if origin is tuple:
if args and args[-1] is Ellipsis:
return Schema(type="array",
items=type_to_schema(args[0], components, _visited_component_names=_visited_component_names))
else:
return Schema(
type="array",
prefixItems=[type_to_schema(a, components, _visited_component_names=_visited_component_names) for a in
args],
minItems=len(args),
maxItems=len(args)
)
# Dict
if origin is dict:
key_type, value_type = args if args else (str, Any)
if key_type is not str:
raise ValueError("OpenAPI dict keys must be strings")
return Schema(
type="object",
additionalProperties=type_to_schema(value_type, components,
_visited_component_names=_visited_component_names)
)
if name in _visited_component_names:
return Reference(f'#/components/schemas/{name}')
_visited_component_names.append(name)
if hasattr(tp, '__init__'):
init_params = get_type_hints(tp.__init__)
props = {}
required = []
# Проходим по параметрам конструктора
for pname, ptype in init_params.items():
if pname == "return":
continue
schema = type_to_schema(ptype, components, _visited_component_names=_visited_component_names)
# Пытаемся получить alias через Annotated / metadata
field_alias = getattr(ptype, "__metadata__", None)
alias = pname
if field_alias:
for meta in field_alias:
if isinstance(meta, dict) and "alias" in meta:
alias = meta["alias"]
props[alias] = schema
# Поля без default — required
param_default = getattr(tp, pname, None)
if param_default is None:
required.append(alias)
comp_schema = Schema(type="object", properties=props)
if required:
comp_schema.required = required
components.schemas[name] = comp_schema
return comp_schema
raise NotImplementedError(f"Тип {tp} не поддержан")
class OpenAPIApp:
info: Info
servers: list[Server]
paths: dict[str, PathItem]
webhooks: dict[str, PathItem]
components: Components
security: list[SecurityRequirement]
tags: list[Tag]
externalDocs: Optional[ExternalDocumentation]
def __init__(self, info: Info,
servers: Optional[list[Server]] = None,
externalDocs: Optional[ExternalDocumentation] = None):
if servers is None:
servers = []
self.info = info
self.servers = servers
self.paths = defaultdict(lambda: PathItem())
self.webhooks = {}
self.components = Components()
self.security = []
self.tags = []
self.externalDocs = externalDocs
def register_endpoint(self,
method: MethodType,
path: str,
parameters: list[Parameter],
responses: dict[str, Response],
summary: Optional[str] = None,
request_body: Optional[RequestBody] = None,
desciption: Optional[str] = None, ):
path_params = []
substs = set(re.findall(r'\{(.*?)}', path))
for s in substs:
path_params.append(Parameter(
name=s,
in_='path',
schema=Schema(type='string'),
required=True
))
self.paths[path].parameters = path_params
op = Operation(
responses,
summary,
desciption,
parameters=parameters,
requestBody=request_body
)
match method:
case 'GET':
self.paths[path].get = op
case 'POST':
self.paths[path].post = op
case 'PUSH':
return
case 'PUT':
self.paths[path].put = op
case 'PATCH':
self.paths[path].patch = op
case 'DELETE':
self.paths[path].delete = op
case 'HEAD':
self.paths[path].head = op
case 'CONNECT':
return
case 'OPTIONS':
self.paths[path].options = op
case 'TRACE':
self.paths[path].trace = op
def register_component(self, tp: type):
tp = resolve_type(tp)
schema = type_to_schema(tp, self.components)
self.components.schemas[f'{tp.__name__}'] = schema
def as_openapi(self) -> OpenAPI:
return OpenAPI(
self.info,
None,
self.servers,
self.paths,
self.webhooks,
self.components,
self.security,
self.tags,
self.externalDocs
)
@staticmethod
def export_openapi(obj: Any):
if is_dataclass(obj):
result = {}
hints = get_type_hints(obj.__class__)
for f in fields(obj):
# Проверяем, есть ли alias в metadata
alias = f.metadata.get("alias") if "alias" in f.metadata else None
# Для Annotated
tp = hints.get(f.name)
if getattr(tp, "__metadata__", None):
for meta in tp.__metadata__:
if isinstance(meta, dict) and "alias" in meta:
alias = meta["alias"]
key = alias or f.name
value = getattr(obj, f.name)
if value is not None:
result[key] = OpenAPIApp.export_openapi(value)
return result
elif isinstance(obj, dict):
return {k: OpenAPIApp.export_openapi(v) for k, v in obj.items() if v is not None}
elif isinstance(obj, (list, tuple, set)):
return [OpenAPIApp.export_openapi(v) for v in obj if v is not None]
else:
return obj
def export_as_dict(self) -> dict:
obj = self.as_openapi()
return self.export_openapi(obj)

368
src/turbosloth/doc/openapi_models.py Normal file
View File

@@ -0,0 +1,368 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Mapping, MutableMapping, Optional, Sequence, Tuple, TypeAlias, Union, Literal
# --- alias metadata key for fields whose JSON key differs from Python attr name ---
ALIAS: Literal["alias"] = "alias"
# --- Generic helpers ---
JsonValue: TypeAlias = Union[None, bool, int, float, str, list["JsonValue"], dict[str, "JsonValue"]]
ReferenceOr: TypeAlias = Union["Reference", Any] # refined per usage below
# --- Spec: https://spec.openapis.org/oas/v3.1.0 ---
# Components -------------------------------------------------------------------
@dataclass
class Components:
schemas: dict[str, Union[Schema, Reference]] = field(default_factory=dict)
responses: dict[str, Union[Response, Reference]] = field(default_factory=dict)
parameters: dict[str, Union[Parameter, Reference]] = field(default_factory=dict)
examples: dict[str, Union[Example, Reference]] = field(default_factory=dict)
requestBodies: dict[str, Union[RequestBody, Reference]] = field(default_factory=dict)
headers: dict[str, Union[Header, Reference]] = field(default_factory=dict)
securitySchemes: dict[str, Union[SecurityScheme, Reference]] = field(default_factory=dict)
links: dict[str, Union[Link, Reference]] = field(default_factory=dict)
callbacks: dict[str, Union[Callback, Reference]] = field(default_factory=dict)
pathItems: dict[str, Union[PathItem, Reference]] = field(default_factory=dict) # OAS 3.1
# Root -------------------------------------------------------------------------
@dataclass
class OpenAPI:
info: Info
jsonSchemaDialect: Optional[str] = None
servers: list[Server] = field(default_factory=list)
paths: dict[str, PathItem] = field(default_factory=list)
webhooks: dict[str, PathItem] = field(default_factory=list)
components: Components = field(default_factory=Components)
security: list[SecurityRequirement] = field(default_factory=list)
tags: list[Tag] = field(default_factory=list)
externalDocs: Optional[ExternalDocumentation] = None
openapi: str = "3.1.0"
# Info -------------------------------------------------------------------------
@dataclass
class Contact:
name: Optional[str] = None
url: Optional[str] = None
email: Optional[str] = None
@dataclass
class License:
name: str = ""
identifier: Optional[str] = None
url: Optional[str] = None
@dataclass
class Info:
title: str
version: str
description: Optional[str] = None
termsOfService: Optional[str] = None
contact: Optional[Contact] = None
license: Optional[License] = None
# Servers ----------------------------------------------------------------------
@dataclass
class ServerVariable:
enum: Optional[list[str]] = None
default: str = ""
description: Optional[str] = None
@dataclass
class Server:
url: str
description: Optional[str] = None
variables: Optional[dict[str, ServerVariable]] = None
# External Docs ----------------------------------------------------------------
@dataclass
class ExternalDocumentation:
url: str = ""
description: Optional[str] = None
# Tags -------------------------------------------------------------------------
@dataclass
class Tag:
name: str = ""
description: Optional[str] = None
externalDocs: Optional[ExternalDocumentation] = None
# Reference --------------------------------------------------------------------
@dataclass
class Reference:
ref: str = field(metadata={ALIAS: "$ref"})
summary: Optional[str] = None
description: Optional[str] = None
# XML Object (for XML-specific representations) --------------------------------
@dataclass
class XML:
name: Optional[str] = None
namespace: Optional[str] = None
prefix: Optional[str] = None
attribute: Optional[bool] = None
wrapped: Optional[bool] = None
# Schema (JSON Schema vocab with OAS extensions) --------------------------------
SchemaType: TypeAlias = Literal[
"null", "boolean", "object", "array", "number", "string", "integer"
]
@dataclass
class Discriminator:
propertyName: str = ""
mapping: Optional[dict[str, str]] = None
@dataclass
class Schema:
# Core JSON Schema
type: Optional[SchemaType] = None
enum: Optional[list[JsonValue]] = None
const: Optional[JsonValue] = None
multipleOf: Optional[float] = None
maximum: Optional[float] = None
exclusiveMaximum: Optional[float] = None
minimum: Optional[float] = None
exclusiveMinimum: Optional[float] = None
maxLength: Optional[int] = None
minLength: Optional[int] = None
pattern: Optional[str] = None
items: Optional[Union[Schema, list[Schema]]] = None
prefixItems: Optional[list[Schema]] = None
additionalItems: Optional[Union[bool, Schema]] = None
unevaluatedItems: Optional[Union[bool, Schema]] = None
maxItems: Optional[int] = None
minItems: Optional[int] = None
uniqueItems: Optional[bool] = None
contains: Optional[Schema] = None
maxProperties: Optional[int] = None
minProperties: Optional[int] = None
required: Optional[list[str]] = None
properties: Optional[dict[str, Schema]] = None
patternProperties: Optional[dict[str, Schema]] = None
additionalProperties: Optional[Union[bool, Schema]] = None
unevaluatedProperties: Optional[Union[bool, Schema]] = None
dependentRequired: Optional[dict[str, list[str]]] = None
dependentSchemas: Optional[dict[str, Schema]] = None
propertyNames: Optional[Schema] = None
allOf: Optional[list[Schema]] = None
anyOf: Optional[list[Schema]] = None
oneOf: Optional[list[Schema]] = None
not_: Optional[Schema] = field(default=None, metadata={ALIAS: "not"})
if_: Optional[Schema] = field(default=None, metadata={ALIAS: "if"})
then: Optional[Schema] = None
else_: Optional[Schema] = field(default=None, metadata={ALIAS: "else"})
format: Optional[str] = None
contentEncoding: Optional[str] = None
contentMediaType: Optional[str] = None
contentSchema: Optional[Schema] = None
examples: Optional[list[JsonValue]] = None
default: Optional[JsonValue] = None
title: Optional[str] = None
description: Optional[str] = None
deprecated: Optional[bool] = None
readOnly: Optional[bool] = None
writeOnly: Optional[bool] = None
# OAS schema add-ons
discriminator: Optional[Discriminator] = None
xml: Optional[XML] = None
externalDocs: Optional[ExternalDocumentation] = None
# Example ----------------------------------------------------------------------
@dataclass
class Example:
summary: Optional[str] = None
description: Optional[str] = None
value: Optional[JsonValue] = None
externalValue: Optional[str] = None
# Encoding (for multipart/form-data etc.) --------------------------------------
@dataclass
class Header:
description: Optional[str] = None
required: Optional[bool] = None
deprecated: Optional[bool] = None
allowEmptyValue: Optional[bool] = None
style: Optional[str] = None
explode: Optional[bool] = None
allowReserved: Optional[bool] = None
schema: Optional[Union[Schema, Reference]] = None
example: Optional[JsonValue] = None
examples: Optional[dict[str, Union[Example, Reference]]] = None
content: Optional[dict[str, "MediaType"]] = None
@dataclass
class Encoding:
contentType: Optional[str] = None
headers: Optional[dict[str, Union[Header, Reference]]] = None
style: Optional[str] = None
explode: Optional[bool] = None
allowReserved: Optional[bool] = None
# Media types / Request & Response bodies --------------------------------------
@dataclass
class MediaType:
schema: Optional[Union[Schema, Reference]] = None
example: Optional[JsonValue] = None
examples: Optional[dict[str, Union[Example, Reference]]] = None
encoding: Optional[dict[str, Encoding]] = None
@dataclass
class RequestBody:
content: dict[str, MediaType] = field(default_factory=dict)
description: Optional[str] = None
required: Optional[bool] = None
@dataclass
class Response:
description: str = ""
headers: Optional[dict[str, Union[Header, Reference]]] = None
content: Optional[dict[str, MediaType]] = None
links: Optional[dict[str, Union["Link", Reference]]] = None
@dataclass
class Responses:
# keys are HTTP status codes or "default"
_map: dict[str, Union[Response, Reference]] = field(default_factory=dict, metadata={ALIAS: ""})
# Links / Callbacks -------------------------------------------------------------
@dataclass
class Link:
operationId: Optional[str] = None
operationRef: Optional[str] = None
parameters: Optional[dict[str, JsonValue]] = None
requestBody: Optional[JsonValue] = None
description: Optional[str] = None
server: Optional[Server] = None
@dataclass
class Callback:
# expression -> PathItem
_map: dict[str, "PathItem"] = field(default_factory=dict, metadata={ALIAS: ""})
# Parameters -------------------------------------------------------------------
ParameterLocation: TypeAlias = Literal["query", "header", "path", "cookie"]
@dataclass
class Parameter:
name: str = ""
in_: ParameterLocation = field(default="query", metadata={ALIAS: "in"})
description: Optional[str] = None
required: Optional[bool] = None
deprecated: Optional[bool] = None
allowEmptyValue: Optional[bool] = None
style: Optional[str] = None
explode: Optional[bool] = None
allowReserved: Optional[bool] = None
schema: Optional[Union[Schema, Reference]] = None
example: Optional[JsonValue] = None
examples: Optional[dict[str, Union[Example, Reference]]] = None
content: Optional[dict[str, MediaType]] = None
# Operation / PathItem ----------------------------------------------------------
@dataclass
class Operation:
responses: dict[str, Union[Response, Reference]] = field(default_factory=dict)
summary: Optional[str] = None
description: Optional[str] = None
operationId: Optional[str] = None
parameters: list[Union[Parameter, Reference]] = field(default_factory=list)
requestBody: Optional[Union[RequestBody, Reference]] = None
callbacks: dict[str, Union[Callback, Reference]] = field(default_factory=dict)
deprecated: Optional[bool] = None
security: list["SecurityRequirement"] = field(default_factory=list)
servers: list[Server] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
externalDocs: Optional[ExternalDocumentation] = None
@dataclass
class PathItem:
summary: Optional[str] = None
description: Optional[str] = None
get: Optional[Operation] = None
put: Optional[Operation] = None
post: Optional[Operation] = None
delete: Optional[Operation] = None
options: Optional[Operation] = None
head: Optional[Operation] = None
patch: Optional[Operation] = None
trace: Optional[Operation] = None
servers: Optional[list[Server]] = None
parameters: Optional[list[Union[Parameter, Reference]]] = None
# Security (incl. OAuth2) ------------------------------------------------------
SecuritySchemeType: TypeAlias = Literal["apiKey", "http", "mutualTLS", "oauth2", "openIdConnect"]
ApiKeyLocation: TypeAlias = Literal["query", "header", "cookie"]
@dataclass
class OAuthFlow:
authorizationUrl: Optional[str] = None
tokenUrl: Optional[str] = None
refreshUrl: Optional[str] = None
scopes: dict[str, str] = field(default_factory=dict)
@dataclass
class OAuthFlows:
implicit: Optional[OAuthFlow] = None
password: Optional[OAuthFlow] = None
clientCredentials: Optional[OAuthFlow] = None
authorizationCode: Optional[OAuthFlow] = None
@dataclass
class SecurityScheme:
type: SecuritySchemeType = "http"
description: Optional[str] = None
# apiKey
name: Optional[str] = None
in_: Optional[ApiKeyLocation] = field(default=None, metadata={ALIAS: "in"})
# http
scheme: Optional[str] = None
bearerFormat: Optional[str] = None
# mutualTLS has no extra fields
# oauth2
flows: Optional[OAuthFlows] = None
# openIdConnect
openIdConnectUrl: Optional[str] = None
# A single security requirement (scheme name -> required scopes)
SecurityRequirement: TypeAlias = dict[str, list[str]]

7
src/turbosloth/interfaces/serialized/base.py
View File

@@ -1,5 +1,5 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass, field
from typing import Any, Mapping from typing import Any, Mapping
from case_insensitive_dict import CaseInsensitiveDict from case_insensitive_dict import CaseInsensitiveDict
@@ -41,10 +41,9 @@ class SerializedRequest:
@dataclass @dataclass
class SerializedResponse: class SerializedResponse:
code: int
headers: Mapping[str, str]
body: dict[str, Any] | list[Any] | str body: dict[str, Any] | list[Any] | str
content_type: str = '' code: int = 200
headers: Mapping[str, str] = field(default_factory=dict)
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
raise NotImplementedError() raise NotImplementedError()

3
src/turbosloth/interfaces/serialized/html.py
View File

@@ -9,9 +9,8 @@ from .base import SerializedResponse
@dataclass @dataclass
class HTMLSerializedResponse(SerializedResponse): class HTMLSerializedResponse(SerializedResponse):
body: str body: str
content_type = 'text/html'
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
headers = CaseInsensitiveDict(self.headers).copy() headers = CaseInsensitiveDict(self.headers).copy()
headers['content-type'] = self.content_type + '; charset=' + charset headers['content-type'] = 'text/html; charset=' + charset
return BasicResponse(self.code, headers, str(self.body).encode(charset)) return BasicResponse(self.code, headers, str(self.body).encode(charset))

3
src/turbosloth/interfaces/serialized/json.py
View File

@@ -19,10 +19,9 @@ class JsonSerializedRequest(SerializedRequest):
class JsonSerializedResponse(SerializedResponse): class JsonSerializedResponse(SerializedResponse):
content_type = 'application/json'
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
headers = CaseInsensitiveDict(self.headers).copy() headers = CaseInsensitiveDict(self.headers).copy()
headers['content-type'] = self.content_type + '; charset=utf-8' headers['content-type'] = 'application/json; charset=utf-8'
b = orjson.dumps(self.body) b = orjson.dumps(self.body)
return BasicResponse(self.code, headers, b) return BasicResponse(self.code, headers, b)

3
src/turbosloth/interfaces/serialized/msgpack.py
View File

@@ -16,10 +16,9 @@ class MessagePackSerializedRequest(SerializedRequest):
class MessagePackSerializedResponse(SerializedResponse): class MessagePackSerializedResponse(SerializedResponse):
content_type = 'application/vnd.msgpack'
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
headers = CaseInsensitiveDict(self.headers).copy() headers = CaseInsensitiveDict(self.headers).copy()
headers['content-type'] = self.content_type headers['content-type'] = 'application/vnd.msgpack'
b = msgpack.packb(self.body) b = msgpack.packb(self.body)
return BasicResponse(self.code, headers, b) return BasicResponse(self.code, headers, b)

3
src/turbosloth/interfaces/serialized/text.py
View File

@@ -12,9 +12,8 @@ class TextSerializedRequest(SerializedRequest):
class TextSerializedResponse(SerializedResponse): class TextSerializedResponse(SerializedResponse):
content_type = 'text/plain'
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
headers = CaseInsensitiveDict(self.headers).copy() headers = CaseInsensitiveDict(self.headers).copy()
headers['content-type'] = self.content_type + '; charset=' + charset headers['content-type'] = 'text/plain; charset=' + charset
return BasicResponse(self.code, headers, str(self.body).encode(charset)) return BasicResponse(self.code, headers, str(self.body).encode(charset))

3
src/turbosloth/interfaces/serialized/xml.py
View File

@@ -21,11 +21,10 @@ class XMLSerializedRequest(SerializedRequest):
class XMLSerializedResponse(SerializedResponse): class XMLSerializedResponse(SerializedResponse):
content_type = 'application/xml'
def into_basic(self, charset: str) -> BasicResponse: def into_basic(self, charset: str) -> BasicResponse:
headers = CaseInsensitiveDict(self.headers).copy() headers = CaseInsensitiveDict(self.headers).copy()
headers['content-type'] = self.content_type + '; charset=' + charset headers['content-type'] = 'application/xml; charset=' + charset
btxt = _into_xml(self.body) btxt = _into_xml(self.body)