Work with Python coroutine in Rust?

[kc0506]

I am wondering if there is anyway to deal with Python coroutine in pydantic_core. I found the async-await section of PyO3 docs, but the feature seems not enabled for pydantic_core. Is there any other workarounds that is equivalent to async def and await in Python?

Context

I am suspecting the return_validator logic in pydantic._validate_call is actually a duplicate of the similar logic in call.rs. I tried just remove the Python part and every thing worked fine except for async function, which is currently working because of pydantic/pydantic#7046. The approach taken was to wrap an async function to await the coroutine:

async def return_val_wrapper(aw: Awaitable[Any]) -> None:
    return validator.validate_python(await aw)

self.__return_pydantic_validator__ = return_val_wrapper

Now that I want to remove the return_validator logic in Python and keep the Rust side, I will have to move this wrapper into call.rs, which is the reason I am opening this issue.

[adriangb]

Pydantic is completely sync. We do not work with or interoperate with async stuff at all.

But you can do something like this:

from collections.abc import Awaitable, Callable
from typing import Annotated, Any

import anyio
import anyio.abc
import anyio.from_thread
import anyio.to_thread
from openai import BaseModel
from pydantic import AfterValidator, ValidationInfo


async def double(x: int) -> int:
    return x * 2


def call_in_context(func: Callable[..., Awaitable[Any]], value: Any, info: ValidationInfo) -> Any:
    ctx = info.context
    if 'portal' not in ctx:
        raise RuntimeError('No portal in context')
    portal: anyio.abc.BlockingPortal = ctx['portal']
    return portal.call(func, value)


def call_in_context_wrapper(func: Callable[..., Awaitable[Any]]) -> Callable[..., Any]:
    def wrapper(value: Any, info: ValidationInfo) -> Any:
        return call_in_context(func, value, info)

    return wrapper


class Model(BaseModel):
    x: Annotated[int, AfterValidator(call_in_context_wrapper(double))]


def validate_model[T: BaseModel](portal: anyio.abc.BlockingPortal, data: dict[str, Any], model: type[T]) -> T:
    return model.model_validate(data, context={'portal': portal})


async def main():
    async with anyio.from_thread.BlockingPortal() as portal:
        model = await anyio.to_thread.run_sync(validate_model, portal, {'x': 21}, Model)
        print(model)


if __name__ == '__main__':
    anyio.run(main)
    # Model(x=42)

[kc0506]

I think I might have misphrased my question a bit. I am trying to contribute to pydantic_core (i.e. with Rust code), and the logic I would like to add is as below:

1. Create a Python async function wrapper
2. Inside wrapper, await a Python coroutine
3. Do stuff to the awaited value

So basically I want to simulate the above with PyO3 API in Rust, but I am quite new to it so I couldn't figure out how to achieve this (without enabling the experimental flag).

Still, your example looks interesting to me (although not what I was trying to ask about). Thx.

[adriangb]

Sorry I misunderstood.

This is a very complex topic to contribute to, and as per above there are workarounds that as far as I can tell handle most use cases and may even perform better than a native solution.

I don't want to discourage you, you can still give it a try it might be a good learning experience, but maybe @sydney-runkle can also suggest other areas of the codebase that need some help?

[sydney-runkle]

I'm guessing @davidhewitt could offer some advice re contributing in this area.

Re other areas, @kc0506 has been helping a bunch with validate_call and generics related improvements on the pydantic side. There's definitely some easier issues on the pydantic-core side though - lmk if you want some refs there :).

I've also put together this issue with a fun collection: pydantic/pydantic#10350

[davidhewitt]

TBH, It's not clear to me what the original motivation for this change is. I think moving the async logic to Rust isn't to obvious benefit, and is really hard. Yes, PyO3 is working on support for creating async functions. But in this case I think the goal would be to have SchemaValidator.validate_python return an awaitable, for certain schemas.