ResourceWarning: unclosed resource in FastAPI app

[lukapecnik]

I'm getting the following warning while using KeycloakOpenID on async FastAPI endpoints:

/lib/python3.12/site-packages/anyio/_backends/_asyncio.py:910: ResourceWarning: unclosed resource <TCPTransport closed=False reading=False 0x1206f01e0>
  result = None

I have the following logic for access token validation implemented in FastAPI framework:

# keycloak validation implementation
class KeycloakIdentityProviderClient(IdentityProviderClient):
    def __init__(
        self,
        keycloak_settings: KeycloakSettings,
        access_token: Optional[str] = None,
        refresh_token: Optional[str] = None,
    ) -> None:
        super().__init__(access_token, refresh_token)
        self.keycloak_client = KeycloakOpenID(
            server_url=keycloak_settings.url,
            realm_name=keycloak_settings.realm,
            client_id=keycloak_settings.client_id,
        )

    async def validate_access_token(self) -> AuthConsumerData:
        if not self.access_token:
            raise IdentityProviderException(message=NO_ACCESS_TOKEN_PROVIDED_ERROR)

        try:
            data = await self.keycloak_client.a_decode_token(token=self.access_token, validate=True)

            roles = set()
            for role in data["realm_access"]["roles"]:
                roles.add(role)

            return AuthConsumerData(
                id=uuid.UUID(data["sub"]),
                email=data["email"],
                username=data["preferred_username"],
                given_name=data["given_name"],
                family_name=data["family_name"],
                roles=roles,
            )
        except Exception as e:
            logger.error(ACCESS_TOKEN_VALIDATION_ERROR, exc_info=e)
            raise IdentityProviderException(message=ACCESS_TOKEN_VALIDATION_ERROR, original_exception=e)

# role guard used on endpoints
class RoleGuard:
    def __init__(self, required_roles: List[ConsumerRole]):
        self.required_roles = required_roles

    async def __call__(
        self, consumer_data: Annotated[AuthConsumerData, Depends(validate_identity)]
    ) -> AuthConsumerData:
        for role in self.required_roles:
            if role.value not in consumer_data.roles:
                raise HTTPException(status_code=403, detail=NO_PERMISSION_TO_ACCESS_RESOURCE_ERROR)

        return consumer_data

# validation dependency
async def validate_identity(
    identity_provider_client: KeycloakIdentityProviderClientDependency, # resolves KeycloakIdentityProviderClient
) -> AuthConsumerData:
    return await identity_provider_client.validate_access_token()


# Example endpoint
@router.get(
    "/test"
)
async def test(
    _: AuthConsumerData = Depends(RoleGuard(required_roles=[ConsumerRole.EXECUTE])),
):
    # whatever logic
    pass