BUG: Dependency Injection with Pydantic Validation #8333

import botocore
import boto3

from typing import Annotated, Any

from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.event_handler.depends import Depends
from aws_lambda_powertools.event_handler.exceptions import (
    NotFoundError,
)
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext
from boto3.dynamodb.types import TypeDeserializer
from pydantic import BaseModel, Field
from types_boto3_dynamodb.client import DynamoDBClient
from types_boto3_dynamodb.type_defs import AttributeValueTypeDef


logger = Logger()
tracer = Tracer()
app: APIGatewayHttpResolver = APIGatewayHttpResolver(
    enable_validation=True,
)


# A simple model
class Example(BaseModel):
    id: Annotated[str, Field(..., description="ID for this object.")]
    info: Annotated[str, Field(..., description="Additional Data")]


def dynamo_to_python(dynamo_object: dict[str, AttributeValueTypeDef]) -> dict[str, Any]:
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in dynamo_object.items()}


def get_botocore_session() -> botocore.session.Session:
    return botocore.session.get_session()


def get_boto3_session(
    botocore_session: Annotated[
        botocore.session.Session,
        Depends(get_botocore_session),
    ],
) -> boto3.Session:
    return boto3.Session(botocore_session=botocore_session)


def get_dynamodb_client(
    boto3_session: Annotated[
        boto3.Session,
        Depends(get_boto3_session),
    ],
) -> DynamoDBClient:
    return boto3_session.client("dynamodb")


@app.get("/example/<id>")
@tracer.capture_method(capture_response=False)
def example(
    id: str,
    dynamodb_client: Annotated[
        DynamoDBClient,
        Depends(get_dynamodb_client),
    ],
) -> Example:
    result = dynamodb_client.get_item(
        TableName="example-table",
        Key={
            "id": {"S": id},
        },
    )
    if "Item" not in result:
        raise NotFoundError
    return Example.model_validate(dynamo_to_python(result["Item"]))


@logger.inject_lambda_context(
    correlation_id_path=correlation_paths.API_GATEWAY_HTTP, clear_state=True
)
@tracer.capture_lambda_handler
def lambda_handler(
    event: dict,
    context: LambdaContext,
) -> dict:
    return app.resolve(event, context)

@pytest.fixture
def example_table(dynamodb: DynamoDBClient) -> str:
    # Create the example table
    dynamodb.create_table(
        TableName="example-table",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    return "example-table"

import os
import json
import time
import uuid

import botocore
import boto3
import pytest

from typing import Final
from aws_lambda_powertools import Logger
from moto import mock_aws
from types_boto3_dynamodb import DynamoDBClient


MOTO_ACCOUNT_ID: Final[str] = "123456789012"
logger = Logger(child=True)


@pytest.fixture(scope="function")
def aws_credentials() -> None:
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"

    # as an additional layer of protection against accidentally touching our real AWS environment
    os.environ["MOTO_ACCOUNT_ID"] = MOTO_ACCOUNT_ID
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"

    # Framework specific
    os.environ["POWERTOOLS_DEV"] = "true"


@pytest.fixture(scope="function")
def mocked_aws(aws_credentials: None):
    with mock_aws():
        yield


@pytest.fixture(scope="function")
def botocore_session(aws_credentials: None) -> botocore.session.Session:  # type: ignore
    with mock_aws():
        yield botocore.session.get_session()


@pytest.fixture(scope="function")
def boto3_session(botocore_session: botocore.session.Session) -> boto3.Session:  # type: ignore
    with mock_aws():
        yield boto3.Session(botocore_session=botocore_session)


@pytest.fixture(scope="function")
def dynamodb(boto3_session: boto3.Session) -> DynamoDBClient:  # type: ignore
    with mock_aws():
        yield boto3_session.client("dynamodb")


@pytest.fixture
def example_table(dynamodb: DynamoDBClient) -> str:
    # Create the example table
    dynamodb.create_table(
        TableName="example-table",
        KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
        AttributeDefinitions=[
            {"AttributeName": "id", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    return "example-table"


@pytest.fixture
def example_item(dynamodb: DynamoDBClient, example_table: str) -> str:
    # Create the example table
    dynamodb.put_item(
        TableName="example-table",
        Item={
            "id": {"S": "example-id"},
            "info": {"S": "example-info"},
        },
    )
    return "example-id"

pydantic.errors.PydanticSchemaGenerationError: Unable to generate pydantic-core schema for <class 'botocore.session.Session'>. Set `arbitrary_types_allowed=True` in the model_config to ignore this error or implement `__get_pydantic_core_schema__` on your type to fully support it.