fix(event_handler): sync alias and validation_alias for Pydantic 2.12+ compatibility (aws-powertools#7901)

* fix: add support for Pydantic 2.12+

* fix: add support for Pydantic 2.12+

* fix: add support for Pydantic 2.12+

* fix: add support for Pydantic 2.12+

Author: leandrodamascena

aws_lambda_powertools/event_handler/openapi/params.py

@@ -103,7 +103,7 @@ def __init__(
         alias_priority: int | None = _Unset,
         # MAINTENANCE: update when deprecating Pydantic v1, import these types
         # MAINTENANCE: validation_alias: str | AliasPath | AliasChoices | None
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         title: str | None = None,
         description: str | None = None,
@@ -217,6 +217,14 @@ def __init__(
 
         self.openapi_examples = openapi_examples
 
+        # Pydantic 2.12+ no longer copies alias to validation_alias automatically
+        # Ensure alias and validation_alias are in sync when only one is provided
+        if validation_alias is _Unset and alias is not None:
+            validation_alias = alias
+        elif alias is None and validation_alias is not _Unset and validation_alias is not None:
+            alias = validation_alias
+            kwargs["alias"] = alias
+
         kwargs.update(
             {
                 "annotation": annotation,
@@ -254,7 +262,7 @@ def __init__(
         alias_priority: int | None = _Unset,
         # MAINTENANCE: update when deprecating Pydantic v1, import these types
         # MAINTENANCE: validation_alias: str | AliasPath | AliasChoices | None
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         title: str | None = None,
         description: str | None = None,
@@ -386,7 +394,7 @@ def __init__(
         annotation: Any | None = None,
         alias: str | None = None,
         alias_priority: int | None = _Unset,
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         title: str | None = None,
         description: str | None = None,
@@ -517,7 +525,7 @@ def __init__(
         alias_priority: int | None = _Unset,
         # MAINTENANCE: update when deprecating Pydantic v1, import these types
         # str | AliasPath | AliasChoices | None
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         convert_underscores: bool = True,
         title: str | None = None,
@@ -667,7 +675,7 @@ def __init__(
         alias_priority: int | None = _Unset,
         # MAINTENANCE: update when deprecating Pydantic v1, import these types
         # str | AliasPath | AliasChoices | None
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         title: str | None = None,
         description: str | None = None,
@@ -720,6 +728,13 @@ def __init__(
             kwargs["openapi_examples"] = openapi_examples
         current_json_schema_extra = json_schema_extra or extra
 
+        # Pydantic 2.12+ no longer copies alias to validation_alias automatically
+        # Ensure alias and validation_alias are in sync when only one is provided
+        if validation_alias is _Unset and alias is not None:
+            validation_alias = alias
+        elif alias is None and validation_alias is not _Unset and validation_alias is not None:
+            alias = validation_alias
+            kwargs["alias"] = alias
         self.openapi_examples = openapi_examples
 
         kwargs.update(
@@ -758,7 +773,7 @@ def __init__(
         alias_priority: int | None = _Unset,
         # MAINTENANCE: update when deprecating Pydantic v1, import these types
         # str | AliasPath | AliasChoices | None
-        validation_alias: str | None = None,
+        validation_alias: str | None = _Unset,
         serialization_alias: str | None = None,
         title: str | None = None,
         description: str | None = None,

docs/core/logger.md

@@ -196,18 +196,18 @@ You can append your own keys to your existing Logger via `append_keys(**addition
 The append_context_keys method allows temporary modification of a Logger instance's context without creating a new logger. It's useful for adding context keys to specific workflows while maintaining the logger's overall state and simplicity.
 
 ???+ danger "Important: Keys are removed on context exit, even if they existed before"
-	All keys added within the context are removed when exiting, **including keys that already existed with the same name**. 
-	
+	All keys added within the context are removed when exiting, **including keys that already existed with the same name**.
+
 	If you need to temporarily override a key's value while preserving the original, use `append_keys()` for persistent keys and avoid key name collisions with `append_context_keys()`.
-	
+
 	**Example of key collision:**
 	```python
 	logger.append_keys(order_id="ORD-123")  # Persistent key
 	logger.info("Order received")  # Has order_id="ORD-123"
-	
+
 	with logger.append_context_keys(order_id="ORD-CHILD"):  # Overwrites
 	    logger.info("Processing")  # Has order_id="ORD-CHILD"
-	
+
 	logger.info("Order completed")  # order_id key is now MISSING!
 	```
 
@@ -1014,7 +1014,7 @@ You can change the order of [standard Logger keys](#standard-structured-keys) or
 By default, this Logger and the standard logging library emit records with the default AWS Lambda timestamp in **UTC**.
 
 <!-- markdownlint-disable MD013 -->
-If you prefer to log in a specific timezone, you can configure it by setting the `TZ` environment variable. You can do this either as an AWS Lambda environment variable or directly within your Lambda function settings. [Click here](https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html#configuration-envvars-runtime){target="_blank"} for a comprehensive list of available Lambda environment variables.
+If you prefer to log in a specific timezone, you can configure it by setting the `TZ` environment variable. You can do this either as an AWS Lambda environment variable or directly within your Lambda function settings. See the [Lambda environment variables documentation](https://docs.aws.amazon.com/lambda/latest/dg/configuration-envvars.html#configuration-envvars-runtime){target="_blank"} for a comprehensive list of available options.
 <!-- markdownlint-enable MD013 -->
 
 ???+ tip

tests/functional/event_handler/_pydantic/test_openapi_params.py

@@ -1269,6 +1269,270 @@ def list_items(limit: Annotated[constrained_int, Query()] = 10):
     assert limit_param.required is False
 
 
+def test_query_alias_sets_validation_alias_automatically():
+    """
+    When alias is set but validation_alias is not,
+    validation_alias should be automatically set to alias value.
+    This ensures compatibility with Pydantic 2.12+.
+    """
+    from annotated_types import Ge, Le
+    from pydantic import StringConstraints
+
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    # AND constrained types using annotated_types
+    IntQuery = Annotated[int, Ge(1), Le(100)]
+    StrQuery = Annotated[str, StringConstraints(min_length=4, max_length=128)]
+
+    @app.get("/foo")
+    def get_foo(
+        str_query: Annotated[StrQuery, Query(alias="strQuery")],
+        int_query: Annotated[IntQuery, Query(alias="intQuery")],
+    ):
+        return {"int_query": int_query, "str_query": str_query}
+
+    # WHEN sending a request with aliased query parameters
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "queryStringParameters": {
+            "intQuery": "20",
+            "strQuery": "fooBarFizzBuzz",
+        },
+    }
+
+    # THEN the request should succeed with correct values
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["int_query"] == 20
+    assert body["str_query"] == "fooBarFizzBuzz"
+
+
+def test_query_alias_with_multivalue_query_string_parameters():
+    """
+    Ensure alias works with multiValueQueryStringParameters.
+    """
+    from annotated_types import Ge, Le
+    from pydantic import StringConstraints
+
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    IntQuery = Annotated[int, Ge(1), Le(100)]
+    StrQuery = Annotated[str, StringConstraints(min_length=4, max_length=128)]
+
+    @app.get("/foo")
+    def get_foo(
+        str_query: Annotated[StrQuery, Query(alias="strQuery")],
+        int_query: Annotated[IntQuery, Query(alias="intQuery")],
+    ):
+        return {"int_query": int_query, "str_query": str_query}
+
+    # WHEN sending a request with multiValueQueryStringParameters
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "multiValueQueryStringParameters": {
+            "intQuery": ["20"],
+            "strQuery": ["fooBarFizzBuzz"],
+        },
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["int_query"] == 20
+    assert body["str_query"] == "fooBarFizzBuzz"
+
+
+def test_query_explicit_validation_alias_takes_precedence():
+    """
+    Explicitly set validation_alias is preserved and not overwritten by alias.
+    The alias is used by Powertools to extract the value from the request,
+    while validation_alias is used by Pydantic for internal validation.
+    """
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/foo")
+    def get_foo(
+        my_param: Annotated[str, Query(alias="aliasName", validation_alias="validationAliasName")],
+    ):
+        return {"my_param": my_param}
+
+    # WHEN sending a request with the alias name (used by Powertools to extract value)
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "queryStringParameters": {
+            "aliasName": "test_value",
+        },
+    }
+
+    # THEN the request should succeed using alias for extraction
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["my_param"] == "test_value"
+
+
+def test_header_alias_sets_validation_alias_automatically():
+    """
+    Header alias should also set validation_alias automatically.
+    """
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/foo")
+    def get_foo(
+        custom_header: Annotated[str, Header(alias="X-Custom-Header")],
+    ):
+        return {"custom_header": custom_header}
+
+    # WHEN sending a request with the aliased header
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "headers": {
+            "X-Custom-Header": "header_value",
+        },
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["custom_header"] == "header_value"
+
+
+def test_query_without_alias_works_normally():
+    """
+    Query without alias continues to work normally.
+    """
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.get("/foo")
+    def get_foo(
+        my_param: Annotated[str, Query()],
+    ):
+        return {"my_param": my_param}
+
+    # WHEN sending a request with the parameter name
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "queryStringParameters": {
+            "my_param": "test_value",
+        },
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["my_param"] == "test_value"
+
+
+def test_query_validation_alias_only_sets_alias_automatically():
+    """
+    When only validation_alias is set (without alias),
+    alias should be automatically set to validation_alias value.
+    This ensures the middleware can find the parameter in the request.
+    """
+    from annotated_types import Ge, Le
+    from pydantic import StringConstraints
+
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    IntQuery = Annotated[int, Ge(1), Le(100)]
+    StrQuery = Annotated[str, StringConstraints(min_length=4, max_length=128)]
+
+    @app.get("/foo")
+    def get_foo(
+        str_query: Annotated[StrQuery, Query(validation_alias="strQuery")],
+        int_query: Annotated[IntQuery, Query(validation_alias="intQuery")],
+    ):
+        return {"int_query": int_query, "str_query": str_query}
+
+    # WHEN sending a request with validation_alias names
+    event = {
+        "httpMethod": "GET",
+        "path": "/foo",
+        "queryStringParameters": {
+            "intQuery": "20",
+            "strQuery": "fooBarFizzBuzz",
+        },
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["int_query"] == 20
+    assert body["str_query"] == "fooBarFizzBuzz"
+
+
+def test_body_alias_sets_validation_alias_automatically():
+    """
+    When alias is set but validation_alias is not in Body,
+    validation_alias should be automatically set to alias value.
+    """
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/foo")
+    def post_foo(
+        my_body: Annotated[str, Body(alias="myBody")],
+    ):
+        return {"my_body": my_body}
+
+    # WHEN sending a request with body
+    event = {
+        "httpMethod": "POST",
+        "path": "/foo",
+        "body": '"test_value"',
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["my_body"] == "test_value"
+
+
+def test_body_validation_alias_only_sets_alias_automatically():
+    """
+    When only validation_alias is set (without alias) in Body,
+    alias should be automatically set to validation_alias value.
+    """
+    # GIVEN an APIGatewayRestResolver with validation enabled
+    app = APIGatewayRestResolver(enable_validation=True)
+
+    @app.post("/foo")
+    def post_foo(
+        my_body: Annotated[str, Body(validation_alias="myBody")],
+    ):
+        return {"my_body": my_body}
+
+    # WHEN sending a request with body
+    event = {
+        "httpMethod": "POST",
+        "path": "/foo",
+        "body": '"test_value"',
+    }
+
+    # THEN the request should succeed
+    result = app(event, {})
+    assert result["statusCode"] == 200
+    body = json.loads(result["body"])
+    assert body["my_body"] == "test_value"
+
+
 def test_body_class_annotation_without_parentheses():
     """
     GIVEN an endpoint using Body class (not instance) in Annotated