Asynchronous call inside a Policy Pack

[LucaCaste]

What happened?

When I'm trying to call an asynchronous API inside a policy pack, it generates this error:
"error: Exception calling application: There is no current event loop in thread 'ThreadPoolExecutor-0_0'. ".
Even if I add async/await to the function or even using pulumi.routine.sync_await the asyncronous call doesn't work.
I think is missing the asyncio.get_event_loop() function in the generation of the thread when is called the policy Pack.

This is the function:

def validator(args: ResourceValidationArgs, report_violation: ReportViolation):
    if args.resource_type != "aws:s3/bucket:Bucket":
        return
    
    resource =_sync_await(aws.pricing.get_product(
    service_code="AmazonS3",
    filters= [
        aws.pricing.GetProductFilterArgs(field = "productFamily", value = "Storage"),
    ]))
    if resource is None:
        report_violation("The resource has not been found")
    else:
        print(resource)
    if resource["price"] > 0:
        report_violation("The resource has a cost")
    else:
        report_violation("The resourse is free")

Expected Behavior

I would expect that at least the function generates an output but it doesn't.

Steps to reproduce

Create a policy pack.
Add the function written in the description.
Create a ResourceValidationPolicy and a PolicyPack
Try to execute it with a project.

Output of pulumi about

error: Exception calling application: There is no current event loop in thread 'ThreadPoolExecutor-0_0'

Additional context

No response

Contributing

Vote on this issue by adding a 👍 reaction.
To contribute a fix for this issue, leave a comment (and link to your pull request, if you've opened one already).

[justinvp]

Hmm, you should be able to have an async validator functio. We have tests for this:

pulumi-policy/sdk/python/lib/test/test_policy.py

Lines 107 to 115 in 8cef30d

	def test_async_validate(self):
	async def validate(args, report_violation: ReportViolation):
	report_violation("first")
	await asyncio.sleep(0.1)
	report_violation("second")
	policy = ResourceValidationPolicy("name", "desc", validate)
	violations = run_policy(policy)
	self.assertEqual(["first", "second"], violations)

I'm not familiar with aws.pricing.get_product. Does it return an Awaitable? If so, what happens if your function looks like this:

async def validator(args: ResourceValidationArgs, report_violation: ReportViolation):
    if args.resource_type != "aws:s3/bucket:Bucket":
        return
    
    resource = await aws.pricing.get_product(
    service_code="AmazonS3",
    filters= [
        aws.pricing.GetProductFilterArgs(field = "productFamily", value = "Storage"),
    ])
    if resource is None:
        report_violation("The resource has not been found")
    else:
        print(resource)
    if resource["price"] > 0:
        report_violation("The resource has a cost")
    else:
        report_violation("The resourse is free")

[LucaCaste]

In that case the validation succeed in any case without printing anything... I've also tried to do some output on files, but nothing.
About your question, yes, the function returns an Awaitable