Dependency Injection with Pydantic Validation

[bdellegrazie]

Hi,

I'm trying to use the Dependency injection feature with Pydantic validation on routes and it fails.
Does anyone have a simple example where the injected dependency is not validated by Pydantic and isn't part of the Open API signature?

Non-working Example:

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)

Simple pytest test class:

@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"

Fixtures:

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"

Results in:

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.

I have tried to:

• add arbitrary_types_allowed=True to the route but it doesn't work.
• use SkipValidation on the dynamodb_client but that doesn't work either.

Any ideas?

Thanks,

Brett

[leandrodamascena]

Hi Brett,

Thank you for reporting this, and for the complete example with tests and fixtures. That made it easy to reproduce on the first try, which almost never happens with these reports.

This is a bug in Depends, not a mistake on your side. When enable_validation=True, we walk the dependency tree and try to build a Pydantic schema for the return type of every dependency function. For your botocore.Session / boto3.Session / DynamoDBClient dependencies that fails, because those are arbitrary types Pydantic can't model.

The schema is never actually used and we were serializing this. I mean, the return value of a dependency is injected straight into your handler, not serialized or validated. So we were generating something we throw away, and the generation itself was crashing. That is also why arbitrary_types_allowed and SkipValidation on the route didn't help you because the problem was in the dependency, out of their reach.

The fix is to stop building the return schema for dependencies and I'm working on that.

While it's not released yet, a workaround is to drop the return annotation on the dependency functions:

def get_dynamodb_client(boto3_session):   # no -> DynamoDBClient
    return boto3_session.client("dynamodb")

Without the return annotation there's nothing to build a schema from, so the error goes away. Your handler's return type stays annotated and validated as normal.

Thanks again for taking the time to write this up so clearly.

[leandrodamascena]

Closing this discussion because I've opened an issue: #8333