feat(event_handler): mark API operation as deprecated for OpenAPI documentation

aws_lambda_powertools/event_handler/api_gateway.py

@@ -8,6 +8,7 @@
 import warnings
 import zlib
 from abc import ABC, abstractmethod
+from collections import defaultdict
 from enum import Enum
 from functools import partial
 from http import HTTPStatus
@@ -309,6 +310,7 @@ def __init__(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Response]] | None = None,
     ):
         """
@@ -348,6 +350,8 @@ def __init__(
             The OpenAPI security for this route
         openapi_extensions: dict[str, Any], optional
             Additional OpenAPI extensions as a dictionary.
+        deprecated: bool
+            Whether or not to mark this route as deprecated in the OpenAPI schema
         middlewares: list[Callable[..., Response]] | None
             The list of route middlewares to be called in order.
         """
@@ -374,6 +378,7 @@ def __init__(
         self.openapi_extensions = openapi_extensions
         self.middlewares = middlewares or []
         self.operation_id = operation_id or self._generate_operation_id()
+        self.deprecated = deprecated
 
         # _middleware_stack_built is used to ensure the middleware stack is only built once.
         self._middleware_stack_built = False
@@ -670,6 +675,10 @@ def _openapi_operation_metadata(self, operation_ids: set[str]) -> dict[str, Any]
         operation_ids.add(self.operation_id)
         operation["operationId"] = self.operation_id
 
+        # Mark as deprecated if necessary
+        if self.deprecated:
+            operation["deprecated"] = True
+
         return operation
 
     @staticmethod
@@ -924,6 +933,7 @@ def route(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         raise NotImplementedError()
@@ -984,6 +994,7 @@ def get(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Get route decorator with GET `method`
@@ -1023,6 +1034,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1041,6 +1053,7 @@ def post(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Post route decorator with POST `method`
@@ -1081,6 +1094,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1099,6 +1113,7 @@ def put(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Put route decorator with PUT `method`
@@ -1139,6 +1154,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1157,6 +1173,7 @@ def delete(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Delete route decorator with DELETE `method`
@@ -1196,6 +1213,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1214,6 +1232,7 @@ def patch(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Patch route decorator with PATCH `method`
@@ -1256,6 +1275,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1274,6 +1294,7 @@ def head(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Head route decorator with HEAD `method`
@@ -1315,6 +1336,7 @@ def lambda_handler(event, context):
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -1629,7 +1651,6 @@ def get_openapi_schema(
 
         # Add routes to the OpenAPI schema
         for route in all_routes:
-
             if route.security and not _validate_openapi_security_parameters(
                 security=route.security,
                 security_schemes=security_schemes,
@@ -1694,7 +1715,6 @@ def _get_openapi_security(
 
     @staticmethod
     def _determine_openapi_version(openapi_version: str):
-
         # Pydantic V2 has no support for OpenAPI schema 3.0
         if not openapi_version.startswith("3.1"):
             warnings.warn(
@@ -1950,6 +1970,7 @@ def route(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         """Route decorator includes parameter `method`"""
@@ -1978,6 +1999,7 @@ def register_resolver(func: AnyCallableT) -> AnyCallableT:
                     include_in_schema,
                     security,
                     openapi_extensions,
+                    deprecated,
                     middlewares,
                 )
 
@@ -2471,7 +2493,7 @@ class Router(BaseRouter):
 
     def __init__(self):
         self._routes: dict[tuple, Callable] = {}
-        self._routes_with_middleware: dict[tuple, list[Callable]] = {}
+        self._routes_with_middleware: defaultdict[tuple, list[Callable]] = defaultdict(list)
         self.api_resolver: BaseRouter | None = None
         self.context = {}  # early init as customers might add context before event resolution
         self._exception_handlers: dict[type, Callable] = {}
@@ -2492,6 +2514,7 @@ def route(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         def register_route(func: AnyCallableT) -> AnyCallableT:
@@ -2517,17 +2540,13 @@ def register_route(func: AnyCallableT) -> AnyCallableT:
                 include_in_schema,
                 frozen_security,
                 fronzen_openapi_extensions,
+                deprecated,
             )
 
             # Collate Middleware for routes
             if middlewares is not None:
                 for handler in middlewares:
-                    if self._routes_with_middleware.get(route_key) is None:
-                        self._routes_with_middleware[route_key] = [handler]
-                    else:
-                        self._routes_with_middleware[route_key].append(handler)
-            else:
-                self._routes_with_middleware[route_key] = []
+                    self._routes_with_middleware[route_key].append(handler)
 
             self._routes[route_key] = func
 
@@ -2598,6 +2617,7 @@ def route(
         include_in_schema: bool = True,
         security: list[dict[str, list[str]]] | None = None,
         openapi_extensions: dict[str, Any] | None = None,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[AnyCallableT], AnyCallableT]:
         # NOTE: see #1552 for more context.
@@ -2616,6 +2636,7 @@ def route(
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -108,9 +108,9 @@ def get(  # type: ignore[override]
         tags: list[str] | None = None,
         operation_id: str | None = None,
         include_in_schema: bool = True,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
-
         openapi_extensions = None
         security = None
 
@@ -128,6 +128,7 @@ def get(  # type: ignore[override]
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -146,6 +147,7 @@ def post(  # type: ignore[override]
         tags: list[str] | None = None,
         operation_id: str | None = None,
         include_in_schema: bool = True,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ):
         openapi_extensions = None
@@ -165,6 +167,7 @@ def post(  # type: ignore[override]
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -183,6 +186,7 @@ def put(  # type: ignore[override]
         tags: list[str] | None = None,
         operation_id: str | None = None,
         include_in_schema: bool = True,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ):
         openapi_extensions = None
@@ -202,6 +206,7 @@ def put(  # type: ignore[override]
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -220,6 +225,7 @@ def patch(  # type: ignore[override]
         tags: list[str] | None = None,
         operation_id: str | None = None,
         include_in_schema: bool = True,
+        deprecated: bool = False,
         middlewares: list[Callable] | None = None,
     ):
         openapi_extensions = None
@@ -239,6 +245,7 @@ def patch(  # type: ignore[override]
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 
@@ -257,6 +264,7 @@ def delete(  # type: ignore[override]
         tags: list[str] | None = None,
         operation_id: str | None = None,
         include_in_schema: bool = True,
+        deprecated: bool = False,
         middlewares: list[Callable[..., Any]] | None = None,
     ):
         openapi_extensions = None
@@ -276,6 +284,7 @@ def delete(  # type: ignore[override]
             include_in_schema,
             security,
             openapi_extensions,
+            deprecated,
             middlewares,
         )
 

aws_lambda_powertools/metrics/provider/cloudwatch_emf/cloudwatch.py

@@ -24,7 +24,6 @@
 from aws_lambda_powertools.shared.functions import resolve_env_var_choice
 from aws_lambda_powertools.warnings import PowertoolsUserWarning
 
-
 if TYPE_CHECKING:
     from aws_lambda_powertools.metrics.provider.cloudwatch_emf.types import CloudWatchEMFOutput
     from aws_lambda_powertools.metrics.types import MetricNameUnitResolution
@@ -295,8 +294,6 @@ def add_dimension(self, name: str, value: str) -> None:
 
         self.dimension_set[name] = value
 
-
-
     def add_metadata(self, key: str, value: Any) -> None:
         """Adds high cardinal metadata for metrics object
 

aws_lambda_powertools/utilities/parser/models/apigw_websocket.py

@@ -9,6 +9,7 @@ class APIGatewayWebSocketEventIdentity(BaseModel):
     source_ip: IPvAnyNetwork = Field(alias="sourceIp")
     user_agent: Optional[str] = Field(None, alias="userAgent")
 
+
 class APIGatewayWebSocketEventRequestContextBase(BaseModel):
     extended_request_id: str = Field(alias="extendedRequestId")
     request_time: str = Field(alias="requestTime")

tests/functional/event_handler/_pydantic/test_openapi_params.py

@@ -44,6 +44,7 @@ def handler():
     get = path.get
     assert get.summary == "GET /"
     assert get.operationId == "handler__get"
+    assert get.deprecated is None
 
     assert get.responses is not None
     assert 200 in get.responses.keys()
@@ -388,6 +389,26 @@ def handler(user: Annotated[User, Body(description="This is a user")]):
     assert request_body.content[JSON_CONTENT_TYPE].schema_.description == "This is a user"
 
 
+def test_openapi_with_deprecated_operations():
+    app = APIGatewayRestResolver()
+
+    @app.get("/", deprecated=True)
+    def _get():
+        raise NotImplementedError()
+
+    @app.post("/", deprecated=True)
+    def _post():
+        raise NotImplementedError()
+
+    schema = app.get_openapi_schema()
+
+    get = schema.paths["/"].get
+    assert get.deprecated is True
+
+    post = schema.paths["/"].post
+    assert post.deprecated is True
+
+
 def test_openapi_with_excluded_operations():
     app = APIGatewayRestResolver()