Skip to content

core

Classes:

  • AnySyncContextManager –

    Abstract base class for an async context manager that can be used synchronously.

  • AnySyncCoroutine –

    Abstract base class for an async function that can be used synchronously.

  • AnySyncGenerator –

    Abstract base class for an async generator that can be used synchronously.

  • AnySyncIterator –

    Abstract base class for an async generator that can be used synchronously.

Functions:

  • contextmanager –

    Allow an async context manager to optionally run synchronously.

  • coroutine –

    Allow an async function to optionally run synchronously by calling run() on the result.

  • generator –

    Allow an async generator to optionally run synchronously.

  • iterator –

    Allow an async iterator to optionally run synchronously.

  • run –

    Run a coroutine synchronously.

  • wrap_async_context_manager –

    Wrap an async context manager so that it can be run synchronously.

  • wrap_async_generator –

    Wrap an async generator so that it can be run synchronously.

  • wrap_async_iterator –

    Wrap an async iterator so that it can be run synchronously.

  • wrap_coroutine –

    Wrap an coroutine so that it can be run synchronously.

AnySyncContextManager

Bases: AbstractContextManager[R], AbstractAsyncContextManager[R]

Abstract base class for an async context manager that can be used synchronously.

AnySyncCoroutine 

AnySyncCoroutine(coro: Coroutine[Y_any, S_any, R])

Bases: Coroutine[Y_any, S_any, R], Generic[R, Y_any, S_any], ABC

Abstract base class for an async function that can be used synchronously.

Methods:

  • run –

    Run the coroutine synchronously.

run 

run(timeout: float | None = None) -> R

Run the coroutine synchronously.

AnySyncGenerator

Bases: AnySyncIterator[Y], AsyncGenerator[Y, S], Generator[Y, S], ABC

Abstract base class for an async generator that can be used synchronously.

Methods:

  • asend –

    Send a value into the generator.

  • athrow –

    Raise an exception in the generator.

  • send –

    Send a value into the generator.

  • throw –

    Raise an exception in the generator.

asend abstractmethod async 

asend(value: S) -> Y

Send a value into the generator.

athrow abstractmethod async 

athrow(
    typ: type[BaseException],
    val: BaseException | Any = None,
    tb: TracebackType | None = None,
) -> Y

Raise an exception in the generator.

send 

send(value: S) -> Y

Send a value into the generator.

throw 

throw(
    typ: type[BaseException],
    val: BaseException | Any = None,
    tb: TracebackType | None = None,
) -> Y

Raise an exception in the generator.

AnySyncIterator

Bases: AsyncIterator[Y], Iterator[Y], ABC

Abstract base class for an async generator that can be used synchronously.

contextmanager 

contextmanager(
    func: Callable[P, AsyncIterator[R]]
) -> Callable[P, AnySyncContextManager[R]]

Allow an async context manager to optionally run synchronously.

coroutine 

coroutine(
    func: Callable[P, Coroutine[Y, S, R]]
) -> Callable[P, AnySyncCoroutine[R, Y, S]]

Allow an async function to optionally run synchronously by calling run() on the result.

generator 

generator(
    func: Callable[P, AsyncGenerator[Y, S]]
) -> Callable[P, AnySyncGenerator[Y, S]]

Allow an async generator to optionally run synchronously.

iterator 

iterator(
    func: Callable[P, AsyncIterator[Y]]
) -> Callable[P, AnySyncIterator[Y]]

Allow an async iterator to optionally run synchronously.

run 

run(coro: Coroutine[Any, Any, R]) -> R

Run a coroutine synchronously.

wrap_async_context_manager 

wrap_async_context_manager(
    manager: AbstractAsyncContextManager[R],
) -> AnySyncContextManager[R]

Wrap an async context manager so that it can be run synchronously.

wrap_async_generator 

wrap_async_generator(
    generator: AsyncGenerator[Y, S]
) -> AnySyncGenerator[Y, S]

Wrap an async generator so that it can be run synchronously.

wrap_async_iterator 

wrap_async_iterator(
    iterator: AsyncIterator[Y],
) -> AnySyncIterator[Y]

Wrap an async iterator so that it can be run synchronously.

wrap_coroutine 

wrap_coroutine(
    coroutine: Coroutine[Y, S, R]
) -> AnySyncCoroutine[R, Y, S]

Wrap an coroutine so that it can be run synchronously.