File: concurrency.py

package info (click to toggle)
httpx 0.28.1-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 5,488 kB
  • sloc: python: 12,075; sh: 111; makefile: 10
file content (15 lines) | stat: -rw-r--r-- 324 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
"""
Async environment-agnostic concurrency utilities that are only used in tests.
"""

import asyncio

import sniffio
import trio


async def sleep(seconds: float) -> None:
    if sniffio.current_async_library() == "trio":
        await trio.sleep(seconds)  # pragma: no cover
    else:
        await asyncio.sleep(seconds)