diff --git a/src/turbosloth/__main__.py b/src/turbosloth/__main__.py index 32817bb..d02ed2e 100644 --- a/src/turbosloth/__main__.py +++ b/src/turbosloth/__main__.py @@ -1,19 +1,21 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Any, Optional +from typing import Optional import uvicorn 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.serialized import SerializedResponse, SerializedRequest -from turbosloth.internal_types import QTYPE, BTYPE, PTYPE, HTYPE -from turbosloth.req_schema import UnwrappedRequest -from turbosloth.schema import RequestBody, HeaderParam, PathParam +from turbosloth.interfaces.serialized import SerializedResponse +from turbosloth.schema import RequestBody, HeaderParam, QueryParam 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("/") @@ -117,7 +119,41 @@ async def test_body(r: RequestBody(UserPostSchema), 'user': user.__dict__, '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 = ''' + + + + + Elements in HTML + + + + + + + + + + +''' + return BasicResponse(200, {}, ret.encode()) if __name__ == '__main__': diff --git a/src/turbosloth/app.py b/src/turbosloth/app.py index c7ae850..ba37e35 100644 --- a/src/turbosloth/app.py +++ b/src/turbosloth/app.py @@ -2,8 +2,9 @@ from __future__ import annotations import html import typing 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, \ - Iterable + Iterable, TypeAliasType, Union import breakshaft.util_mermaid import megasniff.exceptions @@ -11,6 +12,8 @@ from breakshaft.models import ConversionPoint, Callgraph from megasniff import SchemaInflatorGenerator 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 .interfaces.base import BasicRequest, BasicResponse from .interfaces.serialize_selector import SerializeSelector @@ -125,9 +128,9 @@ class HTTPApp(ASGIApp): return except (megasniff.exceptions.FieldValidationException, megasniff.exceptions.MissingFieldException) as e: print(e) - sresp = SerializedResponse(400, {}, 'Schema error') + sresp = SerializedResponse('Schema error', code=400) except HTTPException as e: - sresp = SerializedResponse(e.code, {}, str(e)) + sresp = SerializedResponse(str(e), code=e.code) try: ct = self.extract_content_type(req) @@ -207,13 +210,15 @@ class MethodRoutersApp(ASGIApp): class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp): - di_autodoc_prefix: Optional[str] = None + di_autodoc_prefix: Optional[str] + openapi_app: Optional[OpenAPIApp] def __init__(self, on_startup: Optional[Callable[[], Awaitable[None]]] = None, on_shutdown: Optional[Callable[[], Awaitable[None]]] = None, di_autodoc_prefix: Optional[str] = None, serialize_selector: Optional[SerializeSelector] = None, + openapi_app: Optional[OpenAPIApp] = None ): self.router = Router() self._on_startup = on_startup @@ -223,11 +228,12 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp): self.serialize_selector = serialize_selector self.infl_generator = SchemaInflatorGenerator(strict_mode=True, store_sources=True) self.di_autodoc_prefix = di_autodoc_prefix + self.openapi_app = openapi_app if di_autodoc_prefix is not None: self.inj_repo = ConvRepo(store_sources=True, store_callseq=True) else: - self.inj_repo = ConvRepo() + self.inj_repo = ConvRepo(store_callseq=True) @self.inj_repo.mark_injector() def extract_query(req: BasicRequest | SerializedRequest) -> QTYPE: @@ -245,6 +251,10 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp): def extract_body(req: SerializedRequest) -> BTYPE: 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_accept_type) 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 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 decorator(fn: HandlerType): 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)( 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 decorator @@ -379,6 +563,9 @@ class SlothApp(HTTPApp, WSApp, LifespanApp, MethodRoutersApp): setattr(func, '__turbosloth_DI_name__', breakshaft.util.universal_qualname(main_injects)) fork_with |= set(ConversionPoint.from_fn(func, type_remap=fn_type_hints)) + + setattr(func, '__turbosloth_config__', config) + return fork_with, fn_type_hints def add_injector(self, func: Callable): diff --git a/src/turbosloth/doc/openapi_app.py b/src/turbosloth/doc/openapi_app.py new file mode 100644 index 0000000..f72750c --- /dev/null +++ b/src/turbosloth/doc/openapi_app.py @@ -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) diff --git a/src/turbosloth/doc/openapi_models.py b/src/turbosloth/doc/openapi_models.py new file mode 100644 index 0000000..1c92b8b --- /dev/null +++ b/src/turbosloth/doc/openapi_models.py @@ -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]] diff --git a/src/turbosloth/interfaces/serialized/base.py b/src/turbosloth/interfaces/serialized/base.py index 2e1d130..4ac4cd4 100644 --- a/src/turbosloth/interfaces/serialized/base.py +++ b/src/turbosloth/interfaces/serialized/base.py @@ -1,5 +1,5 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Mapping from case_insensitive_dict import CaseInsensitiveDict @@ -41,10 +41,9 @@ class SerializedRequest: @dataclass class SerializedResponse: - code: int - headers: Mapping[str, 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: raise NotImplementedError() diff --git a/src/turbosloth/interfaces/serialized/html.py b/src/turbosloth/interfaces/serialized/html.py index 4e3884a..da77aa2 100644 --- a/src/turbosloth/interfaces/serialized/html.py +++ b/src/turbosloth/interfaces/serialized/html.py @@ -9,9 +9,8 @@ from .base import SerializedResponse @dataclass class HTMLSerializedResponse(SerializedResponse): body: str - content_type = 'text/html' def into_basic(self, charset: str) -> BasicResponse: 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)) diff --git a/src/turbosloth/interfaces/serialized/json.py b/src/turbosloth/interfaces/serialized/json.py index dba941d..47b080e 100644 --- a/src/turbosloth/interfaces/serialized/json.py +++ b/src/turbosloth/interfaces/serialized/json.py @@ -19,10 +19,9 @@ class JsonSerializedRequest(SerializedRequest): class JsonSerializedResponse(SerializedResponse): - content_type = 'application/json' def into_basic(self, charset: str) -> BasicResponse: 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) return BasicResponse(self.code, headers, b) diff --git a/src/turbosloth/interfaces/serialized/msgpack.py b/src/turbosloth/interfaces/serialized/msgpack.py index 216bc73..396657e 100644 --- a/src/turbosloth/interfaces/serialized/msgpack.py +++ b/src/turbosloth/interfaces/serialized/msgpack.py @@ -16,10 +16,9 @@ class MessagePackSerializedRequest(SerializedRequest): class MessagePackSerializedResponse(SerializedResponse): - content_type = 'application/vnd.msgpack' def into_basic(self, charset: str) -> BasicResponse: headers = CaseInsensitiveDict(self.headers).copy() - headers['content-type'] = self.content_type + headers['content-type'] = 'application/vnd.msgpack' b = msgpack.packb(self.body) return BasicResponse(self.code, headers, b) diff --git a/src/turbosloth/interfaces/serialized/text.py b/src/turbosloth/interfaces/serialized/text.py index e8acaf1..0d59e3c 100644 --- a/src/turbosloth/interfaces/serialized/text.py +++ b/src/turbosloth/interfaces/serialized/text.py @@ -12,9 +12,8 @@ class TextSerializedRequest(SerializedRequest): class TextSerializedResponse(SerializedResponse): - content_type = 'text/plain' def into_basic(self, charset: str) -> BasicResponse: 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)) diff --git a/src/turbosloth/interfaces/serialized/xml.py b/src/turbosloth/interfaces/serialized/xml.py index 24828a9..39dc260 100644 --- a/src/turbosloth/interfaces/serialized/xml.py +++ b/src/turbosloth/interfaces/serialized/xml.py @@ -21,11 +21,10 @@ class XMLSerializedRequest(SerializedRequest): class XMLSerializedResponse(SerializedResponse): - content_type = 'application/xml' def into_basic(self, charset: str) -> BasicResponse: 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)