모던 파이썬 고급 #4 비동기 깊이 — 이벤트 루프, gather/wait, async generator

def run_loop(coros):
    queue = list(coros)
    while queue:
        coro = queue.pop(0)
        try:
            # 코루틴을 다음 await 까지 진행
            future = coro.send(None)
        except StopIteration as e:
            continue   # 코루틴 끝남
        # future가 끝나면 코루틴을 큐에 다시 넣음
        future.add_done_callback(lambda _: queue.append(coro))

async def slow():
    total = 0
    for i in range(10_000_000):    # CPU 바운드, await 없음
        total += i
    return total

async def main():
    await asyncio.gather(slow(), slow(), slow())   # 세 개를 동시에?

async def main():
    loop = asyncio.get_running_loop()
    fut = loop.create_future()

    # 1초 후에 결과를 채움
    loop.call_later(1, fut.set_result, "done")

    result = await fut
    print(result)   # done

async def main():
    task = asyncio.create_task(work())   # Task 생성
    print(isinstance(task, asyncio.Future))   # True

task = asyncio.create_task(work())

await asyncio.sleep(0.1)
print(task.done())     # False / True

if task.done():
    if task.exception():
        print("실패:", task.exception())
    else:
        print("성공:", task.result())

task.cancel()           # 취소 시도

async def main():
    asyncio.create_task(background_work())   # 결과 무시
    # ...

async def main():
    background_tasks = set()

    task = asyncio.create_task(background_work())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

    # ...

results = await asyncio.gather(
    fetch("a"), fetch("b"), fetch("c"),
    return_exceptions=False,
)
# results: [a결과, b결과, c결과]  ← 호출 순서대로

results = await asyncio.gather(
    fetch("a"),
    fetch_might_fail("b"),
    return_exceptions=True,
)
for r in results:
    if isinstance(r, Exception):
        print("실패:", r)
    else:
        print("성공:", r)

done, pending = await asyncio.wait(
    [task1, task2, task3],
    return_when=asyncio.FIRST_COMPLETED,
)

async def main():
    coros = [fetch(u) for u in urls]
    for coro in asyncio.as_completed(coros):
        result = await coro
        print("받음:", result)

import asyncio
import requests   # 동기 라이브러리

async def fetch(url):
    response = await asyncio.to_thread(requests.get, url)
    return response.text

from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_function, arg)

async def stream_chunks(url):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as resp:
            async for chunk in resp.aiter_bytes():
                yield chunk

async def main():
    async for chunk in stream_chunks("https://example.com"):
        print(len(chunk))

class Counter:
    def __init__(self, limit):
        self.limit = limit
        self.n = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.n >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)
        self.n += 1
        return self.n

async def main():
    async for i in Counter(3):
        print(i)
# 0.1초 간격으로 1, 2, 3

from contextlib import asynccontextmanager

@asynccontextmanager
async def open_db():
    conn = await connect_async()
    try:
        yield conn
    finally:
        await conn.close()

async def main():
    async with open_db() as conn:
        await conn.execute(...)

sem = asyncio.Semaphore(10)

async def fetch_one(url):
    async with sem:
        return await fetch(url)

async def fetch_all(urls):
    return await asyncio.gather(*[fetch_one(u) for u in urls])

queue = asyncio.Queue(maxsize=10)

async def producer():
    for i in range(100):
        await queue.put(i)
    await queue.put(None)    # 종료 신호

async def consumer():
    while True:
        item = await queue.get()
        if item is None:
            break
        process(item)
        queue.task_done()

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer())
        tg.create_task(consumer())

async def main():
    tasks = [
        asyncio.create_task(fetch_a()),
        asyncio.create_task(fetch_b()),
    ]
    done, pending = await asyncio.wait(
        tasks, return_when=asyncio.FIRST_COMPLETED
    )
    for t in pending:
        t.cancel()
    return done.pop().result()

async def main():
    try:
        async with asyncio.timeout(5):
            data = await slow_op()
    except TimeoutError:
        print("5초 초과")

async def work():
    try:
        await long_operation()
    except asyncio.CancelledError:
        await cleanup()    # 정리는 비동기로도 가능
        raise              # ← 다시 던져야 함!

async def critical():
    return await asyncio.shield(important_work())

PYTHONASYNCIODEBUG=1 python app.py