From 8d48b3d77190e193c41bf51ea1121157adb55dc0 Mon Sep 17 00:00:00 2001 From: Eric Nielsen <4120606+ericbn@users.noreply.github.com> Date: Thu, 15 Aug 2024 11:08:11 -0500 Subject: [PATCH 1/4] refactor(openapi): add from __future__ import annotations and update code according to ruff rules TCH, UP006, UP007, UP037 and FA100. --- .../event_handler/openapi/compat.py | 60 +-- .../event_handler/openapi/dependant.py | 39 +- .../event_handler/openapi/encoders.py | 52 ++- .../event_handler/openapi/models.py | 347 +++++++------- .../event_handler/openapi/params.py | 426 +++++++++--------- .../event_handler/openapi/swagger_ui/html.py | 9 +- .../openapi/swagger_ui/oauth2.py | 14 +- .../event_handler/openapi/types.py | 18 +- 8 files changed, 492 insertions(+), 473 deletions(-) diff --git a/aws_lambda_powertools/event_handler/openapi/compat.py b/aws_lambda_powertools/event_handler/openapi/compat.py index c55cc55c333..03e9383dbcb 100644 --- a/aws_lambda_powertools/event_handler/openapi/compat.py +++ b/aws_lambda_powertools/event_handler/openapi/compat.py @@ -1,5 +1,7 @@ # mypy: ignore-errors # flake8: noqa +from __future__ import annotations + from collections import deque from copy import copy @@ -8,7 +10,7 @@ from dataclasses import dataclass, is_dataclass from enum import Enum -from typing import Any, Dict, List, Set, Tuple, Type, Union, FrozenSet, Deque, Sequence, Mapping +from typing import Any, Deque, FrozenSet, List, Mapping, Sequence, Set, Tuple, Union from typing_extensions import Annotated, Literal, get_origin, get_args @@ -56,7 +58,7 @@ sequence_types = tuple(sequence_annotation_to_type.keys()) -RequestErrorModel: Type[BaseModel] = create_model("Request") +RequestErrorModel: type[BaseModel] = create_model("Request") class ErrorWrapper(Exception): @@ -101,8 +103,8 @@ def serialize( value: Any, *, mode: Literal["json", "python"] = "json", - include: Union[IncEx, None] = None, - exclude: Union[IncEx, None] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, @@ -120,8 +122,8 @@ def serialize( ) def validate( - self, value: Any, values: Dict[str, Any] = {}, *, loc: Tuple[Union[int, str], ...] = () - ) -> Tuple[Any, Union[List[Dict[str, Any]], None]]: + self, value: Any, values: dict[str, Any] = {}, *, loc: tuple[int | str, ...] = () + ) -> tuple[Any, list[dict[str, Any]] | None]: try: return (self._type_adapter.validate_python(value, from_attributes=True), None) except ValidationError as exc: @@ -136,11 +138,11 @@ def get_schema_from_model_field( *, field: ModelField, model_name_map: ModelNameMap, - field_mapping: Dict[ - Tuple[ModelField, Literal["validation", "serialization"]], + field_mapping: dict[ + tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue, ], -) -> Dict[str, Any]: +) -> dict[str, Any]: json_schema = field_mapping[(field, field.mode)] if "$ref" not in json_schema: # MAINTENANCE: remove when deprecating Pydantic v1 @@ -151,15 +153,15 @@ def get_schema_from_model_field( def get_definitions( *, - fields: List[ModelField], + fields: list[ModelField], schema_generator: GenerateJsonSchema, model_name_map: ModelNameMap, -) -> Tuple[ - Dict[ - Tuple[ModelField, Literal["validation", "serialization"]], - Dict[str, Any], +) -> tuple[ + dict[ + tuple[ModelField, Literal["validation", "serialization"]], + dict[str, Any], ], - Dict[str, Dict[str, Any]], + dict[str, dict[str, Any]], ]: inputs = [(field, field.mode, field._type_adapter.core_schema) for field in fields] field_mapping, definitions = schema_generator.generate_definitions(inputs=inputs) @@ -167,7 +169,7 @@ def get_definitions( return field_mapping, definitions -def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap: +def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap: return {} @@ -175,7 +177,7 @@ def get_annotation_from_field_info(annotation: Any, field_info: FieldInfo, field return annotation -def model_rebuild(model: Type[BaseModel]) -> None: +def model_rebuild(model: type[BaseModel]) -> None: model.model_rebuild() @@ -183,7 +185,7 @@ def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo: return type(field_info).from_annotation(annotation) -def get_missing_field_error(loc: Tuple[str, ...]) -> Dict[str, Any]: +def get_missing_field_error(loc: tuple[str, ...]) -> dict[str, Any]: error = ValidationError.from_exception_data( "Field required", [{"type": "missing", "loc": loc, "input": {}}] ).errors()[0] @@ -220,13 +222,13 @@ def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]: return sequence_annotation_to_type[origin_type](value) # type: ignore[no-any-return] -def _normalize_errors(errors: Sequence[Any]) -> List[Dict[str, Any]]: +def _normalize_errors(errors: Sequence[Any]) -> list[dict[str, Any]]: return errors # type: ignore[return-value] -def create_body_model(*, fields: Sequence[ModelField], model_name: str) -> Type[BaseModel]: +def create_body_model(*, fields: Sequence[ModelField], model_name: str) -> type[BaseModel]: field_params = {f.name: (f.field_info.annotation, f.field_info) for f in fields} - model: Type[BaseModel] = create_model(model_name, **field_params) + model: type[BaseModel] = create_model(model_name, **field_params) return model @@ -241,7 +243,7 @@ def model_json(model: BaseModel, **kwargs: Any) -> Any: # Common code for both versions -def field_annotation_is_complex(annotation: Union[Type[Any], None]) -> bool: +def field_annotation_is_complex(annotation: type[Any] | None) -> bool: origin = get_origin(annotation) if origin is Union or origin is UnionType: return any(field_annotation_is_complex(arg) for arg in get_args(annotation)) @@ -258,11 +260,11 @@ def field_annotation_is_scalar(annotation: Any) -> bool: return annotation is Ellipsis or not field_annotation_is_complex(annotation) -def field_annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool: +def field_annotation_is_sequence(annotation: type[Any] | None) -> bool: return _annotation_is_sequence(annotation) or _annotation_is_sequence(get_origin(annotation)) -def field_annotation_is_scalar_sequence(annotation: Union[Type[Any], None]) -> bool: +def field_annotation_is_scalar_sequence(annotation: type[Any] | None) -> bool: origin = get_origin(annotation) if origin is Union or origin is UnionType: at_least_one_scalar_sequence = False @@ -307,7 +309,7 @@ def value_is_sequence(value: Any) -> bool: return isinstance(value, sequence_types) and not isinstance(value, (str, bytes)) # type: ignore[arg-type] -def _annotation_is_complex(annotation: Union[Type[Any], None]) -> bool: +def _annotation_is_complex(annotation: type[Any] | None) -> bool: return ( lenient_issubclass(annotation, (BaseModel, Mapping)) # TODO: UploadFile or _annotation_is_sequence(annotation) @@ -315,16 +317,14 @@ def _annotation_is_complex(annotation: Union[Type[Any], None]) -> bool: ) -def _annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool: +def _annotation_is_sequence(annotation: type[Any] | None) -> bool: if lenient_issubclass(annotation, (str, bytes)): return False return lenient_issubclass(annotation, sequence_types) -def _regenerate_error_with_loc( - *, errors: Sequence[Any], loc_prefix: Tuple[Union[str, int], ...] -) -> List[Dict[str, Any]]: - updated_loc_errors: List[Any] = [ +def _regenerate_error_with_loc(*, errors: Sequence[Any], loc_prefix: tuple[str | int, ...]) -> list[dict[str, Any]]: + updated_loc_errors: list[Any] = [ {**err, "loc": loc_prefix + err.get("loc", ())} for err in _normalize_errors(errors) ] diff --git a/aws_lambda_powertools/event_handler/openapi/dependant.py b/aws_lambda_powertools/event_handler/openapi/dependant.py index abcb91e90dd..e4f2f822ce1 100644 --- a/aws_lambda_powertools/event_handler/openapi/dependant.py +++ b/aws_lambda_powertools/event_handler/openapi/dependant.py @@ -1,8 +1,8 @@ +from __future__ import annotations + import inspect import re -from typing import Any, Callable, Dict, ForwardRef, List, Optional, Set, Tuple, Type, cast - -from pydantic import BaseModel +from typing import TYPE_CHECKING, Any, Callable, ForwardRef, cast from aws_lambda_powertools.event_handler.openapi.compat import ( ModelField, @@ -26,6 +26,9 @@ ) from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse, OpenAPIResponseContentModel +if TYPE_CHECKING: + from pydantic import BaseModel + """ This turns the opaque function signature into typed, validated models. @@ -76,7 +79,7 @@ def add_param_to_fields( raise AssertionError(f"Unsupported param type: {field_info.in_}") -def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any: +def get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any: """ Evaluates a type annotation, which can be a string or a ForwardRef. """ @@ -128,7 +131,7 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature: return inspect.Signature(typed_params) -def get_path_param_names(path: str) -> Set[str]: +def get_path_param_names(path: str) -> set[str]: """ Returns the path parameter names from a path template. Those are the strings between { and }. @@ -139,7 +142,7 @@ def get_path_param_names(path: str) -> Set[str]: Returns ------- - Set[str] + set[str] The path parameter names """ @@ -150,8 +153,8 @@ def get_dependant( *, path: str, call: Callable[..., Any], - name: Optional[str] = None, - responses: Optional[Dict[int, OpenAPIResponse]] = None, + name: str | None = None, + responses: dict[int, OpenAPIResponse] | None = None, ) -> Dependant: """ Returns a dependant model for a handler function. A dependant model is a model that contains @@ -165,7 +168,7 @@ def get_dependant( The handler function name: str, optional The name of the handler function - responses: List[Dict[int, OpenAPIResponse]], optional + responses: list[dict[int, OpenAPIResponse]], optional The list of extra responses for the handler function Returns @@ -210,7 +213,7 @@ def get_dependant( return dependant -def _add_extra_responses(dependant: Dependant, responses: Optional[Dict[int, OpenAPIResponse]]): +def _add_extra_responses(dependant: Dependant, responses: dict[int, OpenAPIResponse] | None): # Also add the optional extra responses to the dependant model. if not responses: return @@ -278,7 +281,7 @@ def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool: return True -def get_flat_params(dependant: Dependant) -> List[ModelField]: +def get_flat_params(dependant: Dependant) -> list[ModelField]: """ Get a list of all the parameters from a Dependant object. @@ -289,7 +292,7 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]: Returns ------- - List[ModelField] + list[ModelField] A list of ModelField objects containing the flat parameters from the Dependant object. """ @@ -302,7 +305,7 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]: ) -def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]: +def get_body_field(*, dependant: Dependant, name: str) -> ModelField | None: """ Get the Body field for a given Dependant object. """ @@ -348,24 +351,24 @@ def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]: def get_body_field_info( *, - body_model: Type[BaseModel], + body_model: type[BaseModel], flat_dependant: Dependant, required: bool, -) -> Tuple[Type[Body], Dict[str, Any]]: +) -> tuple[type[Body], dict[str, Any]]: """ Get the Body field info and kwargs for a given body model. """ - body_field_info_kwargs: Dict[str, Any] = {"annotation": body_model, "alias": "body"} + body_field_info_kwargs: dict[str, Any] = {"annotation": body_model, "alias": "body"} if not required: body_field_info_kwargs["default"] = None if any(isinstance(f.field_info, _File) for f in flat_dependant.body_params): - # MAINTENANCE: body_field_info: Type[Body] = _File + # MAINTENANCE: body_field_info: type[Body] = _File raise NotImplementedError("_File fields are not supported in request bodies") elif any(isinstance(f.field_info, _Form) for f in flat_dependant.body_params): - # MAINTENANCE: body_field_info: Type[Body] = _Form + # MAINTENANCE: body_field_info: type[Body] = _Form raise NotImplementedError("_Form fields are not supported in request bodies") else: body_field_info = Body diff --git a/aws_lambda_powertools/event_handler/openapi/encoders.py b/aws_lambda_powertools/event_handler/openapi/encoders.py index 846d8030baf..9279e6f27c2 100644 --- a/aws_lambda_powertools/event_handler/openapi/encoders.py +++ b/aws_lambda_powertools/event_handler/openapi/encoders.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import dataclasses import datetime from collections import defaultdict, deque @@ -6,14 +8,16 @@ from pathlib import Path, PurePath from re import Pattern from types import GeneratorType -from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union +from typing import TYPE_CHECKING, Any, Callable from uuid import UUID from pydantic import BaseModel from pydantic.types import SecretBytes, SecretStr from aws_lambda_powertools.event_handler.openapi.compat import _model_dump -from aws_lambda_powertools.event_handler.openapi.types import IncEx + +if TYPE_CHECKING: + from aws_lambda_powertools.event_handler.openapi.types import IncEx """ This module contains the encoders used by jsonable_encoder to convert Python objects to JSON serializable data types. @@ -22,13 +26,13 @@ def jsonable_encoder( # noqa: PLR0911 obj: Any, - include: Optional[IncEx] = None, - exclude: Optional[IncEx] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, - custom_serializer: Optional[Callable[[Any], str]] = None, + custom_serializer: Callable[[Any], str] | None = None, ) -> Any: """ JSON encodes an arbitrary Python object into JSON serializable data types. @@ -40,10 +44,10 @@ def jsonable_encoder( # noqa: PLR0911 ---------- obj : Any The object to encode - include : Optional[IncEx], optional + include : IncEx | None, optional A set or dictionary of strings that specifies which properties should be included, by default None, meaning everything is included - exclude : Optional[IncEx], optional + exclude : IncEx | None, optional A set or dictionary of strings that specifies which properties should be excluded, by default None, meaning nothing is excluded by_alias : bool, optional @@ -155,8 +159,8 @@ def jsonable_encoder( # noqa: PLR0911 def _dump_base_model( *, obj: Any, - include: Optional[IncEx] = None, - exclude: Optional[IncEx] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_none: bool = False, @@ -188,12 +192,12 @@ def _dump_base_model( def _dump_dict( *, obj: Any, - include: Optional[IncEx] = None, - exclude: Optional[IncEx] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_none: bool = False, -) -> Dict[str, Any]: +) -> dict[str, Any]: """ Dump a dict to a dict, using the same parameters as jsonable_encoder """ @@ -228,13 +232,13 @@ def _dump_dict( def _dump_sequence( *, obj: Any, - include: Optional[IncEx] = None, - exclude: Optional[IncEx] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_none: bool = False, exclude_defaults: bool = False, -) -> List[Any]: +) -> list[Any]: """ Dump a sequence to a list, using the same parameters as jsonable_encoder """ @@ -257,8 +261,8 @@ def _dump_sequence( def _dump_other( *, obj: Any, - include: Optional[IncEx] = None, - exclude: Optional[IncEx] = None, + include: IncEx | None = None, + exclude: IncEx | None = None, by_alias: bool = True, exclude_unset: bool = False, exclude_none: bool = False, @@ -270,7 +274,7 @@ def _dump_other( try: data = dict(obj) except Exception as e: - errors: List[Exception] = [e] + errors: list[Exception] = [e] try: data = vars(obj) except Exception as e: @@ -287,14 +291,14 @@ def _dump_other( ) -def iso_format(o: Union[datetime.date, datetime.time]) -> str: +def iso_format(o: datetime.date | datetime.time) -> str: """ ISO format for date and time """ return o.isoformat() -def decimal_encoder(dec_value: Decimal) -> Union[int, float]: +def decimal_encoder(dec_value: Decimal) -> int | float: """ Encodes a Decimal as int of there's no exponent, otherwise float @@ -315,7 +319,7 @@ def decimal_encoder(dec_value: Decimal) -> Union[int, float]: # Encoders for types that are not JSON serializable -ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = { +ENCODERS_BY_TYPE: dict[type[Any], Callable[[Any], Any]] = { bytes: lambda o: o.decode(), datetime.date: iso_format, datetime.datetime: iso_format, @@ -337,9 +341,9 @@ def decimal_encoder(dec_value: Decimal) -> Union[int, float]: # Generates a mapping of encoders to a tuple of classes that they can encode def generate_encoders_by_class_tuples( - type_encoder_map: Dict[Any, Callable[[Any], Any]], -) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]: - encoders: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(tuple) + type_encoder_map: dict[Any, Callable[[Any], Any]], +) -> dict[Callable[[Any], Any], tuple[Any, ...]]: + encoders: dict[Callable[[Any], Any], tuple[Any, ...]] = defaultdict(tuple) for type_, encoder in type_encoder_map.items(): encoders[encoder] += (type_,) return encoders diff --git a/aws_lambda_powertools/event_handler/openapi/models.py b/aws_lambda_powertools/event_handler/openapi/models.py index 57310fd3267..0be1d951573 100644 --- a/aws_lambda_powertools/event_handler/openapi/models.py +++ b/aws_lambda_powertools/event_handler/openapi/models.py @@ -1,8 +1,9 @@ +from __future__ import annotations + from enum import Enum -from typing import Any, Dict, List, Literal, Optional, Set, Union +from typing import Any, Literal from pydantic import AnyUrl, BaseModel, Field -from typing_extensions import Annotated from aws_lambda_powertools.event_handler.openapi.compat import model_rebuild from aws_lambda_powertools.event_handler.openapi.constants import ( @@ -18,9 +19,9 @@ # https://swagger.io/specification/#contact-object class Contact(BaseModel): - name: Optional[str] = None - url: Optional[AnyUrl] = None - email: Optional[str] = None + name: str | None = None + url: AnyUrl | None = None + email: str | None = None model_config = MODEL_CONFIG_ALLOW @@ -28,8 +29,8 @@ class Contact(BaseModel): # https://swagger.io/specification/#license-object class License(BaseModel): name: str - identifier: Optional[str] = None - url: Optional[AnyUrl] = None + identifier: str | None = None + url: AnyUrl | None = None model_config = MODEL_CONFIG_ALLOW @@ -37,30 +38,30 @@ class License(BaseModel): # https://swagger.io/specification/#info-object class Info(BaseModel): title: str - description: Optional[str] = None - termsOfService: Optional[str] = None - contact: Optional[Contact] = None - license: Optional[License] = None # noqa: A003 + description: str | None = None + termsOfService: str | None = None + contact: Contact | None = None + license: License | None = None # noqa: A003 version: str - summary: Optional[str] = None + summary: str | None = None model_config = MODEL_CONFIG_IGNORE # https://swagger.io/specification/#server-variable-object class ServerVariable(BaseModel): - enum: Annotated[Optional[List[str]], Field(min_length=1)] = None + enum: list[str] | None = Field(default=None, min_length=1) default: str - description: Optional[str] = None + description: str | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#server-object class Server(BaseModel): - url: Union[AnyUrl, str] - description: Optional[str] = None - variables: Optional[Dict[str, ServerVariable]] = None + url: AnyUrl | str + description: str | None = None + variables: dict[str, ServerVariable] | None = None model_config = MODEL_CONFIG_ALLOW @@ -73,23 +74,23 @@ class Reference(BaseModel): # https://swagger.io/specification/#discriminator-object class Discriminator(BaseModel): propertyName: str - mapping: Optional[Dict[str, str]] = None + mapping: dict[str, str] | None = None # https://swagger.io/specification/#xml-object class XML(BaseModel): - name: Optional[str] = None - namespace: Optional[str] = None - prefix: Optional[str] = None - attribute: Optional[bool] = None - wrapped: Optional[bool] = None + name: str | None = None + namespace: str | None = None + prefix: str | None = None + attribute: bool | None = None + wrapped: bool | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#external-documentation-object class ExternalDocumentation(BaseModel): - description: Optional[str] = None + description: str | None = None url: AnyUrl model_config = MODEL_CONFIG_ALLOW @@ -99,96 +100,96 @@ class ExternalDocumentation(BaseModel): class Schema(BaseModel): # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-the-json-schema-core-vocabu # Core Vocabulary - schema_: Optional[str] = Field(default=None, alias="$schema") - vocabulary: Optional[str] = Field(default=None, alias="$vocabulary") - id: Optional[str] = Field(default=None, alias="$id") # noqa: A003 - anchor: Optional[str] = Field(default=None, alias="$anchor") - dynamicAnchor: Optional[str] = Field(default=None, alias="$dynamicAnchor") - ref: Optional[str] = Field(default=None, alias="$ref") - dynamicRef: Optional[str] = Field(default=None, alias="$dynamicRef") - defs: Optional[Dict[str, "SchemaOrBool"]] = Field(default=None, alias="$defs") - comment: Optional[str] = Field(default=None, alias="$comment") + schema_: str | None = Field(default=None, alias="$schema") + vocabulary: str | None = Field(default=None, alias="$vocabulary") + id: str | None = Field(default=None, alias="$id") # noqa: A003 + anchor: str | None = Field(default=None, alias="$anchor") + dynamicAnchor: str | None = Field(default=None, alias="$dynamicAnchor") + ref: str | None = Field(default=None, alias="$ref") + dynamicRef: str | None = Field(default=None, alias="$dynamicRef") + defs: dict[str, SchemaOrBool] | None = Field(default=None, alias="$defs") + comment: str | None = Field(default=None, alias="$comment") # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-a-vocabulary-for-applying-s # A Vocabulary for Applying Subschemas - allOf: Optional[List["SchemaOrBool"]] = None - anyOf: Optional[List["SchemaOrBool"]] = None - oneOf: Optional[List["SchemaOrBool"]] = None - not_: Optional["SchemaOrBool"] = Field(default=None, alias="not") - if_: Optional["SchemaOrBool"] = Field(default=None, alias="if") - then: Optional["SchemaOrBool"] = None - else_: Optional["SchemaOrBool"] = Field(default=None, alias="else") - dependentSchemas: Optional[Dict[str, "SchemaOrBool"]] = None - prefixItems: Optional[List["SchemaOrBool"]] = None + allOf: list[SchemaOrBool] | None = None + anyOf: list[SchemaOrBool] | None = None + oneOf: list[SchemaOrBool] | None = None + not_: SchemaOrBool | None = Field(default=None, alias="not") + if_: SchemaOrBool | None = Field(default=None, alias="if") + then: SchemaOrBool | None = None + else_: SchemaOrBool | None = Field(default=None, alias="else") + dependentSchemas: dict[str, SchemaOrBool] | None = None + prefixItems: list[SchemaOrBool] | None = None # MAINTENANCE: uncomment and remove below when deprecating Pydantic v1 # MAINTENANCE: It generates a list of schemas for tuples, before prefixItems was available - # MAINTENANCE: items: Optional["SchemaOrBool"] = None - items: Optional[Union["SchemaOrBool", List["SchemaOrBool"]]] = None - contains: Optional["SchemaOrBool"] = None - properties: Optional[Dict[str, "SchemaOrBool"]] = None - patternProperties: Optional[Dict[str, "SchemaOrBool"]] = None - additionalProperties: Optional["SchemaOrBool"] = None - propertyNames: Optional["SchemaOrBool"] = None - unevaluatedItems: Optional["SchemaOrBool"] = None - unevaluatedProperties: Optional["SchemaOrBool"] = None + # MAINTENANCE: items: SchemaOrBool | None = None + items: SchemaOrBool | list[SchemaOrBool] | None = None + contains: SchemaOrBool | None = None + properties: dict[str, SchemaOrBool] | None = None + patternProperties: dict[str, SchemaOrBool] | None = None + additionalProperties: SchemaOrBool | None = None + propertyNames: SchemaOrBool | None = None + unevaluatedItems: SchemaOrBool | None = None + unevaluatedProperties: SchemaOrBool | None = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-structural # A Vocabulary for Structural Validation - type: Optional[str] = None # noqa: A003 - enum: Optional[List[Any]] = None - const: Optional[Any] = None - multipleOf: Optional[float] = Field(default=None, gt=0) - maximum: Optional[float] = None - exclusiveMaximum: Optional[float] = None - minimum: Optional[float] = None - exclusiveMinimum: Optional[float] = None - maxLength: Optional[int] = Field(default=None, ge=0) - minLength: Optional[int] = Field(default=None, ge=0) - pattern: Optional[str] = None - maxItems: Optional[int] = Field(default=None, ge=0) - minItems: Optional[int] = Field(default=None, ge=0) - uniqueItems: Optional[bool] = None - maxContains: Optional[int] = Field(default=None, ge=0) - minContains: Optional[int] = Field(default=None, ge=0) - maxProperties: Optional[int] = Field(default=None, ge=0) - minProperties: Optional[int] = Field(default=None, ge=0) - required: Optional[List[str]] = None - dependentRequired: Optional[Dict[str, Set[str]]] = None + type: str | None = None # noqa: A003 + enum: list[Any] | None = None + const: Any | None = None + multipleOf: float | None = Field(default=None, gt=0) + maximum: float | None = None + exclusiveMaximum: float | None = None + minimum: float | None = None + exclusiveMinimum: float | None = None + maxLength: int | None = Field(default=None, ge=0) + minLength: int | None = Field(default=None, ge=0) + pattern: str | None = None + maxItems: int | None = Field(default=None, ge=0) + minItems: int | None = Field(default=None, ge=0) + uniqueItems: bool | None = None + maxContains: int | None = Field(default=None, ge=0) + minContains: int | None = Field(default=None, ge=0) + maxProperties: int | None = Field(default=None, ge=0) + minProperties: int | None = Field(default=None, ge=0) + required: list[str] | None = None + dependentRequired: dict[str, set[str]] | None = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-vocabularies-for-semantic-c # Vocabularies for Semantic Content With "format" - format: Optional[str] = None # noqa: A003 + format: str | None = None # noqa: A003 # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-the-conten # A Vocabulary for the Contents of String-Encoded Data - contentEncoding: Optional[str] = None - contentMediaType: Optional[str] = None - contentSchema: Optional["SchemaOrBool"] = None + contentEncoding: str | None = None + contentMediaType: str | None = None + contentSchema: SchemaOrBool | None = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-basic-meta # A Vocabulary for Basic Meta-Data Annotations - title: Optional[str] = None - description: Optional[str] = None - default: Optional[Any] = None - deprecated: Optional[bool] = None - readOnly: Optional[bool] = None - writeOnly: Optional[bool] = None - examples: Optional[List["Example"]] = None + title: str | None = None + description: str | None = None + default: Any | None = None + deprecated: bool | None = None + readOnly: bool | None = None + writeOnly: bool | None = None + examples: list[Example] | None = None # Ref: OpenAPI 3.0.0: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.0.0.md#schema-object # Schema Object - discriminator: Optional[Discriminator] = None - xml: Optional[XML] = None - externalDocs: Optional[ExternalDocumentation] = None + discriminator: Discriminator | None = None + xml: XML | None = None + externalDocs: ExternalDocumentation | None = None model_config = MODEL_CONFIG_ALLOW # Ref: https://json-schema.org/draft/2020-12/json-schema-core.html#name-json-schema-documents # A JSON Schema MUST be an object or a boolean. -SchemaOrBool = Union[Schema, bool] +SchemaOrBool = Schema | bool # https://swagger.io/specification/#example-object class Example(BaseModel): - summary: Optional[str] = None - description: Optional[str] = None - value: Optional[Any] = None - externalValue: Optional[AnyUrl] = None + summary: str | None = None + description: str | None = None + value: Any | None = None + externalValue: AnyUrl | None = None model_config = MODEL_CONFIG_ALLOW @@ -202,37 +203,37 @@ class ParameterInType(Enum): # https://swagger.io/specification/#encoding-object class Encoding(BaseModel): - contentType: Optional[str] = None - headers: Optional[Dict[str, Union["Header", Reference]]] = None - style: Optional[str] = None - explode: Optional[bool] = None - allowReserved: Optional[bool] = None + contentType: str | None = None + headers: dict[str, Header | Reference] | None = None + style: str | None = None + explode: bool | None = None + allowReserved: bool | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#media-type-object class MediaType(BaseModel): - schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema") - examples: Optional[Dict[str, Union[Example, Reference]]] = None - encoding: Optional[Dict[str, Encoding]] = None + schema_: Schema | Reference | None = Field(default=None, alias="schema") + examples: dict[str, Example | Reference] | None = None + encoding: dict[str, Encoding] | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#parameter-object class ParameterBase(BaseModel): - description: Optional[str] = None - required: Optional[bool] = None - deprecated: Optional[bool] = None + description: str | None = None + required: bool | None = None + deprecated: bool | None = None # Serialization rules for simple scenarios - style: Optional[str] = None - explode: Optional[bool] = None - allowReserved: Optional[bool] = None - schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema") - examples: Optional[Dict[str, Union[Example, Reference]]] = None + style: str | None = None + explode: bool | None = None + allowReserved: bool | None = None + schema_: Schema | Reference | None = Field(default=None, alias="schema") + examples: dict[str, Example | Reference] | None = None # Serialization rules for more complex scenarios - content: Optional[Dict[str, MediaType]] = None + content: dict[str, MediaType] | None = None model_config = MODEL_CONFIG_ALLOW @@ -248,21 +249,21 @@ class Header(ParameterBase): # https://swagger.io/specification/#request-body-object class RequestBody(BaseModel): - description: Optional[str] = None - content: Dict[str, MediaType] - required: Optional[bool] = None + description: str | None = None + content: dict[str, MediaType] + required: bool | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#link-object class Link(BaseModel): - operationRef: Optional[str] = None - operationId: Optional[str] = None - parameters: Optional[Dict[str, Union[Any, str]]] = None - requestBody: Optional[Union[Any, str]] = None - description: Optional[str] = None - server: Optional[Server] = None + operationRef: str | None = None + operationId: str | None = None + parameters: dict[str, Any | str] | None = None + requestBody: Any | str | None = None + description: str | None = None + server: Server | None = None model_config = MODEL_CONFIG_ALLOW @@ -270,9 +271,9 @@ class Link(BaseModel): # https://swagger.io/specification/#response-object class Response(BaseModel): description: str - headers: Optional[Dict[str, Union[Header, Reference]]] = None - content: Optional[Dict[str, MediaType]] = None - links: Optional[Dict[str, Union[Link, Reference]]] = None + headers: dict[str, Header | Reference] | None = None + content: dict[str, MediaType] | None = None + links: dict[str, Link | Reference] | None = None model_config = MODEL_CONFIG_ALLOW @@ -280,46 +281,46 @@ class Response(BaseModel): # https://swagger.io/specification/#tag-object class Tag(BaseModel): name: str - description: Optional[str] = None - externalDocs: Optional[ExternalDocumentation] = None + description: str | None = None + externalDocs: ExternalDocumentation | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#operation-object class Operation(BaseModel): - tags: Optional[List[str]] = None - summary: Optional[str] = None - description: Optional[str] = None - externalDocs: Optional[ExternalDocumentation] = None - operationId: Optional[str] = None - parameters: Optional[List[Union[Parameter, Reference]]] = None - requestBody: Optional[Union[RequestBody, Reference]] = None + tags: list[str] | None = None + summary: str | None = None + description: str | None = None + externalDocs: ExternalDocumentation | None = None + operationId: str | None = None + parameters: list[Parameter | Reference] | None = None + requestBody: RequestBody | Reference | None = None # Using Any for Specification Extensions - responses: Optional[Dict[int, Union[Response, Any]]] = None - callbacks: Optional[Dict[str, Union[Dict[str, "PathItem"], Reference]]] = None - deprecated: Optional[bool] = None - security: Optional[List[Dict[str, List[str]]]] = None - servers: Optional[List[Server]] = None + responses: dict[int, Response | Any] | None = None + callbacks: dict[str, dict[str, PathItem] | Reference] | None = None + deprecated: bool | None = None + security: list[dict[str, list[str]]] | None = None + servers: list[Server] | None = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#path-item-object class PathItem(BaseModel): - ref: Optional[str] = Field(default=None, alias="$ref") - 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 + ref: str | None = Field(default=None, alias="$ref") + summary: str | None = None + description: str | None = None + get: Operation | None = None + put: Operation | None = None + post: Operation | None = None + delete: Operation | None = None + options: Operation | None = None + head: Operation | None = None + patch: Operation | None = None + trace: Operation | None = None + servers: list[Server] | None = None + parameters: list[Parameter | Reference] | None = None model_config = MODEL_CONFIG_ALLOW @@ -334,7 +335,7 @@ class SecuritySchemeType(Enum): class SecurityBase(BaseModel): type_: SecuritySchemeType = Field(alias="type") - description: Optional[str] = None + description: str | None = None model_config = {"extra": "allow", "populate_by_name": True} @@ -358,12 +359,12 @@ class HTTPBase(SecurityBase): class HTTPBearer(HTTPBase): scheme: Literal["bearer"] = "bearer" - bearerFormat: Optional[str] = None + bearerFormat: str | None = None class OAuthFlow(BaseModel): - refreshUrl: Optional[str] = None - scopes: Dict[str, str] = {} + refreshUrl: str | None = None + scopes: dict[str, str] = {} model_config = MODEL_CONFIG_ALLOW @@ -386,10 +387,10 @@ class OAuthFlowAuthorizationCode(OAuthFlow): class OAuthFlows(BaseModel): - implicit: Optional[OAuthFlowImplicit] = None - password: Optional[OAuthFlowPassword] = None - clientCredentials: Optional[OAuthFlowClientCredentials] = None - authorizationCode: Optional[OAuthFlowAuthorizationCode] = None + implicit: OAuthFlowImplicit | None = None + password: OAuthFlowPassword | None = None + clientCredentials: OAuthFlowClientCredentials | None = None + authorizationCode: OAuthFlowAuthorizationCode | None = None model_config = MODEL_CONFIG_ALLOW @@ -407,22 +408,22 @@ class OpenIdConnect(SecurityBase): openIdConnectUrl: str -SecurityScheme = Union[APIKey, HTTPBase, OAuth2, OpenIdConnect, HTTPBearer] +SecurityScheme = APIKey | HTTPBase | OAuth2 | OpenIdConnect | HTTPBearer # https://swagger.io/specification/#components-object class Components(BaseModel): - schemas: Optional[Dict[str, Union[Schema, Reference]]] = None - responses: Optional[Dict[str, Union[Response, Reference]]] = None - parameters: Optional[Dict[str, Union[Parameter, Reference]]] = None - examples: Optional[Dict[str, Union[Example, Reference]]] = None - requestBodies: Optional[Dict[str, Union[RequestBody, Reference]]] = None - headers: Optional[Dict[str, Union[Header, Reference]]] = None - securitySchemes: Optional[Dict[str, Union[SecurityScheme, Reference]]] = None - links: Optional[Dict[str, Union[Link, Reference]]] = None + schemas: dict[str, Schema | Reference] | None = None + responses: dict[str, Response | Reference] | None = None + parameters: dict[str, Parameter | Reference] | None = None + examples: dict[str, Example | Reference] | None = None + requestBodies: dict[str, RequestBody | Reference] | None = None + headers: dict[str, Header | Reference] | None = None + securitySchemes: dict[str, SecurityScheme | Reference] | None = None + links: dict[str, Link | Reference] | None = None # Using Any for Specification Extensions - callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference, Any]]] = None - pathItems: Optional[Dict[str, Union[PathItem, Reference]]] = None + callbacks: dict[str, dict[str, PathItem] | Reference | Any] | None = None + pathItems: dict[str, PathItem | Reference] | None = None model_config = MODEL_CONFIG_ALLOW @@ -431,15 +432,15 @@ class Components(BaseModel): class OpenAPI(BaseModel): openapi: str info: Info - jsonSchemaDialect: Optional[str] = None - servers: Optional[List[Server]] = None + jsonSchemaDialect: str | None = None + servers: list[Server] | None = None # Using Any for Specification Extensions - paths: Optional[Dict[str, Union[PathItem, Any]]] = None - webhooks: Optional[Dict[str, Union[PathItem, Reference]]] = None - components: Optional[Components] = None - security: Optional[List[Dict[str, List[str]]]] = None - tags: Optional[List[Tag]] = None - externalDocs: Optional[ExternalDocumentation] = None + paths: dict[str, PathItem | Any] | None = None + webhooks: dict[str, PathItem | Reference] | None = None + components: Components | None = None + security: list[dict[str, list[str]]] | None = None + tags: list[Tag] | None = None + externalDocs: ExternalDocumentation | None = None model_config = MODEL_CONFIG_ALLOW diff --git a/aws_lambda_powertools/event_handler/openapi/params.py b/aws_lambda_powertools/event_handler/openapi/params.py index 4aa22882c3d..3374e228096 100644 --- a/aws_lambda_powertools/event_handler/openapi/params.py +++ b/aws_lambda_powertools/event_handler/openapi/params.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import inspect from enum import Enum -from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Type, Union +from typing import TYPE_CHECKING, Any, Callable, Literal from pydantic import BaseConfig from pydantic.fields import FieldInfo @@ -16,7 +18,9 @@ field_annotation_is_scalar, get_annotation_from_field_info, ) -from aws_lambda_powertools.event_handler.openapi.types import CacheKey + +if TYPE_CHECKING: + from aws_lambda_powertools.event_handler.openapi.types import CacheKey """ This turns the low-level function signature into typed, validated Pydantic models for consumption. @@ -42,21 +46,21 @@ class Dependant: def __init__( self, *, - path_params: Optional[List[ModelField]] = None, - query_params: Optional[List[ModelField]] = None, - header_params: Optional[List[ModelField]] = None, - cookie_params: Optional[List[ModelField]] = None, - body_params: Optional[List[ModelField]] = None, - return_param: Optional[ModelField] = None, - response_extra_models: Optional[List[ModelField]] = None, - name: Optional[str] = None, - call: Optional[Callable[..., Any]] = None, - request_param_name: Optional[str] = None, - websocket_param_name: Optional[str] = None, - http_connection_param_name: Optional[str] = None, - response_param_name: Optional[str] = None, - background_tasks_param_name: Optional[str] = None, - path: Optional[str] = None, + path_params: list[ModelField] | None = None, + query_params: list[ModelField] | None = None, + header_params: list[ModelField] | None = None, + cookie_params: list[ModelField] | None = None, + body_params: list[ModelField] | None = None, + return_param: ModelField | None = None, + response_extra_models: list[ModelField] | None = None, + name: str | None = None, + call: Callable[..., Any] | None = None, + request_param_name: str | None = None, + websocket_param_name: str | None = None, + http_connection_param_name: str | None = None, + response_param_name: str | None = None, + background_tasks_param_name: str | None = None, + path: str | None = None, ) -> None: self.path_params = path_params or [] self.query_params = query_params or [] @@ -89,33 +93,33 @@ def __init__( self, default: Any = Undefined, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # MAINTENANCE: validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): """ @@ -167,13 +171,13 @@ def __init__( Only applies to Decimals, requires the field to have a maxmium number of digits within the decimal. decimal_places: int, optional Only applies to Decimals, requires the field to have at most a number of decimal places - examples: List[Any], optional + examples: list[Any], optional A list of examples for the parameter deprecated: bool, optional If `True`, the parameter will be marked as deprecated include_in_schema: bool, optional If `False`, the parameter will be excluded from the generated OpenAPI schema - json_schema_extra: Dict[str, Any], optional + json_schema_extra: dict[str, Any], optional Extra values to include in the generated OpenAPI schema """ self.deprecated = deprecated @@ -234,33 +238,33 @@ def __init__( self, default: Any = ..., *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # MAINTENANCE: validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): """ @@ -312,13 +316,13 @@ def __init__( Only applies to Decimals, requires the field to have a maxmium number of digits within the decimal. decimal_places: int, optional Only applies to Decimals, requires the field to have at most a number of decimal places - examples: List[Any], optional + examples: list[Any], optional A list of examples for the parameter deprecated: bool, optional If `True`, the parameter will be marked as deprecated include_in_schema: bool, optional If `False`, the parameter will be excluded from the generated OpenAPI schema - json_schema_extra: Dict[str, Any], optional + json_schema_extra: dict[str, Any], optional Extra values to include in the generated OpenAPI schema """ if default is not ...: @@ -366,31 +370,31 @@ def __init__( self, default: Any = _Unset, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, + alias: str | None = None, + alias_priority: int | None = _Unset, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): """ @@ -442,13 +446,13 @@ def __init__( Only applies to Decimals, requires the field to have a maxmium number of digits within the decimal. decimal_places: int, optional Only applies to Decimals, requires the field to have at most a number of decimal places - examples: List[Any], optional + examples: list[Any], optional A list of examples for the parameter deprecated: bool, optional If `True`, the parameter will be marked as deprecated include_in_schema: bool, optional If `False`, the parameter will be excluded from the generated OpenAPI schema - json_schema_extra: Dict[str, Any], optional + json_schema_extra: dict[str, Any], optional Extra values to include in the generated OpenAPI schema """ super().__init__( @@ -493,34 +497,34 @@ def __init__( self, default: Any = Undefined, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, convert_underscores: bool = True, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): """ @@ -575,13 +579,13 @@ def __init__( Only applies to Decimals, requires the field to have a maxmium number of digits within the decimal. decimal_places: int, optional Only applies to Decimals, requires the field to have at most a number of decimal places - examples: List[Any], optional + examples: list[Any], optional A list of examples for the parameter deprecated: bool, optional If `True`, the parameter will be marked as deprecated include_in_schema: bool, optional If `False`, the parameter will be excluded from the generated OpenAPI schema - json_schema_extra: Dict[str, Any], optional + json_schema_extra: dict[str, Any], optional Extra values to include in the generated OpenAPI schema """ self.convert_underscores = convert_underscores @@ -622,7 +626,7 @@ def alias(self): return self._alias @alias.setter - def alias(self, value: Optional[str] = None): + def alias(self, value: str | None = None): if value is not None: # Headers are case-insensitive according to RFC 7540 (HTTP/2), so we lower the parameter name # This ensures that customers can access headers with any casing, as per the RFC guidelines. @@ -639,35 +643,35 @@ def __init__( self, default: Any = Undefined, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, embed: bool = False, media_type: str = "application/json", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): self.embed = embed @@ -726,34 +730,34 @@ def __init__( self, default: Any = Undefined, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, media_type: str = "application/x-www-form-urlencoded", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): super().__init__( @@ -798,34 +802,34 @@ def __init__( self, default: Any = Undefined, *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, + default_factory: Callable[[], Any] | None = _Unset, + annotation: Any | None = None, media_type: str = "multipart/form-data", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, + alias: str | None = None, + alias_priority: int | None = _Unset, # MAINTENANCE: update when deprecating Pydantic v1, import these types # str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - deprecated: Optional[bool] = None, + validation_alias: str | None = None, + serialization_alias: str | None = None, + title: str | None = None, + description: str | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | None = None, + discriminator: str | None = None, + strict: bool | None = _Unset, + multiple_of: float | None = _Unset, + allow_inf_nan: bool | None = _Unset, + max_digits: int | None = _Unset, + decimal_places: int | None = _Unset, + examples: list[Any] | None = None, + deprecated: bool | None = None, include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, + json_schema_extra: dict[str, Any] | None = None, **extra: Any, ): super().__init__( @@ -862,7 +866,7 @@ def __init__( def get_flat_dependant( dependant: Dependant, - visited: Optional[List[CacheKey]] = None, + visited: list[CacheKey] | None = None, ) -> Dependant: """ Flatten a recursive Dependant model structure. @@ -877,7 +881,7 @@ def get_flat_dependant( The dependant model to flatten skip_repeats: bool If True, child Dependents already visited will be skipped to avoid duplicates - visited: List[CacheKey], optional + visited: list[CacheKey], optional Keeps track of visited Dependents to avoid infinite recursion. Defaults to empty list. Returns @@ -906,7 +910,7 @@ def analyze_param( value: Any, is_path_param: bool, is_response_param: bool, -) -> Optional[ModelField]: +) -> ModelField | None: """ Analyze a parameter annotation and value to determine the type and default value of the parameter. @@ -925,7 +929,7 @@ def analyze_param( Returns ------- - Optional[ModelField] + ModelField | None The type annotation and the Pydantic field representing the parameter """ field_info, type_annotation = get_field_info_and_type_annotation(annotation, value, is_path_param) @@ -958,11 +962,11 @@ def analyze_param( return field -def get_field_info_and_type_annotation(annotation, value, is_path_param: bool) -> Tuple[Optional[FieldInfo], Any]: +def get_field_info_and_type_annotation(annotation, value, is_path_param: bool) -> tuple[FieldInfo | None, Any]: """ Get the FieldInfo and type annotation from an annotation and value. """ - field_info: Optional[FieldInfo] = None + field_info: FieldInfo | None = None type_annotation: Any = Any if annotation is not inspect.Signature.empty: @@ -979,7 +983,7 @@ def get_field_info_and_type_annotation(annotation, value, is_path_param: bool) - return field_info, type_annotation -def get_field_info_response_type(annotation, value) -> Tuple[Optional[FieldInfo], Any]: +def get_field_info_response_type(annotation, value) -> tuple[FieldInfo | None, Any]: # Example: get_args(Response[inner_type]) == (inner_type,) # noqa: ERA001 (inner_type,) = get_args(annotation) @@ -987,11 +991,11 @@ def get_field_info_response_type(annotation, value) -> Tuple[Optional[FieldInfo] return get_field_info_and_type_annotation(inner_type, value, False) -def get_field_info_annotated_type(annotation, value, is_path_param: bool) -> Tuple[Optional[FieldInfo], Any]: +def get_field_info_annotated_type(annotation, value, is_path_param: bool) -> tuple[FieldInfo | None, Any]: """ Get the FieldInfo and type annotation from an Annotated type. """ - field_info: Optional[FieldInfo] = None + field_info: FieldInfo | None = None annotated_args = get_args(annotation) type_annotation = annotated_args[0] powertools_annotations = [arg for arg in annotated_args[1:] if isinstance(arg, FieldInfo)] @@ -1022,12 +1026,12 @@ def get_field_info_annotated_type(annotation, value, is_path_param: bool) -> Tup def create_response_field( name: str, - type_: Type[Any], - default: Optional[Any] = Undefined, - required: Union[bool, UndefinedType] = Undefined, - model_config: Type[BaseConfig] = BaseConfig, - field_info: Optional[FieldInfo] = None, - alias: Optional[str] = None, + type_: type[Any], + default: Any | None = Undefined, + required: bool | UndefinedType = Undefined, + model_config: type[BaseConfig] = BaseConfig, + field_info: FieldInfo | None = None, + alias: str | None = None, mode: Literal["validation", "serialization"] = "validation", ) -> ModelField: """ @@ -1045,11 +1049,11 @@ def create_response_field( def _create_model_field( - field_info: Optional[FieldInfo], + field_info: FieldInfo | None, type_annotation: Any, param_name: str, is_path_param: bool, -) -> Optional[ModelField]: +) -> ModelField | None: """ Create a new ModelField from a FieldInfo and type annotation. """ diff --git a/aws_lambda_powertools/event_handler/openapi/swagger_ui/html.py b/aws_lambda_powertools/event_handler/openapi/swagger_ui/html.py index 85e041f2f56..953e55fd3ad 100644 --- a/aws_lambda_powertools/event_handler/openapi/swagger_ui/html.py +++ b/aws_lambda_powertools/event_handler/openapi/swagger_ui/html.py @@ -1,6 +1,9 @@ -from typing import Optional +from __future__ import annotations -from aws_lambda_powertools.event_handler.openapi.swagger_ui.oauth2 import OAuth2Config +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from aws_lambda_powertools.event_handler.openapi.swagger_ui.oauth2 import OAuth2Config def generate_swagger_html( @@ -9,7 +12,7 @@ def generate_swagger_html( swagger_js: str, swagger_css: str, swagger_base_url: str, - oauth2_config: Optional[OAuth2Config], + oauth2_config: OAuth2Config | None, persist_authorization: bool = False, ) -> str: """ diff --git a/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py b/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py index 53c046eb6a5..3fafe6b05e9 100644 --- a/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py +++ b/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py @@ -1,6 +1,8 @@ # ruff: noqa: E501 +from __future__ import annotations + import warnings -from typing import Dict, Optional, Sequence +from typing import Sequence from pydantic import BaseModel, Field, field_validator @@ -17,14 +19,14 @@ class OAuth2Config(BaseModel): """ # The client ID for the OAuth2 application - clientId: Optional[str] = Field(alias="client_id", default=None) + clientId: str | None = Field(alias="client_id", default=None) # The client secret for the OAuth2 application. This is sensitive information and requires the explicit presence # of the POWERTOOLS_DEV environment variable. - clientSecret: Optional[str] = Field(alias="client_secret", default=None) + clientSecret: str | None = Field(alias="client_secret", default=None) # The realm in which the OAuth2 application is registered. Optional. - realm: Optional[str] = Field(default=None) + realm: str | None = Field(default=None) # The name of the OAuth2 application appName: str = Field(alias="app_name") @@ -33,7 +35,7 @@ class OAuth2Config(BaseModel): scopes: Sequence[str] = Field(default=[]) # Additional query string parameters to be included in the OAuth2 request. Defaults to an empty dictionary. - additionalQueryStringParams: Dict[str, str] = Field(alias="additional_query_string_params", default={}) + additionalQueryStringParams: dict[str, str] = Field(alias="additional_query_string_params", default={}) # Whether to use basic authentication with the access code grant type. Defaults to False. useBasicAuthenticationWithAccessCodeGrant: bool = Field( @@ -47,7 +49,7 @@ class OAuth2Config(BaseModel): model_config = MODEL_CONFIG_ALLOW @field_validator("clientSecret") - def client_secret_only_on_dev(cls, v: Optional[str]) -> Optional[str]: + def client_secret_only_on_dev(cls, v: str | None) -> str | None: if not v: return None diff --git a/aws_lambda_powertools/event_handler/openapi/types.py b/aws_lambda_powertools/event_handler/openapi/types.py index 33a6ad4df4e..6f721e9ea38 100644 --- a/aws_lambda_powertools/event_handler/openapi/types.py +++ b/aws_lambda_powertools/event_handler/openapi/types.py @@ -1,16 +1,18 @@ +from __future__ import annotations + import types from enum import Enum -from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Set, Type, TypedDict, Union +from typing import TYPE_CHECKING, Any, Callable, TypedDict, Union from typing_extensions import NotRequired if TYPE_CHECKING: - from pydantic import BaseModel # noqa: F401 + from pydantic import BaseModel -CacheKey = Optional[Callable[..., Any]] -IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any]] -ModelNameMap = Dict[Union[Type["BaseModel"], Type[Enum]], str] -TypeModelOrEnum = Union[Type["BaseModel"], Type[Enum]] +CacheKey = Callable[..., Any] | None +IncEx = set[int] | set[str] | dict[int, Any] | dict[str, Any] +TypeModelOrEnum = type[BaseModel] | type[Enum] +ModelNameMap = dict[TypeModelOrEnum, str] UnionType = getattr(types, "UnionType", Union) @@ -48,7 +50,7 @@ class OpenAPIResponseContentSchema(TypedDict, total=False): - schema: Dict + schema: dict class OpenAPIResponseContentModel(TypedDict): @@ -57,4 +59,4 @@ class OpenAPIResponseContentModel(TypedDict): class OpenAPIResponse(TypedDict): description: str - content: NotRequired[Dict[str, Union[OpenAPIResponseContentSchema, OpenAPIResponseContentModel]]] + content: NotRequired[dict[str, OpenAPIResponseContentSchema | OpenAPIResponseContentModel]] From f7495534da2c9bcdd1be0f32ecd7ee02c11050f6 Mon Sep 17 00:00:00 2001 From: Eric Nielsen <4120606+ericbn@users.noreply.github.com> Date: Thu, 15 Aug 2024 13:13:18 -0500 Subject: [PATCH 2/4] Fix type alias with Python 3.8 See https://bugs.python.org/issue45117 --- .../event_handler/openapi/models.py | 6 +++--- .../event_handler/openapi/types.py | 14 ++++++-------- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/aws_lambda_powertools/event_handler/openapi/models.py b/aws_lambda_powertools/event_handler/openapi/models.py index 0be1d951573..faa09b51878 100644 --- a/aws_lambda_powertools/event_handler/openapi/models.py +++ b/aws_lambda_powertools/event_handler/openapi/models.py @@ -1,7 +1,7 @@ from __future__ import annotations from enum import Enum -from typing import Any, Literal +from typing import Any, Literal, Union from pydantic import AnyUrl, BaseModel, Field @@ -181,7 +181,7 @@ class Schema(BaseModel): # Ref: https://json-schema.org/draft/2020-12/json-schema-core.html#name-json-schema-documents # A JSON Schema MUST be an object or a boolean. -SchemaOrBool = Schema | bool +SchemaOrBool = Union[Schema, bool] # https://swagger.io/specification/#example-object @@ -408,7 +408,7 @@ class OpenIdConnect(SecurityBase): openIdConnectUrl: str -SecurityScheme = APIKey | HTTPBase | OAuth2 | OpenIdConnect | HTTPBearer +SecurityScheme = Union[APIKey, HTTPBase, OAuth2, OpenIdConnect, HTTPBearer] # https://swagger.io/specification/#components-object diff --git a/aws_lambda_powertools/event_handler/openapi/types.py b/aws_lambda_powertools/event_handler/openapi/types.py index 6f721e9ea38..96e99380230 100644 --- a/aws_lambda_powertools/event_handler/openapi/types.py +++ b/aws_lambda_powertools/event_handler/openapi/types.py @@ -2,17 +2,15 @@ import types from enum import Enum -from typing import TYPE_CHECKING, Any, Callable, TypedDict, Union +from typing import Any, Callable, Dict, Set, TypedDict, Union +from pydantic import BaseModel from typing_extensions import NotRequired -if TYPE_CHECKING: - from pydantic import BaseModel - -CacheKey = Callable[..., Any] | None -IncEx = set[int] | set[str] | dict[int, Any] | dict[str, Any] -TypeModelOrEnum = type[BaseModel] | type[Enum] -ModelNameMap = dict[TypeModelOrEnum, str] +CacheKey = Union[Callable[..., Any], None] +IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any]] +TypeModelOrEnum = Union[type[BaseModel], type[Enum]] +ModelNameMap = Dict[TypeModelOrEnum, str] UnionType = getattr(types, "UnionType", Union) From e56c2cd5547653d9f0acbd26fe0a815136f81b52 Mon Sep 17 00:00:00 2001 From: Eric Nielsen <4120606+ericbn@users.noreply.github.com> Date: Thu, 15 Aug 2024 16:28:04 -0500 Subject: [PATCH 3/4] Fix pydantic not working with Python 3.8 TypeError: You have a type annotation 'str | None' which makes use of newer typing features than are supported in your version of Python. To handle this error, you should either remove the use of new syntax or install the `eval_type_backport` package. --- .../event_handler/openapi/models.py | 344 +++++++++--------- .../openapi/swagger_ui/oauth2.py | 16 +- .../event_handler/openapi/types.py | 4 +- 3 files changed, 181 insertions(+), 183 deletions(-) diff --git a/aws_lambda_powertools/event_handler/openapi/models.py b/aws_lambda_powertools/event_handler/openapi/models.py index faa09b51878..69d0e96cc9f 100644 --- a/aws_lambda_powertools/event_handler/openapi/models.py +++ b/aws_lambda_powertools/event_handler/openapi/models.py @@ -1,9 +1,9 @@ -from __future__ import annotations - +# ruff: noqa: FA100 from enum import Enum -from typing import Any, Literal, Union +from typing import Any, Dict, List, Literal, Optional, Set, Union from pydantic import AnyUrl, BaseModel, Field +from typing_extensions import Annotated from aws_lambda_powertools.event_handler.openapi.compat import model_rebuild from aws_lambda_powertools.event_handler.openapi.constants import ( @@ -19,9 +19,9 @@ # https://swagger.io/specification/#contact-object class Contact(BaseModel): - name: str | None = None - url: AnyUrl | None = None - email: str | None = None + name: Optional[str] = None + url: Optional[AnyUrl] = None + email: Optional[str] = None model_config = MODEL_CONFIG_ALLOW @@ -29,8 +29,8 @@ class Contact(BaseModel): # https://swagger.io/specification/#license-object class License(BaseModel): name: str - identifier: str | None = None - url: AnyUrl | None = None + identifier: Optional[str] = None + url: Optional[AnyUrl] = None model_config = MODEL_CONFIG_ALLOW @@ -38,30 +38,30 @@ class License(BaseModel): # https://swagger.io/specification/#info-object class Info(BaseModel): title: str - description: str | None = None - termsOfService: str | None = None - contact: Contact | None = None - license: License | None = None # noqa: A003 + description: Optional[str] = None + termsOfService: Optional[str] = None + contact: Optional[Contact] = None + license: Optional[License] = None # noqa: A003 version: str - summary: str | None = None + summary: Optional[str] = None model_config = MODEL_CONFIG_IGNORE # https://swagger.io/specification/#server-variable-object class ServerVariable(BaseModel): - enum: list[str] | None = Field(default=None, min_length=1) + enum: Annotated[Optional[List[str]], Field(min_length=1)] = None default: str - description: str | None = None + description: Optional[str] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#server-object class Server(BaseModel): - url: AnyUrl | str - description: str | None = None - variables: dict[str, ServerVariable] | None = None + url: Union[AnyUrl, str] + description: Optional[str] = None + variables: Optional[Dict[str, ServerVariable]] = None model_config = MODEL_CONFIG_ALLOW @@ -74,23 +74,23 @@ class Reference(BaseModel): # https://swagger.io/specification/#discriminator-object class Discriminator(BaseModel): propertyName: str - mapping: dict[str, str] | None = None + mapping: Optional[Dict[str, str]] = None # https://swagger.io/specification/#xml-object class XML(BaseModel): - name: str | None = None - namespace: str | None = None - prefix: str | None = None - attribute: bool | None = None - wrapped: bool | None = None + name: Optional[str] = None + namespace: Optional[str] = None + prefix: Optional[str] = None + attribute: Optional[bool] = None + wrapped: Optional[bool] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#external-documentation-object class ExternalDocumentation(BaseModel): - description: str | None = None + description: Optional[str] = None url: AnyUrl model_config = MODEL_CONFIG_ALLOW @@ -100,81 +100,81 @@ class ExternalDocumentation(BaseModel): class Schema(BaseModel): # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-the-json-schema-core-vocabu # Core Vocabulary - schema_: str | None = Field(default=None, alias="$schema") - vocabulary: str | None = Field(default=None, alias="$vocabulary") - id: str | None = Field(default=None, alias="$id") # noqa: A003 - anchor: str | None = Field(default=None, alias="$anchor") - dynamicAnchor: str | None = Field(default=None, alias="$dynamicAnchor") - ref: str | None = Field(default=None, alias="$ref") - dynamicRef: str | None = Field(default=None, alias="$dynamicRef") - defs: dict[str, SchemaOrBool] | None = Field(default=None, alias="$defs") - comment: str | None = Field(default=None, alias="$comment") + schema_: Optional[str] = Field(default=None, alias="$schema") + vocabulary: Optional[str] = Field(default=None, alias="$vocabulary") + id: Optional[str] = Field(default=None, alias="$id") # noqa: A003 + anchor: Optional[str] = Field(default=None, alias="$anchor") + dynamicAnchor: Optional[str] = Field(default=None, alias="$dynamicAnchor") + ref: Optional[str] = Field(default=None, alias="$ref") + dynamicRef: Optional[str] = Field(default=None, alias="$dynamicRef") + defs: Optional[Dict[str, "SchemaOrBool"]] = Field(default=None, alias="$defs") + comment: Optional[str] = Field(default=None, alias="$comment") # Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-a-vocabulary-for-applying-s # A Vocabulary for Applying Subschemas - allOf: list[SchemaOrBool] | None = None - anyOf: list[SchemaOrBool] | None = None - oneOf: list[SchemaOrBool] | None = None - not_: SchemaOrBool | None = Field(default=None, alias="not") - if_: SchemaOrBool | None = Field(default=None, alias="if") - then: SchemaOrBool | None = None - else_: SchemaOrBool | None = Field(default=None, alias="else") - dependentSchemas: dict[str, SchemaOrBool] | None = None - prefixItems: list[SchemaOrBool] | None = None + allOf: Optional[List["SchemaOrBool"]] = None + anyOf: Optional[List["SchemaOrBool"]] = None + oneOf: Optional[List["SchemaOrBool"]] = None + not_: Optional["SchemaOrBool"] = Field(default=None, alias="not") + if_: Optional["SchemaOrBool"] = Field(default=None, alias="if") + then: Optional["SchemaOrBool"] = None + else_: Optional["SchemaOrBool"] = Field(default=None, alias="else") + dependentSchemas: Optional[Dict[str, "SchemaOrBool"]] = None + prefixItems: Optional[List["SchemaOrBool"]] = None # MAINTENANCE: uncomment and remove below when deprecating Pydantic v1 # MAINTENANCE: It generates a list of schemas for tuples, before prefixItems was available - # MAINTENANCE: items: SchemaOrBool | None = None - items: SchemaOrBool | list[SchemaOrBool] | None = None - contains: SchemaOrBool | None = None - properties: dict[str, SchemaOrBool] | None = None - patternProperties: dict[str, SchemaOrBool] | None = None - additionalProperties: SchemaOrBool | None = None - propertyNames: SchemaOrBool | None = None - unevaluatedItems: SchemaOrBool | None = None - unevaluatedProperties: SchemaOrBool | None = None + # MAINTENANCE: items: Optional["SchemaOrBool"] = None + items: Optional[Union["SchemaOrBool", List["SchemaOrBool"]]] = None + contains: Optional["SchemaOrBool"] = None + properties: Optional[Dict[str, "SchemaOrBool"]] = None + patternProperties: Optional[Dict[str, "SchemaOrBool"]] = None + additionalProperties: Optional["SchemaOrBool"] = None + propertyNames: Optional["SchemaOrBool"] = None + unevaluatedItems: Optional["SchemaOrBool"] = None + unevaluatedProperties: Optional["SchemaOrBool"] = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-structural # A Vocabulary for Structural Validation - type: str | None = None # noqa: A003 - enum: list[Any] | None = None - const: Any | None = None - multipleOf: float | None = Field(default=None, gt=0) - maximum: float | None = None - exclusiveMaximum: float | None = None - minimum: float | None = None - exclusiveMinimum: float | None = None - maxLength: int | None = Field(default=None, ge=0) - minLength: int | None = Field(default=None, ge=0) - pattern: str | None = None - maxItems: int | None = Field(default=None, ge=0) - minItems: int | None = Field(default=None, ge=0) - uniqueItems: bool | None = None - maxContains: int | None = Field(default=None, ge=0) - minContains: int | None = Field(default=None, ge=0) - maxProperties: int | None = Field(default=None, ge=0) - minProperties: int | None = Field(default=None, ge=0) - required: list[str] | None = None - dependentRequired: dict[str, set[str]] | None = None + type: Optional[str] = None # noqa: A003 + enum: Optional[List[Any]] = None + const: Optional[Any] = None + multipleOf: Optional[float] = Field(default=None, gt=0) + maximum: Optional[float] = None + exclusiveMaximum: Optional[float] = None + minimum: Optional[float] = None + exclusiveMinimum: Optional[float] = None + maxLength: Optional[int] = Field(default=None, ge=0) + minLength: Optional[int] = Field(default=None, ge=0) + pattern: Optional[str] = None + maxItems: Optional[int] = Field(default=None, ge=0) + minItems: Optional[int] = Field(default=None, ge=0) + uniqueItems: Optional[bool] = None + maxContains: Optional[int] = Field(default=None, ge=0) + minContains: Optional[int] = Field(default=None, ge=0) + maxProperties: Optional[int] = Field(default=None, ge=0) + minProperties: Optional[int] = Field(default=None, ge=0) + required: Optional[List[str]] = None + dependentRequired: Optional[Dict[str, Set[str]]] = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-vocabularies-for-semantic-c # Vocabularies for Semantic Content With "format" - format: str | None = None # noqa: A003 + format: Optional[str] = None # noqa: A003 # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-the-conten # A Vocabulary for the Contents of String-Encoded Data - contentEncoding: str | None = None - contentMediaType: str | None = None - contentSchema: SchemaOrBool | None = None + contentEncoding: Optional[str] = None + contentMediaType: Optional[str] = None + contentSchema: Optional["SchemaOrBool"] = None # Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-basic-meta # A Vocabulary for Basic Meta-Data Annotations - title: str | None = None - description: str | None = None - default: Any | None = None - deprecated: bool | None = None - readOnly: bool | None = None - writeOnly: bool | None = None - examples: list[Example] | None = None + title: Optional[str] = None + description: Optional[str] = None + default: Optional[Any] = None + deprecated: Optional[bool] = None + readOnly: Optional[bool] = None + writeOnly: Optional[bool] = None + examples: Optional[List["Example"]] = None # Ref: OpenAPI 3.0.0: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.0.0.md#schema-object # Schema Object - discriminator: Discriminator | None = None - xml: XML | None = None - externalDocs: ExternalDocumentation | None = None + discriminator: Optional[Discriminator] = None + xml: Optional[XML] = None + externalDocs: Optional[ExternalDocumentation] = None model_config = MODEL_CONFIG_ALLOW @@ -186,10 +186,10 @@ class Schema(BaseModel): # https://swagger.io/specification/#example-object class Example(BaseModel): - summary: str | None = None - description: str | None = None - value: Any | None = None - externalValue: AnyUrl | None = None + summary: Optional[str] = None + description: Optional[str] = None + value: Optional[Any] = None + externalValue: Optional[AnyUrl] = None model_config = MODEL_CONFIG_ALLOW @@ -203,37 +203,37 @@ class ParameterInType(Enum): # https://swagger.io/specification/#encoding-object class Encoding(BaseModel): - contentType: str | None = None - headers: dict[str, Header | Reference] | None = None - style: str | None = None - explode: bool | None = None - allowReserved: bool | None = None + contentType: Optional[str] = None + headers: Optional[Dict[str, Union["Header", Reference]]] = None + style: Optional[str] = None + explode: Optional[bool] = None + allowReserved: Optional[bool] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#media-type-object class MediaType(BaseModel): - schema_: Schema | Reference | None = Field(default=None, alias="schema") - examples: dict[str, Example | Reference] | None = None - encoding: dict[str, Encoding] | None = None + schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema") + examples: Optional[Dict[str, Union[Example, Reference]]] = None + encoding: Optional[Dict[str, Encoding]] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#parameter-object class ParameterBase(BaseModel): - description: str | None = None - required: bool | None = None - deprecated: bool | None = None + description: Optional[str] = None + required: Optional[bool] = None + deprecated: Optional[bool] = None # Serialization rules for simple scenarios - style: str | None = None - explode: bool | None = None - allowReserved: bool | None = None - schema_: Schema | Reference | None = Field(default=None, alias="schema") - examples: dict[str, Example | Reference] | None = None + style: Optional[str] = None + explode: Optional[bool] = None + allowReserved: Optional[bool] = None + schema_: Optional[Union[Schema, Reference]] = Field(default=None, alias="schema") + examples: Optional[Dict[str, Union[Example, Reference]]] = None # Serialization rules for more complex scenarios - content: dict[str, MediaType] | None = None + content: Optional[Dict[str, MediaType]] = None model_config = MODEL_CONFIG_ALLOW @@ -249,21 +249,21 @@ class Header(ParameterBase): # https://swagger.io/specification/#request-body-object class RequestBody(BaseModel): - description: str | None = None - content: dict[str, MediaType] - required: bool | None = None + description: Optional[str] = None + content: Dict[str, MediaType] + required: Optional[bool] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#link-object class Link(BaseModel): - operationRef: str | None = None - operationId: str | None = None - parameters: dict[str, Any | str] | None = None - requestBody: Any | str | None = None - description: str | None = None - server: Server | None = None + operationRef: Optional[str] = None + operationId: Optional[str] = None + parameters: Optional[Dict[str, Union[Any, str]]] = None + requestBody: Optional[Union[Any, str]] = None + description: Optional[str] = None + server: Optional[Server] = None model_config = MODEL_CONFIG_ALLOW @@ -271,9 +271,9 @@ class Link(BaseModel): # https://swagger.io/specification/#response-object class Response(BaseModel): description: str - headers: dict[str, Header | Reference] | None = None - content: dict[str, MediaType] | None = None - links: dict[str, Link | Reference] | None = None + headers: Optional[Dict[str, Union[Header, Reference]]] = None + content: Optional[Dict[str, MediaType]] = None + links: Optional[Dict[str, Union[Link, Reference]]] = None model_config = MODEL_CONFIG_ALLOW @@ -281,46 +281,46 @@ class Response(BaseModel): # https://swagger.io/specification/#tag-object class Tag(BaseModel): name: str - description: str | None = None - externalDocs: ExternalDocumentation | None = None + description: Optional[str] = None + externalDocs: Optional[ExternalDocumentation] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#operation-object class Operation(BaseModel): - tags: list[str] | None = None - summary: str | None = None - description: str | None = None - externalDocs: ExternalDocumentation | None = None - operationId: str | None = None - parameters: list[Parameter | Reference] | None = None - requestBody: RequestBody | Reference | None = None + tags: Optional[List[str]] = None + summary: Optional[str] = None + description: Optional[str] = None + externalDocs: Optional[ExternalDocumentation] = None + operationId: Optional[str] = None + parameters: Optional[List[Union[Parameter, Reference]]] = None + requestBody: Optional[Union[RequestBody, Reference]] = None # Using Any for Specification Extensions - responses: dict[int, Response | Any] | None = None - callbacks: dict[str, dict[str, PathItem] | Reference] | None = None - deprecated: bool | None = None - security: list[dict[str, list[str]]] | None = None - servers: list[Server] | None = None + responses: Optional[Dict[int, Union[Response, Any]]] = None + callbacks: Optional[Dict[str, Union[Dict[str, "PathItem"], Reference]]] = None + deprecated: Optional[bool] = None + security: Optional[List[Dict[str, List[str]]]] = None + servers: Optional[List[Server]] = None model_config = MODEL_CONFIG_ALLOW # https://swagger.io/specification/#path-item-object class PathItem(BaseModel): - ref: str | None = Field(default=None, alias="$ref") - summary: str | None = None - description: str | None = None - get: Operation | None = None - put: Operation | None = None - post: Operation | None = None - delete: Operation | None = None - options: Operation | None = None - head: Operation | None = None - patch: Operation | None = None - trace: Operation | None = None - servers: list[Server] | None = None - parameters: list[Parameter | Reference] | None = None + ref: Optional[str] = Field(default=None, alias="$ref") + 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 model_config = MODEL_CONFIG_ALLOW @@ -335,7 +335,7 @@ class SecuritySchemeType(Enum): class SecurityBase(BaseModel): type_: SecuritySchemeType = Field(alias="type") - description: str | None = None + description: Optional[str] = None model_config = {"extra": "allow", "populate_by_name": True} @@ -359,12 +359,12 @@ class HTTPBase(SecurityBase): class HTTPBearer(HTTPBase): scheme: Literal["bearer"] = "bearer" - bearerFormat: str | None = None + bearerFormat: Optional[str] = None class OAuthFlow(BaseModel): - refreshUrl: str | None = None - scopes: dict[str, str] = {} + refreshUrl: Optional[str] = None + scopes: Dict[str, str] = {} model_config = MODEL_CONFIG_ALLOW @@ -387,10 +387,10 @@ class OAuthFlowAuthorizationCode(OAuthFlow): class OAuthFlows(BaseModel): - implicit: OAuthFlowImplicit | None = None - password: OAuthFlowPassword | None = None - clientCredentials: OAuthFlowClientCredentials | None = None - authorizationCode: OAuthFlowAuthorizationCode | None = None + implicit: Optional[OAuthFlowImplicit] = None + password: Optional[OAuthFlowPassword] = None + clientCredentials: Optional[OAuthFlowClientCredentials] = None + authorizationCode: Optional[OAuthFlowAuthorizationCode] = None model_config = MODEL_CONFIG_ALLOW @@ -413,17 +413,17 @@ class OpenIdConnect(SecurityBase): # https://swagger.io/specification/#components-object class Components(BaseModel): - schemas: dict[str, Schema | Reference] | None = None - responses: dict[str, Response | Reference] | None = None - parameters: dict[str, Parameter | Reference] | None = None - examples: dict[str, Example | Reference] | None = None - requestBodies: dict[str, RequestBody | Reference] | None = None - headers: dict[str, Header | Reference] | None = None - securitySchemes: dict[str, SecurityScheme | Reference] | None = None - links: dict[str, Link | Reference] | None = None + schemas: Optional[Dict[str, Union[Schema, Reference]]] = None + responses: Optional[Dict[str, Union[Response, Reference]]] = None + parameters: Optional[Dict[str, Union[Parameter, Reference]]] = None + examples: Optional[Dict[str, Union[Example, Reference]]] = None + requestBodies: Optional[Dict[str, Union[RequestBody, Reference]]] = None + headers: Optional[Dict[str, Union[Header, Reference]]] = None + securitySchemes: Optional[Dict[str, Union[SecurityScheme, Reference]]] = None + links: Optional[Dict[str, Union[Link, Reference]]] = None # Using Any for Specification Extensions - callbacks: dict[str, dict[str, PathItem] | Reference | Any] | None = None - pathItems: dict[str, PathItem | Reference] | None = None + callbacks: Optional[Dict[str, Union[Dict[str, PathItem], Reference, Any]]] = None + pathItems: Optional[Dict[str, Union[PathItem, Reference]]] = None model_config = MODEL_CONFIG_ALLOW @@ -432,15 +432,15 @@ class Components(BaseModel): class OpenAPI(BaseModel): openapi: str info: Info - jsonSchemaDialect: str | None = None - servers: list[Server] | None = None + jsonSchemaDialect: Optional[str] = None + servers: Optional[List[Server]] = None # Using Any for Specification Extensions - paths: dict[str, PathItem | Any] | None = None - webhooks: dict[str, PathItem | Reference] | None = None - components: Components | None = None - security: list[dict[str, list[str]]] | None = None - tags: list[Tag] | None = None - externalDocs: ExternalDocumentation | None = None + paths: Optional[Dict[str, Union[PathItem, Any]]] = None + webhooks: Optional[Dict[str, Union[PathItem, Reference]]] = None + components: Optional[Components] = None + security: Optional[List[Dict[str, List[str]]]] = None + tags: Optional[List[Tag]] = None + externalDocs: Optional[ExternalDocumentation] = None model_config = MODEL_CONFIG_ALLOW diff --git a/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py b/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py index 3fafe6b05e9..490029d1f73 100644 --- a/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py +++ b/aws_lambda_powertools/event_handler/openapi/swagger_ui/oauth2.py @@ -1,8 +1,6 @@ -# ruff: noqa: E501 -from __future__ import annotations - +# ruff: noqa: E501 FA100 import warnings -from typing import Sequence +from typing import Dict, Optional, Sequence from pydantic import BaseModel, Field, field_validator @@ -19,14 +17,14 @@ class OAuth2Config(BaseModel): """ # The client ID for the OAuth2 application - clientId: str | None = Field(alias="client_id", default=None) + clientId: Optional[str] = Field(alias="client_id", default=None) # The client secret for the OAuth2 application. This is sensitive information and requires the explicit presence # of the POWERTOOLS_DEV environment variable. - clientSecret: str | None = Field(alias="client_secret", default=None) + clientSecret: Optional[str] = Field(alias="client_secret", default=None) # The realm in which the OAuth2 application is registered. Optional. - realm: str | None = Field(default=None) + realm: Optional[str] = Field(default=None) # The name of the OAuth2 application appName: str = Field(alias="app_name") @@ -35,7 +33,7 @@ class OAuth2Config(BaseModel): scopes: Sequence[str] = Field(default=[]) # Additional query string parameters to be included in the OAuth2 request. Defaults to an empty dictionary. - additionalQueryStringParams: dict[str, str] = Field(alias="additional_query_string_params", default={}) + additionalQueryStringParams: Dict[str, str] = Field(alias="additional_query_string_params", default={}) # Whether to use basic authentication with the access code grant type. Defaults to False. useBasicAuthenticationWithAccessCodeGrant: bool = Field( @@ -49,7 +47,7 @@ class OAuth2Config(BaseModel): model_config = MODEL_CONFIG_ALLOW @field_validator("clientSecret") - def client_secret_only_on_dev(cls, v: str | None) -> str | None: + def client_secret_only_on_dev(cls, v: Optional[str]) -> Optional[str]: if not v: return None diff --git a/aws_lambda_powertools/event_handler/openapi/types.py b/aws_lambda_powertools/event_handler/openapi/types.py index 96e99380230..ef6b096aad0 100644 --- a/aws_lambda_powertools/event_handler/openapi/types.py +++ b/aws_lambda_powertools/event_handler/openapi/types.py @@ -2,14 +2,14 @@ import types from enum import Enum -from typing import Any, Callable, Dict, Set, TypedDict, Union +from typing import Any, Callable, Dict, Set, Type, TypedDict, Union from pydantic import BaseModel from typing_extensions import NotRequired CacheKey = Union[Callable[..., Any], None] IncEx = Union[Set[int], Set[str], Dict[int, Any], Dict[str, Any]] -TypeModelOrEnum = Union[type[BaseModel], type[Enum]] +TypeModelOrEnum = Union[Type[BaseModel], Type[Enum]] ModelNameMap = Dict[TypeModelOrEnum, str] UnionType = getattr(types, "UnionType", Union) From f321e09be4c89109d361f9cafe106dae081bbcd4 Mon Sep 17 00:00:00 2001 From: Leandro Damascena Date: Thu, 15 Aug 2024 23:24:42 +0100 Subject: [PATCH 4/4] Removing pydantic v1 reference --- tests/e2e/utils/data_fetcher/common.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/e2e/utils/data_fetcher/common.py b/tests/e2e/utils/data_fetcher/common.py index 15354f8f948..1300daa55aa 100644 --- a/tests/e2e/utils/data_fetcher/common.py +++ b/tests/e2e/utils/data_fetcher/common.py @@ -8,7 +8,7 @@ import requests from mypy_boto3_lambda.client import LambdaClient from mypy_boto3_lambda.type_defs import InvocationResponseTypeDef -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict from requests import Request, Response from requests.exceptions import RequestException from retry import retry @@ -22,9 +22,9 @@ class GetLambdaResponseOptions(BaseModel): client: Optional[LambdaClient] = None raise_on_error: bool = True - # Maintenance: Pydantic v2 deprecated it; we should update in v3 - class Config: - arbitrary_types_allowed = True + model_config = ConfigDict( + arbitrary_types_allowed=True, + ) def get_lambda_response( @@ -106,7 +106,7 @@ def get_lambda_response_in_parallel( # we can assert on the correct output. time.sleep(0.5 * len(running_tasks)) - get_lambda_response_callback = functools.partial(get_lambda_response, **options.dict()) + get_lambda_response_callback = functools.partial(get_lambda_response, **options.model_dump()) running_tasks.append( executor.submit(get_lambda_response_callback), )