Skill Vault

python-asyncio-concurrency

verified

ca8a7651-3ff0-424d-beef-3ea0d1a2061e

Write correct asyncio Python ā€” event loop, tasks, locking, timeouts, and avoiding blocked/leaked coroutines.

Metadata

Skill ID
ca8a7651-3ff0-424d-beef-3ea0d1a2061e
Version
1
Owner
global
Tags
pythonasyncioconcurrencyasyncbackend
Signature
verified
Integrity
OK
Content hash
6b1ac07fac7c69eef3dae53c8d2cf2f37ee5a303c1432ebf18515cc4668cb1a5
Created
2026-08-05T18:29:01Z

Skill file

Raw skill file (markdown source) 
# Correct Python asyncio

Use when writing concurrent async code ā€” background tasks, parallel I/O, or
orchestrating many awaitables without blocking the loop.

## Run and await

```python
import asyncio


async def fetch(url: str) -> str: ...


async def main() -> None:
    results = await asyncio.gather(fetch("a"), fetch("b"))
```

## Never block the loop

One event loop runs all coroutines ā€” a blocking `time.sleep()` or sync I/O inside
a coroutine stalls everything. Use `asyncio.sleep()`, re-encode sync work with
`await asyncio.to_thread(...)`, or push CPU-bound work to a process.

## Tasks and cancellation

- `asyncio.create_task()` schedules background work; keep a reference or it may
  be garbage-collected mid-flight.
- Use `asyncio.timeout(...)` (3.11+) for bounds; handle `TimeoutError`.

## Locking across coroutines

```python
lock = asyncio.Lock()
async with lock:
    ...  # critical section
```

An `asyncio.Lock` is per-loop ā€” a plain `threading.RLock` is for threads and
won't serialize coroutines correctly. Match the lock to your concurrency model.

## Pitfalls

- Forgetting to `await` returns a coroutine that never runs ā†’ "coroutine was
  never awaited" warning. Intentional fire-and-forget still needs a task.
- Don't share mutable state across tasks without a lock.

Attached files