timeout-executor

how to install

$ pip install timeout_executor
# or
$ pip install "timeout_executor[uvloop]"
# or
$ pip install "timeout_executor[jinja]"

how to use

from __future__ import annotations

import time

import anyio

from timeout_executor import AsyncResult, TimeoutExecutor


def sample_sync_func(x: float) -> str:
    time.sleep(x)
    return "done"


async def sample_async_func(x: float) -> str:
    await anyio.sleep(x)
    return "done"


def main() -> None:
    executor = TimeoutExecutor(2)
    result = executor.apply(sample_sync_func, 10)
    assert isinstance(result, AsyncResult)

    try:
        value = result.result()
    except Exception as exc:
        assert isinstance(exc, TimeoutError)

    result = executor.apply(sample_async_func, 1)
    assert isinstance(result, AsyncResult)
    value = result.result()
    assert value == "done"

    result = executor.apply(lambda: "done")
    assert isinstance(result, AsyncResult)
    value = result.result()
    assert value == "done"


async def async_main() -> None:
    executor = TimeoutExecutor(2)
    result = await executor.delay(sample_sync_func, 10)
    assert isinstance(result, AsyncResult)

    try:
        value = await result.delay()
    except Exception as exc:
        assert isinstance(exc, TimeoutError)

    result = await executor.delay(sample_async_func, 1)
    assert isinstance(result, AsyncResult)
    value = await result.delay()
    assert value == "done"

    result = await executor.delay(lambda: "done")
    assert isinstance(result, AsyncResult)
    value = await result.delay()
    assert value == "done"


if __name__ == "__main__":
    main()
    anyio.run(async_main)

License

MIT, see LICENSE.

Metadata

Release files for timeout-executor 0.9.1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for timeout-executor 0.9.1
File	Size	Uploaded
timeout_executor-0.9.1.tar.gz	13.9 kB	Oct 17, 2024	Details

Built distribution (wheel)

Table of built distributions (wheels) for timeout-executor 0.9.1
File	Interpreter	ABI	Platform	Reset
timeout_executor-0.9.1-py3-none-any.whl 18.9 kB Oct 17, 2024	Python 3	none	any	Details

Total release size: 32.7 kB

Release files / timeout_executor-0.9.1.tar.gz

Download URL	timeout_executor-0.9.1.tar.gz
Size	13.9 kB
Tags	Source
SHA-256 checksum How to use checksums	`e499b05897c8e46dcabd95ea62e30fc589b2441f0416c7cfd84b3a82effcfe6d`
BLAKE2b-256 checksum How to use checksums	`2386077caa9fd4df1dc29ade7af5186065ca5c244aa8496a88d4d6606686cfab`
Upload date	Oct 17, 2024
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`uv/0.4.22`

Release files / timeout_executor-0.9.1-py3-none-any.whl

Download URL	timeout_executor-0.9.1-py3-none-any.whl
Size	18.9 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`1f7f0ce010253c0dbb49a659717b77ff4d6cb43a6a160ca6db376dd9e857ce29`
BLAKE2b-256 checksum How to use checksums	`22fa76e4bee654f838e248572292e29164ac0a0fee2d01b6e6df4268312e97c6`
Upload date	Oct 17, 2024
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`uv/0.4.22`