Process a page of sessions and write them to ClickHouse
(offset: int, limit: int)
| 492 | |
| 493 | |
| 494 | async def process_page(offset: int, limit: int) -> None: |
| 495 | """Process a page of sessions and write them to ClickHouse""" |
| 496 | pending_tasks = set() |
| 497 | async for trace in get_sessions_as_traces(offset=offset, limit=limit): |
| 498 | if len(pending_tasks) >= MAX_CONCURRENT: |
| 499 | done, pending_tasks = await asyncio.wait(pending_tasks, return_when=asyncio.FIRST_COMPLETED) |
| 500 | |
| 501 | for task in done: |
| 502 | await task |
| 503 | |
| 504 | task = asyncio.create_task(write_trace_with_timeout(trace)) |
| 505 | pending_tasks.add(task) |
| 506 | |
| 507 | if pending_tasks: # let the pool drain |
| 508 | done, pending = await asyncio.wait(pending_tasks) |
| 509 | for task in done: |
| 510 | await task |
| 511 | await close_supabase_pool() |
| 512 | |
| 513 | |
| 514 | async def count_session_rows() -> int: |
no test coverage detected
searching dependent graphs…