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..69d0e96cc9f 100644 --- a/aws_lambda_powertools/event_handler/openapi/models.py +++ b/aws_lambda_powertools/event_handler/openapi/models.py @@ -1,3 +1,4 @@ +# ruff: noqa: FA100 from enum import Enum from typing import Any, Dict, List, Literal, Optional, Set, Union 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..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,4 +1,4 @@ -# ruff: noqa: E501 +# ruff: noqa: E501 FA100 import warnings from typing import Dict, Optional, Sequence diff --git a/aws_lambda_powertools/event_handler/openapi/types.py b/aws_lambda_powertools/event_handler/openapi/types.py index 33a6ad4df4e..ef6b096aad0 100644 --- a/aws_lambda_powertools/event_handler/openapi/types.py +++ b/aws_lambda_powertools/event_handler/openapi/types.py @@ -1,16 +1,16 @@ +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 Any, Callable, Dict, Set, Type, TypedDict, Union +from pydantic import BaseModel from typing_extensions import NotRequired -if TYPE_CHECKING: - from pydantic import BaseModel # noqa: F401 - -CacheKey = Optional[Callable[..., Any]] +CacheKey = Union[Callable[..., Any], None] 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]] +TypeModelOrEnum = Union[Type[BaseModel], Type[Enum]] +ModelNameMap = Dict[TypeModelOrEnum, str] UnionType = getattr(types, "UnionType", Union) @@ -48,7 +48,7 @@ class OpenAPIResponseContentSchema(TypedDict, total=False): - schema: Dict + schema: dict class OpenAPIResponseContentModel(TypedDict): @@ -57,4 +57,4 @@ class OpenAPIResponseContentModel(TypedDict): class OpenAPIResponse(TypedDict): description: str - content: NotRequired[Dict[str, Union[OpenAPIResponseContentSchema, OpenAPIResponseContentModel]]] + content: NotRequired[dict[str, OpenAPIResponseContentSchema | OpenAPIResponseContentModel]] 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), )