Python 비동기 완전 정리

Python code

import asyncio
import time

# 기본 코루틴
async def fetch_data(url: str, delay: float) -> str:
    print(f"시작: {url}")
    await asyncio.sleep(delay)   # 비차단 대기
    print(f"완료: {url}")
    return f"data:{url}"

# asyncio.gather — 병렬 실행
async def main_parallel():
    start = time.perf_counter()

    results = await asyncio.gather(
        fetch_data("url1", 1.0),
        fetch_data("url2", 1.5),
        fetch_data("url3", 0.5),
    )

    elapsed = time.perf_counter() - start
    print(f"총 소요: {elapsed:.2f}초")   # ~1.5초 (병렬)
    print(results)

# asyncio.create_task — 태스크 생성
async def main_tasks():
    task1 = asyncio.create_task(fetch_data("a", 1.0))
    task2 = asyncio.create_task(fetch_data("b", 0.5))

    # 여기서 다른 작업 가능
    print("태스크 생성 후 바로 실행 가능")

    r1 = await task1
    r2 = await task2
    return r1, r2

# 타임아웃
async def with_timeout():
    try:
        result = await asyncio.wait_for(
            fetch_data("slow", 5.0),
            timeout=2.0
        )
    except asyncio.TimeoutError:
        print("타임아웃!")

asyncio.run(main_parallel())

Python code

import asyncio
from contextlib import asynccontextmanager

# async context manager
class AsyncDB:
    async def __aenter__(self):
        print("DB 연결")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, *args):
        print("DB 연결 해제")
        await asyncio.sleep(0.05)

    async def query(self, sql: str) -> list:
        await asyncio.sleep(0.1)
        return [{"id": 1, "name": "Alice"}]

# @asynccontextmanager 데코레이터 방식
@asynccontextmanager
async def managed_connection(url: str):
    print(f"연결: {url}")
    connection = {"url": url, "open": True}
    try:
        yield connection
    finally:
        connection["open"] = False
        print(f"해제: {url}")

# async generator — async for 사용
async def paginate(total: int, page_size: int = 10):
    for page in range(0, total, page_size):
        await asyncio.sleep(0.01)   # DB 조회 시뮬레이션
        yield list(range(page, min(page + page_size, total)))

async def main():
    async with AsyncDB() as db:
        rows = await db.query("SELECT * FROM users")
        print(rows)

    async with managed_connection("postgres://localhost") as conn:
        print(f"연결 상태: {conn['open']}")

    # async for
    async for page in paginate(35):
        print(f"페이지: {page}")

asyncio.run(main())

Python code

import asyncio

# asyncio.Queue — Producer/Consumer 패턴
async def producer(queue: asyncio.Queue, count: int):
    for i in range(count):
        await asyncio.sleep(0.1)
        await queue.put(f"item-{i}")
        print(f"생산: item-{i}")
    await queue.put(None)  # 종료 신호

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 다음 컨슈머에게 전달
            break
        await asyncio.sleep(0.2)
        print(f"{name} 소비: {item}")
        queue.task_done()

# Semaphore — 동시 실행 수 제한
async def fetch_with_limit(
    sem: asyncio.Semaphore,
    url: str
) -> str:
    async with sem:   # 동시에 최대 3개만 실행
        await asyncio.sleep(0.1)
        return f"data:{url}"

async def main():
    # Queue 예제
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 5),
        consumer(queue, "Worker-1"),
        consumer(queue, "Worker-2"),
    )

    # Semaphore 예제
    sem = asyncio.Semaphore(3)  # 동시 3개 제한
    urls = [f"url-{i}" for i in range(10)]
    results = await asyncio.gather(
        *[fetch_with_limit(sem, url) for url in urls]
    )
    print(f"완료: {len(results)}개")

asyncio.run(main())

Python code

<span class="cm">// uv 패키지 매니저 예제
// data/prompts.js의 생성 프롬프트로 상세 코드 생성 가능</span>
fun main() { println("uv 패키지 매니저") }

Python code

import time
import math
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed

def is_prime(n):
    """CPU 바운드 작업: 소수 판별"""
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def count_primes(start, end):
    """범위 내 소수 개수 계산"""
    return sum(1 for n in range(start, end) if is_prime(n))

def benchmark(executor_class, name, ranges):
    start = time.perf_counter()
    results = []
    with executor_class(max_workers=4) as executor:
        futures = {
            executor.submit(count_primes, s, e): (s, e)
            for s, e in ranges
        }
        for future in as_completed(futures):
            s, e = futures[future]
            count = future.result()
            results.append(count)
    elapsed = time.perf_counter() - start
    total = sum(results)
    print(f"  {name}: {total}개 소수, {elapsed:.3f}초")
    return elapsed

if __name__ == '__main__':
    RANGE = 500_000
    CHUNKS = 4
    chunk_size = RANGE // CHUNKS
    ranges = [(i * chunk_size, (i+1) * chunk_size) for i in range(CHUNKS)]

    print(f"0~{RANGE:,} 범위 소수 계산 (4 workers):")

    # 순차 실행
    start = time.perf_counter()
    total = count_primes(0, RANGE)
    seq_time = time.perf_counter() - start
    print(f"  순차 실행: {total}개 소수, {seq_time:.3f}초")

    # 멀티프로세스 (CPU 바운드에 효과적)
    proc_time = benchmark(ProcessPoolExecutor, "프로세스풀", ranges)

    # 멀티스레드 (GIL로 인해 CPU 바운드에 비효율)
    thread_time = benchmark(ThreadPoolExecutor, "스레드풀  ", ranges)

    print(f"\n속도 향상: 프로세스풀 {seq_time/proc_time:.1f}x")

Python code

import asyncio

async def say(text: str, delay: float):
    await asyncio.sleep(delay)
    print(f"[{delay}s] {text}")
    return text

async def main():
    # 순차 실행: 3초
    await say("첫 번째", 1)
    await say("두 번째", 2)

    # 병렬 실행: 2초 (가장 긴 시간)
    results = await asyncio.gather(
        say("동시-A", 2),
        say("동시-B", 1),
        say("동시-C", 1.5),
    )
    print(f"결과: {results}")

# 실행
asyncio.run(main())

# 이벤트 루프 직접 제어 (드물게 필요)
# loop = asyncio.get_event_loop()
# loop.run_until_complete(main())

Python code

import asyncio

async def fetch(url: str, delay: float) -> dict:
    await asyncio.sleep(delay)  # 네트워크 요청 시뮬레이션
    if "error" in url:
        raise ConnectionError(f"{url} 연결 실패")
    return {"url": url, "status": 200}

async def main():
    # 타임아웃 설정
    try:
        result = await asyncio.wait_for(
            fetch("https://slow.api", 10), timeout=2.0
        )
    except asyncio.TimeoutError:
        print("타임아웃 발생!")

    # gather로 에러 처리
    results = await asyncio.gather(
        fetch("https://api1.com", 1),
        fetch("https://error.api", 1),
        fetch("https://api3.com", 0.5),
        return_exceptions=True,  # 예외를 결과로 반환
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"  에러: {r}")
        else:
            print(f"  성공: {r}")

    # as_completed: 먼저 끝나는 순서대로
    tasks = [
        asyncio.create_task(fetch(f"https://api{i}.com", 3-i))
        for i in range(3)
    ]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"  완료: {result['url']}")

asyncio.run(main())

Python code

import asyncio

async def async_range(start: int, stop: int, delay: float = 0.1):
    """비동기 제너레이터"""
    for i in range(start, stop):
        await asyncio.sleep(delay)
        yield i

async def countdown(name: str, n: int):
    async for i in async_range(0, n, 0.2):
        print(f"  {name}: {n - i}")
    print(f"  {name}: 완료!")

async def main():
    # 비동기 컴프리헨션
    squares = [x ** 2 async for x in async_range(1, 6)]
    print(f"비동기 제곱: {squares}")

    # 비동기 필터링
    evens = [x async for x in async_range(1, 11) if x % 2 == 0]
    print(f"짝수: {evens}")

    # 여러 비동기 스트림 동시 처리
    await asyncio.gather(
        countdown("A", 3),
        countdown("B", 5),
    )

asyncio.run(main())

Python code

import asyncio
from contextlib import asynccontextmanager

# 클래스 기반 비동기 컨텍스트 매니저
class AsyncConnection:
    def __init__(self, host: str):
        self.host = host

    async def __aenter__(self):
        print(f"연결 중: {self.host}")
        await asyncio.sleep(0.5)  # 연결 시뮬레이션
        print(f"연결 완료: {self.host}")
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print(f"연결 해제: {self.host}")
        await asyncio.sleep(0.1)
        return False  # 예외 전파

    async def query(self, sql: str) -> str:
        await asyncio.sleep(0.2)
        return f"결과: {sql}"

# 함수 기반 (asynccontextmanager)
@asynccontextmanager
async def timer(name: str):
    import time
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name}: {elapsed:.3f}초")

async def main():
    async with AsyncConnection("db.example.com") as conn:
        result = await conn.query("SELECT * FROM users")
        print(result)

    async with timer("작업"):
        await asyncio.sleep(1)

asyncio.run(main())

Python code

import asyncio

# aiohttp 없이 시뮬레이션
async def fake_fetch(url: str, delay: float = 1.0) -> dict:
    """HTTP 요청 시뮬레이션"""
    await asyncio.sleep(delay)
    return {"url": url, "status": 200, "body": f"응답: {url}"}

async def fetch_all(urls: list[str]) -> list[dict]:
    """여러 URL을 동시에 요청"""
    tasks = [fake_fetch(url, 0.5) for url in urls]
    return await asyncio.gather(*tasks)

async def fetch_with_limit(urls, max_concurrent=3):
    """동시 요청 수 제한"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_fetch(url):
        async with semaphore:
            return await fake_fetch(url, 1.0)

    return await asyncio.gather(
        *[limited_fetch(url) for url in urls]
    )

async def main():
    urls = [f"https://api.example.com/item/{i}" for i in range(5)]

    import time
    start = time.perf_counter()
    results = await fetch_all(urls)
    elapsed = time.perf_counter() - start
    print(f"{len(results)}개 요청 완료: {elapsed:.2f}초")

    start = time.perf_counter()
    results = await fetch_with_limit(urls, max_concurrent=2)
    elapsed = time.perf_counter() - start
    print(f"제한 요청 완료: {elapsed:.2f}초")

asyncio.run(main())

Python code

import asyncio

# run_in_executor로 동기 I/O를 비동기로
async def read_file_async(path: str) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None,  # 기본 ThreadPoolExecutor
        lambda: open(path).read() if __import__('os').path.exists(path)
                else f"파일 없음: {path}"
    )

# 여러 파일 동시 읽기
async def read_many(paths: list[str]):
    tasks = [read_file_async(p) for p in paths]
    return await asyncio.gather(*tasks)

# aiofiles 패턴 시뮬레이션
class AsyncFile:
    def __init__(self, content=""):
        self.content = content

    async def __aenter__(self):
        await asyncio.sleep(0.01)  # I/O 시뮬레이션
        return self

    async def __aexit__(self, *args):
        pass

    async def read(self):
        await asyncio.sleep(0.01)
        return self.content

    async def write(self, data):
        await asyncio.sleep(0.01)
        self.content += data

async def main():
    # 동시 파일 처리
    async with AsyncFile("Hello, World!") as f:
        content = await f.read()
        print(f"내용: {content}")

    print("비동기 파일 I/O 완료")

asyncio.run(main())

Python code

import asyncio
import random

async def producer(queue: asyncio.Queue, name: str, count: int):
    for i in range(count):
        item = f"{name}-{i}"
        await asyncio.sleep(random.uniform(0.1, 0.5))
        await queue.put(item)
        print(f"  생산: {item}")
    await queue.put(None)  # 종료 신호

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        await asyncio.sleep(random.uniform(0.1, 0.3))
        print(f"  소비 [{name}]: {item}")
        queue.task_done()

async def main():
    queue: asyncio.Queue = asyncio.Queue(maxsize=5)

    # 생산자 2개, 소비자 3개
    producers = [
        asyncio.create_task(producer(queue, f"P{i}", 3))
        for i in range(2)
    ]
    consumers = [
        asyncio.create_task(consumer(queue, f"C{i}"))
        for i in range(3)
    ]

    await asyncio.gather(*producers)
    # 남은 소비자에 종료 신호
    for _ in range(len(consumers) - len(producers)):
        await queue.put(None)
    await asyncio.gather(*consumers)
    print("모든 작업 완료")

asyncio.run(main())

Python code

import asyncio
import time

async def limited_task(sem: asyncio.Semaphore, task_id: int):
    async with sem:
        print(f"  시작 Task-{task_id} (t={time.perf_counter():.1f})")
        await asyncio.sleep(1)  # 작업 시뮬레이션
        print(f"  완료 Task-{task_id}")
        return task_id

async def main():
    sem = asyncio.Semaphore(3)  # 최대 3개 동시 실행

    start = time.perf_counter()
    tasks = [limited_task(sem, i) for i in range(9)]
    results = await asyncio.gather(*tasks)
    elapsed = time.perf_counter() - start

    print(f"결과: {results}")
    print(f"소요: {elapsed:.1f}초")  # 약 3초 (3배치 × 1초)

    # BoundedSemaphore: 릴리스 횟수 초과 방지
    bounded = asyncio.BoundedSemaphore(2)
    async with bounded:
        print("BoundedSemaphore 사용")
    # bounded.release()  # ValueError! (acquire 없이 release)

asyncio.run(main())

Python code

import asyncio

class AsyncCounter:
    """비동기 카운터 이터레이터"""
    def __init__(self, start: int, stop: int):
        self.current = start
        self.stop = stop

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.stop:
            raise StopAsyncIteration
        value = self.current
        self.current += 1
        await asyncio.sleep(0.1)  # 비동기 작업
        return value

class AsyncChain:
    """여러 비동기 이터러블을 연결"""
    def __init__(self, *iterables):
        self._iterables = iterables

    def __aiter__(self):
        return self._iterate()

    async def _iterate(self):
        for iterable in self._iterables:
            async for item in iterable:
                yield item

async def main():
    # 기본 사용
    async for num in AsyncCounter(0, 5):
        print(f"  카운트: {num}")

    # 비동기 이터러블 체이닝
    chained = AsyncChain(
        AsyncCounter(0, 3),
        AsyncCounter(10, 13),
    )
    result = [x async for x in chained]
    print(f"체이닝 결과: {result}")

asyncio.run(main())

Python code

import asyncio

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    if name == "fail":
        raise ValueError(f"{name} 실패!")
    return f"{name} 완료 ({delay}s)"

async def main():
    # TaskGroup: 모든 태스크가 완료될 때까지 대기
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch("A", 1))
        task2 = tg.create_task(fetch("B", 0.5))
        task3 = tg.create_task(fetch("C", 1.5))

    # TaskGroup 종료 후 결과 접근
    print(task1.result())
    print(task2.result())
    print(task3.result())

    # 에러 처리: ExceptionGroup
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch("ok", 0.5))
            tg.create_task(fetch("fail", 0.3))
            tg.create_task(fetch("ok2", 0.8))
    except* ValueError as eg:
        for err in eg.exceptions:
            print(f"  에러: {err}")

asyncio.run(main())

Python code

# trio 스타일을 asyncio로 시뮬레이션
import asyncio

async def child_task(name: str, delay: float):
    print(f"  {name} 시작")
    await asyncio.sleep(delay)
    print(f"  {name} 완료")
    return f"{name} 결과"

async def structured_concurrency():
    """trio의 nursery 패턴을 asyncio TaskGroup으로 구현"""
    print("부모 태스크 시작")

    # trio: async with trio.open_nursery() as nursery:
    # asyncio 3.11+:
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(child_task("작업1", 1))
        t2 = tg.create_task(child_task("작업2", 0.5))
        t3 = tg.create_task(child_task("작업3", 1.5))
    # 여기서 모든 자식 태스크가 완료됨을 보장

    print("모든 자식 태스크 완료")
    print(f"  결과: {t1.result()}, {t2.result()}, {t3.result()}")

async def cancellation_example():
    """취소 전파 시연"""
    async def long_task():
        try:
            await asyncio.sleep(100)
        except asyncio.CancelledError:
            print("  태스크 취소됨, 정리 중...")
            raise

    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("취소 확인됨")

asyncio.run(structured_concurrency())
asyncio.run(cancellation_example())

Python code

import asyncio

# 테스트할 비동기 코드
async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # DB 조회 시뮬레이션
    if user_id <= 0:
        raise ValueError("유효하지 않은 ID")
    return {"id": user_id, "name": f"User-{user_id}"}

async def get_users(ids: list[int]) -> list[dict]:
    tasks = [fetch_user(uid) for uid in ids]
    return await asyncio.gather(*tasks)

# pytest-asyncio 패턴 시뮬레이션
async def test_fetch_user():
    user = await fetch_user(1)
    assert user["id"] == 1
    assert user["name"] == "User-1"
    print("test_fetch_user 통과")

async def test_fetch_user_invalid():
    try:
        await fetch_user(-1)
        assert False, "예외가 발생해야 함"
    except ValueError as e:
        assert "유효하지 않은 ID" in str(e)
        print("test_fetch_user_invalid 통과")

async def test_get_users():
    users = await get_users([1, 2, 3])
    assert len(users) == 3
    assert all(u["name"].startswith("User-") for u in users)
    print("test_get_users 통과")

async def run_tests():
    await test_fetch_user()
    await test_fetch_user_invalid()
    await test_get_users()
    print("모든 비동기 테스트 통과!")

asyncio.run(run_tests())

Python code

import asyncio
import traceback

# 디버그 모드 활성화
# PYTHONASYNCIODEBUG=1 python script.py

async def buggy_task():
    await asyncio.sleep(0.1)
    return 42

async def main():
    # 1. await 빠뜨린 코루틴 감지
    # buggy_task()  # RuntimeWarning: coroutine was never awaited
    result = await buggy_task()
    print(f"결과: {result}")

    # 2. 태스크 모니터링
    async def monitored_task(name, delay):
        await asyncio.sleep(delay)
        return f"{name} 완료"

    tasks = [
        asyncio.create_task(monitored_task("A", 0.5), name="Task-A"),
        asyncio.create_task(monitored_task("B", 1.0), name="Task-B"),
    ]

    # 현재 실행 중인 태스크 확인
    all_tasks = asyncio.all_tasks()
    for t in all_tasks:
        print(f"  태스크: {t.get_name()}, done={t.done()}")

    await asyncio.gather(*tasks)

    # 3. 타임아웃으로 교착 감지
    async def maybe_deadlock():
        await asyncio.sleep(10)

    try:
        await asyncio.wait_for(maybe_deadlock(), timeout=1.0)
    except asyncio.TimeoutError:
        print("교착 상태 가능성 감지 (타임아웃)")

    # 4. 예외 추적 개선
    async def failing():
        raise RuntimeError("비동기 에러")

    task = asyncio.create_task(failing())
    try:
        await task
    except RuntimeError:
        print(f"스택 트레이스:\n{traceback.format_exc()[:200]}")

asyncio.run(main())

Python code

import asyncio

async def fetch_chunk(chunk_id: int, data: list) -> dict:
    """개별 청크 처리 (fan-out)"""
    await asyncio.sleep(0.3)
    result = sum(data)
    print(f"  청크 {chunk_id}: 합계={result}")
    return {"chunk": chunk_id, "sum": result}

def split_data(data: list, n_chunks: int) -> list[list]:
    """데이터를 n개 청크로 분할"""
    size = len(data) // n_chunks
    return [data[i*size:(i+1)*size] for i in range(n_chunks)]

async def parallel_sum(data: list, n_workers: int = 4) -> int:
    """fan-out/fan-in 패턴"""
    # Fan-out: 작업 분산
    chunks = split_data(data, n_workers)
    tasks = [
        fetch_chunk(i, chunk)
        for i, chunk in enumerate(chunks)
    ]

    # Fan-in: 결과 수집
    results = await asyncio.gather(*tasks)
    total = sum(r["sum"] for r in results)
    return total

async def pipeline_pattern():
    """비동기 파이프라인"""
    q1 = asyncio.Queue()
    q2 = asyncio.Queue()

    async def stage1():
        for i in range(5):
            await q1.put(i * 10)
        await q1.put(None)

    async def stage2():
        while (item := await q1.get()) is not None:
            await q2.put(item + 1)
        await q2.put(None)

    async def stage3():
        results = []
        while (item := await q2.get()) is not None:
            results.append(item * 2)
        print(f"파이프라인 결과: {results}")

    await asyncio.gather(stage1(), stage2(), stage3())

async def main():
    data = list(range(100))
    total = await parallel_sum(data)
    print(f"병렬 합계: {total}")  # 4950

    await pipeline_pattern()

asyncio.run(main())

Python code

import asyncio
from concurrent.futures import ThreadPoolExecutor
import time

# 동기 함수 (변경 불가한 라이브러리)
def sync_heavy_io(name: str, seconds: float) -> str:
    time.sleep(seconds)  # 블로킹 I/O
    return f"{name} 완료"

def sync_cpu_bound(n: int) -> int:
    return sum(i * i for i in range(n))

# 전략 1: run_in_executor (I/O 바운드)
async def async_io(name: str, seconds: float) -> str:
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(
        None, sync_heavy_io, name, seconds
    )

# 전략 2: ProcessPoolExecutor (CPU 바운드)
async def async_cpu(n: int) -> int:
    loop = asyncio.get_running_loop()
    from concurrent.futures import ProcessPoolExecutor
    with ProcessPoolExecutor() as pool:
        return await loop.run_in_executor(pool, sync_cpu_bound, n)

# 전략 3: asyncio.to_thread (Python 3.9+)
async def main():
    # I/O 바운드 병렬화
    start = time.perf_counter()
    results = await asyncio.gather(
        asyncio.to_thread(sync_heavy_io, "A", 1),
        asyncio.to_thread(sync_heavy_io, "B", 1),
        asyncio.to_thread(sync_heavy_io, "C", 1),
    )
    elapsed = time.perf_counter() - start
    print(f"결과: {results} ({elapsed:.1f}초)")  # ~1초

asyncio.run(main())