mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
Joe SpadolaandGitHub
19ffd53f86
* implement native async client * add pylint exceptions * linting * improve test config setup * fix some tests * enable log tables on startup * improve error handling * linting * fix deprecation warning in aiohttp * error handling * separate sync/async stream failure tests * more error handling issues * linting * test updates and improvements * test adjustments * increase async stream buffer size * don't parallelize cloud tests * apply consistency settings to client factory fixture * up cloud par to 4 * enforce sequential consistency in helper * update enable_cleanup_closed vers * add conn abort to wrong port assertion * small test refactor * more consistency * add deadlock check in async queue * enforce consistent use of async iter stream consumption * linting * DO materialize non-streaming queries * accept either concurrent session error * update async client to always reset context * add pipelining to arrow methods * feature parity with recent sync client changes * fix jwt test after bad merge conflict resolution * 0.12.0rc1 release prep * fix assertion in racy test * clean up streaming source on exception * linting * more linting * deprecate legacy async client and replace with async native * update changelog
38 lines
1016 B
Python
38 lines
1016 B
Python
#!/usr/bin/env python -u
|
|
|
|
"""
|
|
Demonstrates concurrent async queries using clickhouse-connect.
|
|
|
|
Executes 10 queries with a concurrency limit of 2. Each query sleeps for 2 seconds,
|
|
so the total wall time is ~10 seconds rather than ~20.
|
|
|
|
Sample output:
|
|
Completed query 1, elapsed: 2002ms
|
|
Completed query 0, elapsed: 2003ms
|
|
Completed query 3, elapsed: 4005ms
|
|
Completed query 2, elapsed: 4005ms
|
|
...
|
|
"""
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
import clickhouse_connect
|
|
|
|
|
|
async def concurrent_queries():
|
|
async with await clickhouse_connect.get_async_client() as client:
|
|
semaphore = asyncio.Semaphore(2)
|
|
start = time.monotonic()
|
|
|
|
async def run_query(num: int):
|
|
async with semaphore:
|
|
await client.query("SELECT sleep(2)")
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
print(f"Completed query {num}, elapsed: {elapsed}ms")
|
|
|
|
await asyncio.gather(*(run_query(i) for i in range(10)))
|
|
|
|
|
|
asyncio.run(concurrent_queries())
|