Python Academy · 강의

비동기 컨텍스트 관리자와 반복자

async with와 async for 프로토콜을 구현합니다.

비동기 컨텍스트 관리자와 반복자은(는) CoddyKit의 무료 Python Academy 강의입니다. 이것은 4개 중 3번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 Python Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. Python Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

async with 문

설정 및 해제 과정에서 입출력 대기가 필요한 컨텍스트 관리자에는 async with를 사용하세요. 관리되는 객체는 __aenter__와 __aexit__을 구현해야 합니다.

import asyncio

class AsyncFile:
    async def __aenter__(self):
        print("open")
        return self
    async def __aexit__(self, *args):
        print("close")

async def main():
    async with AsyncFile() as f:
        print("using")

asyncio.run(main())

aenter 및 aexit

둘 다 코루틴입니다. 진입할 때 __aenter__를 기다리고, 예외가 발생했더라도 종료할 때 __aexit__를 기다립니다.

import asyncio

class DBSession:
    async def __aenter__(self):
        self.conn = await async_connect()
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            await self.conn.rollback()
        else:
            await self.conn.commit()
        await self.conn.close()
        return False

@asynccontextmanager

contextlib.asynccontextmanager를 사용하면 비동기 제너레이터 함수로 비동기 컨텍스트 관리자를 작성할 수 있습니다.

from contextlib import asynccontextmanager
import asyncio

@asynccontextmanager
async def managed_resource():
    print("acquire")
    try:
        yield {"status": "ready"}
    finally:
        print("release")

async def main():
    async with managed_resource() as r:
        print(r["status"])

asyncio.run(main())

async for 문

async for는 비동기 반복 가능 객체를 순회합니다. 이 객체는 __aiter__와 __anext__을 구현하며, 각 반복에서 입출력이 발생할 수 있습니다.

import asyncio

class AsyncCounter:
    def __init__(self, n): self.n, self.i = n, 0
    def __aiter__(self): return self
    async def __anext__(self):
        if self.i >= self.n: raise StopAsyncIteration
        await asyncio.sleep(0)
        self.i += 1
        return self.i

async def main():
    async for val in AsyncCounter(3):
        print(val)

asyncio.run(main())

비동기 제너레이터

yield를 포함하는 async def 함수는 비동기 제너레이터입니다. 이를 소비하려면 async for를 사용하세요.

import asyncio

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

async def main():
    async for val in ticker(3):
        print(val)   # 0, 1, 2 with 1-s pauses

asyncio.run(main())

비동기 파일 입출력을 위한 aiofiles

aiofiles 라이브러리는 파일 작업을 비동기 컨텍스트 관리자와 비동기 반복자로 감쌉니다.

# pip install aiofiles
import aiofiles, asyncio

async def main():
    async with aiofiles.open("data.txt", "r") as f:
        async for line in f:
            print(line.strip())

aiohttp 클라이언트 세션

aiohttp.ClientSession은 동시에 여러 HTTP 요청을 보내기 위한 비동기 컨텍스트 관리자입니다.

# pip install aiohttp
import aiohttp, asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

asyncio.run(fetch("https://api.example.com/data"))

여러 비동기 컨텍스트 관리자

한 줄에 여러 async with 절을 나열할 수 있습니다(Python 3.10 이상에서 괄호로 묶는 형식).

import asyncio

async def main():
    async with (
        open_db() as db,
        open_cache() as cache,
    ):
        data = await db.fetch("SELECT 1")
        await cache.set("key", data)

AsyncExitStack

contextlib.AsyncExitStack은 동적으로 구성되는 비동기 컨텍스트 관리자 집합을 관리합니다.

from contextlib import AsyncExitStack
import asyncio

async def main():
    async with AsyncExitStack() as stack:
        conn1 = await stack.enter_async_context(connect("db1"))
        conn2 = await stack.enter_async_context(connect("db2"))
        await process(conn1, conn2)

pytest-asyncio로 비동기 코드 테스트하기

pytest-asyncio를 사용하면 @pytest.mark.asyncio 표시가 있는 비동기 테스트 함수를 작성할 수 있습니다.

# pip install pytest-asyncio
import pytest, asyncio

async def fetch(): return 42

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch()
    assert result == 42

비동기 반복자 프로토콜 요약

객체에 __aiter__가 있으면 비동기 반복 가능 객체입니다. 여기에 __anext__도 있으면 비동기 반복자입니다. 비동기 제너레이터는 두 가지를 모두 자동으로 구현합니다.

import asyncio

async def gen():
    yield 1
    yield 2

async def main():
    g = gen()
    print(await g.__anext__())   # 1
    print(await g.__anext__())   # 2

빠른 확인

async with와 함께 사용하려면 객체가 어떤 던더 메서드를 구현해야 합니까?

복습

비동기 설정 및 해제가 필요한 리소스에는 async with를 사용하세요. 비동기적으로 값을 생성하는 반복 가능 객체에는 async for를 사용하세요. 비동기 제너레이터를 사용하면 비동기 반복자를 간단하게 만들 수 있습니다. aiofiles와 aiohttp 같은 라이브러리는 비동기 방식의 파일 및 HTTP 작업을 제공합니다.

자주 묻는 질문