aiohttp

1️⃣ aiohttp 기초

aiohttp는 Python에서 비동기 HTTP 클라이언트 및 서버를 구현하기 위한 라이브러리이다. asyncio를 기반으로 하며, 고성능 비동기 요청 처리를 가능하게 한다.

주요 특징:

비동기 HTTP 클라이언트 및 서버 지원
WebSocket 지원
미들웨어 기능 제공
플러그인 시스템
신호 처리 메커니즘
라우팅 시스템

설치:

pip install aiohttp

기본 서버 예제:

import asyncio
from aiohttp import web

async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = f"안녕하세요, {name}님!"
    return web.Response(text=text)

async def main():
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])
    
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    
    print("서버 시작: http://localhost:8080")
    # 서버가 계속 실행되도록 유지
    await asyncio.Event().wait()

if __name__ == '__main__':
    asyncio.run(main())

라우팅 시스템:

aiohttp는 URL 라우팅을 위한 다양한 방법을 제공한다.

app.add_routes([
    web.get('/users', get_users),
    web.post('/users', create_user),
    web.get('/users/{id}', get_user),
    web.put('/users/{id}', update_user),
    web.delete('/users/{id}', delete_user),
])

# 또는 데코레이터 사용
routes = web.RouteTableDef()

@routes.get('/')
async def hello(request):
    return web.Response(text="Hello, world")

@routes.view('/users/{id}')
class UserView(web.View):
    async def get(self):
        return web.Response(text=f"사용자 정보 조회")
    
    async def put(self):
        return web.Response(text=f"사용자 정보 업데이트")

app.add_routes(routes)

요청 및 응답 처리:

aiohttp는 다양한 형식의 요청과 응답을 처리할 수 있다.

async def handle_json(request):
    # JSON 요청 처리
    data = await request.json()
    
    # JSON 응답 반환
    return web.json_response({
        'status': 'success',
        'data': data
    })

async def handle_form(request):
    # 폼 데이터 처리
    data = await request.post()
    name = data.get('name')
    
    # 템플릿 응답
    return web.Response(
        text=f"<html><body><h1>안녕하세요, {name}님!</h1></body></html>",
        content_type='text/html'
    )

async def handle_file(request):
    # 파일 응답
    return web.FileResponse('/path/to/file.pdf')

✅ 특징:

비동기 프로그래밍 패턴
유연한 라우팅 시스템
다양한 응답 형식 지원
미들웨어 체인 구성

2️⃣ aiohttp 클라이언트

aiohttp는 비동기 HTTP 클라이언트 기능을 제공하여 여러 웹 요청을 동시에 처리할 수 있게 한다.

기본 사용법:

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                return await response.text()
            return None

async def main():
    result = await fetch('https://python.org')
    print(f"결과 길이: {len(result) if result else 'N/A'}")

asyncio.run(main())

다양한 HTTP 메서드:

aiohttp 클라이언트는 모든 표준 HTTP 메서드를 지원한다.

async def main():
    async with aiohttp.ClientSession() as session:
        # GET 요청
        async with session.get('https://api.example.com/users') as resp:
            users = await resp.json()
        
        # POST 요청
        user_data = {'name': '홍길동', 'email': '[email protected]'}
        async with session.post('https://api.example.com/users', json=user_data) as resp:
            result = await resp.json()
        
        # PUT 요청
        update_data = {'name': '김길동'}
        async with session.put('https://api.example.com/users/1', json=update_data) as resp:
            updated = await resp.json()
        
        # DELETE 요청
        async with session.delete('https://api.example.com/users/1') as resp:
            if resp.status == 204:
                print("사용자가 삭제되었습니다")

동시 요청 처리:

여러 요청을 동시에 처리하는 것은 aiohttp의 핵심 기능이다.

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(fetch_url(session, url))
        results = await asyncio.gather(*tasks)
        return results

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'https://python.org',
        'https://github.com',
        'https://stackoverflow.com'
    ]
    results = await fetch_all(urls)
    for url, html in zip(urls, results):
        print(f"{url}: {len(html)} 바이트")

asyncio.run(main())

고급 클라이언트 기능:

1. 요청 타임아웃 설정:

async with session.get('https://slow-api.example.com', timeout=aiohttp.ClientTimeout(total=5)) as resp:
    # 5초 이내에 응답이 오지 않으면 예외 발생
    data = await resp.json()

2. 쿠키 및 세션 관리:

# 쿠키 저장 및 전송
jar = aiohttp.CookieJar()
async with aiohttp.ClientSession(cookie_jar=jar) as session:
    await session.get('https://auth.example.com/login')
    # 이제 jar에 인증 쿠키가 저장됨
    
    # 동일한 세션에서 요청하면 쿠키가 자동으로 전송됨
    await session.get('https://auth.example.com/profile')

3. 요청 헤더 설정:

headers = {
    'User-Agent': 'My Custom User Agent',
    'Authorization': 'Bearer token123',
    'Content-Type': 'application/json'
}

async with session.get('https://api.example.com', headers=headers) as resp:
    data = await resp.json()

4. 프록시 설정:

async with session.get('https://example.com', proxy='http://proxy.example.com:8080') as resp:
    data = await resp.text()

5. SSL 인증서 검증:

# SSL 검증 비활성화 (보안 위험, 개발 환경에서만 사용)
ssl_context = False

# 또는 사용자 정의 인증서 사용
import ssl
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain('client.crt', 'client.key')

async with session.get('https://secure.example.com', ssl=ssl_context) as resp:
    data = await resp.text()

✅ 특징:

세션 관리 및 재사용
비동기 요청 병렬 처리
타임아웃 및 재시도 설정
요청/응답 인터셉터
다양한 인증 방식 지원

3️⃣ 미들웨어와 핸들러

aiohttp는 요청 처리 전후에 공통 로직을 실행할 수 있는 미들웨어를 지원한다.

@web.middleware
async def error_middleware(request, handler):
    try:
        response = await handler(request)
        return response
    except web.HTTPException as ex:
        return web.json_response(
            {'error': str(ex)},
            status=ex.status
        )
    except Exception as ex:
        return web.json_response(
            {'error': 'Internal Server Error'},
            status=500
        )

async def handle_404(request):
    return web.json_response(
        {'error': 'Not Found'},
        status=404
    )

app = web.Application(middlewares=[error_middleware])
app.router.add_routes([
    web.get('/', hello),
    web.get('/users/{id}', get_user)
])
app.router.set_error_handler(404, handle_404)

✅ 특징:

전역 에러 처리
커스텀 미들웨어 지원
라우팅 에러 처리
요청/응답 변환
인증 및 권한 제어

4️⃣ 데이터베이스 통합

aiohttp 애플리케이션에서 비동기 데이터베이스 접근을 구현하는 방법이다.

import aiopg
from aiohttp import web

async def init_pg(app):
    conf = app['config']['postgres']
    engine = await aiopg.create_pool(
        database=conf['database'],
        user=conf['user'],
        password=conf['password'],
        host=conf['host'],
        port=conf['port'],
        minsize=1,
        maxsize=5
    )
    app['db'] = engine

async def close_pg(app):
    app['db'].close()
    await app['db'].wait_closed()

class UserHandler:
    async def get_user(self, request):
        user_id = request.match_info['id']
        async with request.app['db'].acquire() as conn:
            async with conn.cursor() as cur:
                await cur.execute(
                    'SELECT * FROM users WHERE id = %s',
                    (user_id,)
                )
                user = await cur.fetchone()
                if not user:
                    raise web.HTTPNotFound()
                return web.json_response({
                    'id': user[0],
                    'name': user[1],
                    'email': user[2]
                })

app = web.Application()
app.on_startup.append(init_pg)
app.on_cleanup.append(close_pg)

✅ 특징:

비동기 데이터베이스 연결
커넥션 풀링
애플리케이션 생명주기 관리
트랜잭션 처리
매개변수화된 쿼리

5️⃣ WebSocket 구현

aiohttp는 클라이언트와 서버 측 모두에서 WebSocket 연결을 지원하여 실시간 양방향 통신을 가능하게 한다.

서버 측 WebSocket:

from aiohttp import web
import json

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    # 클라이언트에게 환영 메시지 전송
    await ws.send_str(json.dumps({'action': 'connect', 'status': 'success'}))
    
    try:
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                data = json.loads(msg.data)
                print(f"메시지 수신: {data}")
                
                # 에코 응답
                await ws.send_str(json.dumps({
                    'action': 'echo',
                    'data': data
                }))
                
            elif msg.type == web.WSMsgType.ERROR:
                print(f"WebSocket 오류: {ws.exception()}")
    finally:
        # 연결 종료 시 정리 작업
        print("WebSocket 연결 종료")
    
    return ws

app = web.Application()
app.add_routes([web.get('/ws', websocket_handler)])

if __name__ == '__main__':
    web.run_app(app)

클라이언트 측 WebSocket:

import aiohttp
import asyncio
import json

async def websocket_client():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect('ws://localhost:8080/ws') as ws:
            # 서버로부터 환영 메시지 수신
            async for msg in ws:
                if msg.type == aiohttp.WSMsgType.TEXT:
                    data = json.loads(msg.data)
                    print(f"서버로부터 수신: {data}")
                    
                    # 메시지 전송
                    await ws.send_str(json.dumps({
                        'message': '안녕하세요, WebSocket!',
                        'timestamp': asyncio.get_event_loop().time()
                    }))
                    
                elif msg.type == aiohttp.WSMsgType.CLOSED:
                    break
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    break

asyncio.run(websocket_client())

실시간 채팅 애플리케이션 예제:

# server.py
from aiohttp import web
import json

# 연결된 모든 클라이언트를 저장
connected_clients = []

async def chat_websocket(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    # 새 클라이언트 추가
    connected_clients.append(ws)
    client_id = len(connected_clients)
    
    # 모든 클라이언트에게 새 사용자 접속 알림
    for client in connected_clients:
        if client != ws:
            await client.send_str(json.dumps({
                'action': 'user_joined',
                'user_id': client_id
            }))
    
    try:
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                data = json.loads(msg.data)
                
                # 채팅 메시지를 모든 클라이언트에게 브로드캐스트
                for client in connected_clients:
                    await client.send_str(json.dumps({
                        'action': 'chat',
                        'user_id': client_id,
                        'message': data.get('message', '')
                    }))
                    
            elif msg.type == web.WSMsgType.ERROR:
                print(f"WebSocket 오류: {ws.exception()}")
    finally:
        # 연결 종료 시 클라이언트 제거
        connected_clients.remove(ws)
        # 사용자 퇴장 알림
        for client in connected_clients:
            await client.send_str(json.dumps({
                'action': 'user_left',
                'user_id': client_id
            }))
    
    return ws

app = web.Application()
app.add_routes([web.get('/chat', chat_websocket)])

if __name__ == '__main__':
    web.run_app(app)

WebSocket 이벤트 관리:

WebSocket 연결의 다양한 이벤트를 처리할 수 있다.

async def websocket_handler(request):
    ws = web.WebSocketResponse()
    await ws.prepare(request)
    
    # 연결 시작 이벤트
    print("WebSocket 연결 시작")
    
    try:
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                # 텍스트 메시지 이벤트
                print(f"텍스트 수신: {msg.data}")
                await ws.send_str(f"에코: {msg.data}")
                
            elif msg.type == web.WSMsgType.BINARY:
                # 바이너리 메시지 이벤트
                print(f"바이너리 데이터 수신: {len(msg.data)} 바이트")
                await ws.send_bytes(msg.data)
                
            elif msg.type == web.WSMsgType.PING:
                # Ping 이벤트
                await ws.pong()
                
            elif msg.type == web.WSMsgType.CLOSE:
                # 클라이언트가 연결 종료 요청
                await ws.close()
                
            elif msg.type == web.WSMsgType.ERROR:
                # 오류 이벤트
                print(f"오류 발생: {ws.exception()}")
    finally:
        # 연결 종료 이벤트
        print("WebSocket 연결 종료")
    
    return ws

✅ 특징:

양방향 실시간 통신
이벤트 기반 메시지 교환
JSON 직렬화된 메시지
클라이언트 연결 관리
브로드캐스팅 지원

6️⃣ 실용적인 예제

aiohttp를 사용한 실제 애플리케이션 구현 예제를 살펴보자.

REST API 서비스:

from aiohttp import web
import json

class RestAPI:
    def __init__(self):
        self.items = {}
    
    async def get_items(self, request):
        return web.json_response(list(self.items.values()))
    
    async def get_item(self, request):
        item_id = request.match_info['id']
        if item_id not in self.items:
            raise web.HTTPNotFound()
        return web.json_response(self.items[item_id])
    
    async def create_item(self, request):
        data = await request.json()
        item_id = str(len(self.items) + 1)
        self.items[item_id] = {
            'id': item_id,
            **data
        }
        return web.json_response(
            self.items[item_id],
            status=201
        )
    
    async def update_item(self, request):
        item_id = request.match_info['id']
        if item_id not in self.items:
            raise web.HTTPNotFound()
        
        data = await request.json()
        self.items[item_id].update(data)
        return web.json_response(self.items[item_id])

api = RestAPI()
app.router.add_get('/items', api.get_items)
app.router.add_get('/items/{id}', api.get_item)
app.router.add_post('/items', api.create_item)
app.router.add_put('/items/{id}', api.update_item)

파일 업로드 처리:

import os

async def handle_upload(request):
    reader = await request.multipart()
    
    # 파일 필드 처리
    field = await reader.next()
    filename = field.filename
    
    # 파일 저장
    size = 0
    with open(os.path.join('uploads', filename), 'wb') as f:
        while True:
            chunk = await field.read_chunk()
            if not chunk:
                break
            size += len(chunk)
            f.write(chunk)
    
    return web.json_response({
        'filename': filename,
        'size': size
    })

app.router.add_post('/upload', handle_upload)

세션 관리:

import aiohttp_session
from aiohttp_session import setup, get_session
from aiohttp_session.cookie_storage import EncryptedCookieStorage
import base64

async def init_app():
    app = web.Application()
    
    # 세션 설정
    secret_key = base64.urlsafe_b64decode(
        'your-secret-key=='
    )
    setup(app, EncryptedCookieStorage(secret_key))
    
    return app

async def login(request):
    session = await get_session(request)
    data = await request.json()
    
    # 인증 로직
    user_id = authenticate(data)
    if user_id:
        session['user_id'] = user_id
        return web.json_response({'status': 'logged in'})
    
    raise web.HTTPUnauthorized()

async def logout(request):
    session = await get_session(request)
    session.invalidate()
    return web.json_response({'status': 'logged out'})

✅ 특징:

비동기 코드 최적화
연결 풀링 활용
에러 처리 구현
세션 관리
보안 설정
로깅 구현
테스트 작성
성능 모니터링

7️⃣ aiohttp 애플리케이션 테스트

aiohttp는 애플리케이션 테스트를 위한 강력한 도구를 제공한다.

테스트 클라이언트:

aiohttp는 내장된 테스트 클라이언트를 통해 애플리케이션에 요청을 보내고 응답을 검증할 수 있다.

import pytest
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer

async def hello(request):
    return web.Response(text="Hello, World!")

async def test_hello(aiohttp_client):
    # 테스트할 애플리케이션 생성
    app = web.Application()
    app.router.add_get('/', hello)
    
    # 테스트 클라이언트 생성
    client = await aiohttp_client(app)
    
    # 요청 전송 및 응답 검증
    resp = await client.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert text == "Hello, World!"

pytest 픽스처 사용하기:

pytest-aiohttp 패키지를 설치하면 테스트에 사용할 수 있는 유용한 픽스처가 제공된다.

# pytest-aiohttp 패키지 설치:
# pip install pytest-aiohttp

# conftest.py
import pytest
from aiohttp import web

# 테스트용 애플리케이션 생성
@pytest.fixture
def app():
    app = web.Application()
    
    async def hello(request):
        return web.Response(text="Hello, World!")
    
    async def echo(request):
        data = await request.json()
        return web.json_response(data)
    
    app.router.add_get('/', hello)
    app.router.add_post('/echo', echo)
    
    return app

# test_app.py
async def test_hello(aiohttp_client, app):
    client = await aiohttp_client(app)
    resp = await client.get('/')
    assert resp.status == 200
    text = await resp.text()
    assert text == "Hello, World!"

async def test_echo(aiohttp_client, app):
    client = await aiohttp_client(app)
    data = {"message": "Hello"}
    resp = await client.post('/echo', json=data)
    assert resp.status == 200
    json_data = await resp.json()
    assert json_data == data

모킹과 의존성 주입:

외부 의존성이 있는 애플리케이션을 테스트하기 위해 모킹 기법을 사용한다.

import pytest
from unittest.mock import MagicMock, patch
from aiohttp import web

# 데이터베이스 접근 클래스 (실제 코드)
class Database:
    async def get_user(self, user_id):
        # 실제로는 데이터베이스에서 사용자 조회
        pass

# 애플리케이션 핸들러
async def get_user_handler(request):
    user_id = request.match_info['id']
    db = request.app['db']
    
    user = await db.get_user(user_id)
    if user:
        return web.json_response(user)
    return web.Response(status=404)

# 테스트 함수
async def test_get_user(aiohttp_client):
    # 데이터베이스 객체 모킹
    mock_db = MagicMock()
    mock_db.get_user = MagicMock(return_value={"id": "1", "name": "테스트 사용자"})
    
    # 애플리케이션 생성
    app = web.Application()
    app['db'] = mock_db
    app.router.add_get('/users/{id}', get_user_handler)
    
    # 테스트 클라이언트 생성
    client = await aiohttp_client(app)
    
    # 요청 전송 및 응답 검증
    resp = await client.get('/users/1')
    assert resp.status == 200
    
    data = await resp.json()
    assert data['id'] == "1"
    assert data['name'] == "테스트 사용자"
    
    # 모킹된 메서드가 호출되었는지 확인
    mock_db.get_user.assert_called_once_with("1")

WebSocket 테스트:

WebSocket 기능을 테스트하기 위한 방법도 제공된다.

async def test_websocket(aiohttp_client):
    # WebSocket 핸들러
    async def websocket_handler(request):
        ws = web.WebSocketResponse()
        await ws.prepare(request)
        
        async for msg in ws:
            if msg.type == web.WSMsgType.TEXT:
                await ws.send_str(f"echo: {msg.data}")
        
        return ws
    
    # 애플리케이션 설정
    app = web.Application()
    app.router.add_get('/ws', websocket_handler)
    
    # 테스트 클라이언트 생성
    client = await aiohttp_client(app)
    
    # WebSocket 연결
    ws = await client.ws_connect('/ws')
    
    # 메시지 전송 및 응답 확인
    await ws.send_str("hello")
    msg = await ws.receive()
    assert msg.data == "echo: hello"
    
    # 연결 종료
    await ws.close()

통합 테스트 작성:

전체 애플리케이션 흐름을 테스트하기 위한 통합 테스트를 작성할 수 있다.

import jwt
import pytest
from aiohttp import web

# 보안 키
SECRET_KEY = "test_secret_key"

# 사용자 인증 미들웨어
@web.middleware
async def auth_middleware(request, handler):
    if request.path == '/login':
        return await handler(request)
    
    token = request.headers.get('Authorization', '').replace('Bearer ', '')
    if not token:
        return web.json_response({"error": "인증 필요"}, status=401)
    
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        request['user'] = payload
    except Exception:
        return web.json_response({"error": "유효하지 않은 토큰"}, status=401)
    
    return await handler(request)

# 로그인 핸들러
async def login_handler(request):
    data = await request.json()
    if data.get('username') == 'admin' and data.get('password') == 'password':
        token = jwt.encode({"user_id": 1, "username": "admin"}, SECRET_KEY, algorithm='HS256')
        return web.json_response({"token": token})
    return web.json_response({"error": "잘못된 자격 증명"}, status=401)

# 보호된 리소스 핸들러
async def protected_handler(request):
    user = request.get('user')
    return web.json_response({"message": f"안녕하세요, {user['username']}님!"})

# 테스트 함수
async def test_auth_flow(aiohttp_client):
    # 애플리케이션 설정
    app = web.Application(middlewares=[auth_middleware])
    app.router.add_post('/login', login_handler)
    app.router.add_get('/protected', protected_handler)
    
    # 테스트 클라이언트 생성
    client = await aiohttp_client(app)
    
    # 로그인 없이 보호된 리소스 접근 시도
    resp = await client.get('/protected')
    assert resp.status == 401
    
    # 잘못된 자격 증명으로 로그인
    resp = await client.post('/login', json={"username": "admin", "password": "wrong"})
    assert resp.status == 401
    
    # 올바른 자격 증명으로 로그인
    resp = await client.post('/login', json={"username": "admin", "password": "password"})
    assert resp.status == 200
    data = await resp.json()
    assert "token" in data
    
    # 토큰으로 보호된 리소스 접근
    token = data['token']
    headers = {"Authorization": f"Bearer {token}"}
    resp = await client.get('/protected', headers=headers)
    assert resp.status == 200
    
    data = await resp.json()
    assert data["message"] == "안녕하세요, admin님!"

비동기 테스트 팁:

aiohttp 테스트 시 비동기 코드를 효과적으로 처리하기 위한 몇 가지 팁이다.

# 여러 비동기 작업을 동시에 실행
async def test_concurrent_requests(aiohttp_client, app):
    client = await aiohttp_client(app)
    
    # 동시에 여러 요청 보내기
    import asyncio
    responses = await asyncio.gather(
        client.get('/api/1'),
        client.get('/api/2'),
        client.get('/api/3')
    )
    
    # 각 응답 검증
    for resp in responses:
        assert resp.status == 200

# 시간 제약 있는 테스트
async def test_with_timeout(aiohttp_client, app):
    client = await aiohttp_client(app)
    
    # 5초 타임아웃으로 요청 실행
    import asyncio
    try:
        async with asyncio.timeout(5):
            resp = await client.get('/slow-api')
            assert resp.status == 200
    except asyncio.TimeoutError:
        pytest.fail("API 응답이 너무 느립니다")

✅ 특징:

애플리케이션 단위 테스트
서버 엔드포인트 검증
비동기 작업 모킹
타임아웃 처리
병렬 요청 테스트
인증 워크플로우 테스트

8️⃣ aiohttp 애플리케이션 배포

aiohttp 애플리케이션을 프로덕션 환경에 배포하기 위한 다양한 방법과 모범 사례를 알아보자.

gunicorn과 함께 사용하기:

aiohttp 애플리케이션은 gunicorn과 함께 배포하는 것이 일반적이다.

# app.py - 애플리케이션 진입점
from aiohttp import web

async def hello(request):
    return web.Response(text="Hello, World!")

async def init_app():
    app = web.Application()
    app.router.add_get('/', hello)
    return app

# gunicorn 워커를 위한 애플리케이션 팩토리 함수
def create_app():
    return init_app()

gunicorn 설치 및 실행:

pip install gunicorn
pip install uvloop  # 선택적으로 성능 향상을 위해

# aiohttp 워커 클래스를 사용하여 실행
gunicorn app:create_app --bind 0.0.0.0:8080 --worker-class aiohttp.GunicornWebWorker

Docker와 함께 배포하기:

Docker를 사용하여 aiohttp 애플리케이션을 컨테이너화하는 방법이다.

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 애플리케이션 코드 복사
COPY . .

# 포트 노출
EXPOSE 8080

# 애플리케이션 실행
CMD ["gunicorn", "app:create_app", "--bind", "0.0.0.0:8080", "--worker-class", "aiohttp.GunicornWebWorker", "--workers", "4"]

requirements.txt:

aiohttp==3.8.5
gunicorn==21.2.0
uvloop==0.17.0

Docker 이미지 빌드 및 실행:

docker build -t aiohttp-app .
docker run -p 8080:8080 aiohttp-app

프로덕션 환경 설정:

프로덕션 환경에서 aiohttp 애플리케이션을 실행할 때 고려해야 할 구성이다.

import os
import logging
from aiohttp import web

# 환경 변수에서 구성 가져오기
HOST = os.getenv('HOST', '0.0.0.0')
PORT = int(os.getenv('PORT', 8080))
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')
DEBUG = os.getenv('DEBUG', 'false').lower() == 'true'

# 로깅 설정
logging.basicConfig(
    level=getattr(logging, LOG_LEVEL),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('aiohttp_app')

# 애플리케이션 초기화
async def init_app():
    app = web.Application(debug=DEBUG)
    
    # 각 환경(개발, 테스트, 프로덕션)에 맞는 구성 로드
    app['config'] = {
        'database_url': os.getenv('DATABASE_URL'),
        'redis_url': os.getenv('REDIS_URL'),
        'secret_key': os.getenv('SECRET_KEY', 'insecure-secret'),
    }
    
    # 종속성 설정, 라우트 등록
    # [생략]
    
    return app

# 개발 환경에서 직접 실행
if __name__ == '__main__':
    web.run_app(init_app(), host=HOST, port=PORT)

상태 모니터링 및 관리:

애플리케이션 상태 모니터링을 위한 엔드포인트를 추가한다.

import time
import psutil
import socket
from aiohttp import web

# 시작 시간 기록
start_time = time.time()

async def health_check(request):
    """기본 상태 확인 엔드포인트"""
    return web.json_response({"status": "ok"})

async def readiness_check(request):
    """서비스가 준비되었는지 확인하는 엔드포인트"""
    # 데이터베이스, 캐시 등 외부 종속성 연결 확인
    database_ok = await check_database(request.app)
    cache_ok = await check_cache(request.app)
    
    if database_ok and cache_ok:
        return web.json_response({"status": "ready"})
    
    # 준비되지 않음 응답 (Kubernetes 등이 인식할 수 있음)
    return web.json_response(
        {"status": "not ready", "details": {"database": database_ok, "cache": cache_ok}},
        status=503
    )

async def metrics(request):
    """애플리케이션 메트릭 엔드포인트"""
    process = psutil.Process()
    
    # 다양한 메트릭 수집
    uptime = time.time() - start_time
    cpu_usage = process.cpu_percent()
    memory_info = process.memory_info()
    memory_usage = memory_info.rss / (1024 * 1024)  # MB 단위
    open_connections = len(process.connections())
    
    # 시스템 정보
    hostname = socket.gethostname()
    
    return web.json_response({
        "uptime_seconds": uptime,
        "cpu_usage_percent": cpu_usage,
        "memory_usage_mb": memory_usage,
        "open_connections": open_connections,
        "hostname": hostname
    })

# 애플리케이션에 상태 엔드포인트 등록
app = web.Application()
app.router.add_get('/health', health_check)
app.router.add_get('/ready', readiness_check)
app.router.add_get('/metrics', metrics)

보안 강화:

프로덕션 환경에서의 보안 강화 방법이다.

from aiohttp import web
from cryptography.fernet import Fernet
import secrets
import aiohttp_session
from aiohttp_session.cookie_storage import EncryptedCookieStorage

# 보안 헤더 미들웨어
@web.middleware
async def security_headers_middleware(request, handler):
    response = await handler(request)
    
    # 보안 관련 헤더 추가
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'DENY'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    response.headers['Content-Security-Policy'] = "default-src 'self'"
    
    return response

# 애플리케이션 설정
async def init_app():
    # 보안 미들웨어 적용
    app = web.Application(middlewares=[security_headers_middleware])
    
    # 안전한 세션 설정
    fernet_key = Fernet.generate_key()
    secret_key = base64.urlsafe_b64decode(fernet_key)
    aiohttp_session.setup(app, EncryptedCookieStorage(secret_key))
    
    # 안전한 비밀번호 해싱을 위한 설정
    app['pwd_context'] = CryptContext(
        schemes=["argon2", "bcrypt"],
        deprecated="auto"
    )
    
    # CSRF 토큰 생성 도우미
    app['create_csrf_token'] = lambda: secrets.token_hex(32)
    
    return app

# CSRF 보호 미들웨어
@web.middleware
async def csrf_middleware(request, handler):
    if request.method not in ('GET', 'HEAD', 'OPTIONS', 'TRACE'):
        token = request.headers.get('X-CSRF-Token')
        session = await aiohttp_session.get_session(request)
        
        if token is None or token != session.get('csrf_token'):
            return web.json_response(
                {"error": "CSRF 토큰이 유효하지 않습니다."},
                status=403
            )
    
    return await handler(request)

스케일링 전략:

애플리케이션을 수평적으로 확장하기 위한 전략이다.

# 로드 밸런서 뒤에서 여러 인스턴스를 실행하기 위한 설정

# nginx.conf
server {
    listen 80;
    server_name example.com;

    location / {
        proxy_pass http://aiohttp_servers;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

upstream aiohttp_servers {
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    server 127.0.0.1:8003;
    server 127.0.0.1:8004;
}

무중단 배포:

애플리케이션을 중단 없이 업데이트하는 방법이다.

# systemd 서비스 파일 예시
# /etc/systemd/system/aiohttp-app.service

[Unit]
Description=aiohttp application
After=network.target

[Service]
User=appuser
Group=appuser
WorkingDirectory=/path/to/app
ExecStart=/path/to/venv/bin/gunicorn app:create_app --bind 0.0.0.0:8080 --worker-class aiohttp.GunicornWebWorker
Restart=on-failure
RestartSec=5s

# 그레이스풀 종료 관련 설정
ExecReload=/bin/kill -s HUP $MAINPID
KillMode=mixed
KillSignal=SIGTERM
TimeoutStopSec=30s

[Install]
WantedBy=multi-user.target

클라우드 환경에서의 배포:

클라우드 환경에서 aiohttp 애플리케이션을 배포하기 위한 설정이다.

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: aiohttp-app
spec:
  replicas: 3
  selector:
    matchLabels:
      app: aiohttp-app
  template:
    metadata:
      labels:
        app: aiohttp-app
    spec:
      containers:
      - name: aiohttp-app
        image: your-registry/aiohttp-app:latest
        ports:
        - containerPort: 8080
        env:
        - name: PORT
          value: "8080"
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: app-secrets
              key: database-url
        # 상태 확인 프로브
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "256Mi"
            cpu: "500m"

✅ 특징:

그레이스풀 종료 처리
상태 모니터링 엔드포인트
컨테이너화 지원
수평 확장 전략
환경 기반 구성
보안 모범 사례
클라우드 네이티브 배포

주요 팁

✅ 모범 사례:

세션 재사용: 여러 요청에 동일한 ClientSession 인스턴스를 재사용하여 성능 최적화
비동기 컨텍스트 관리자: 리소스 관리를 위해 async with 구문 활용
타임아웃 설정: 모든 외부 서비스 호출에 적절한 타임아웃 설정
적절한 상태 코드: 의미 있는 HTTP 상태 코드와 응답 본문 반환
에러 처리: 모든 예외를 잡아 적절히 처리하는 미들웨어 구현
연결 풀링: 데이터베이스 및 외부 서비스 연결 풀링
청크 처리: 대용량 데이터는 청크 단위로 스트리밍
경로 정규화: 명확한 경로 구조와 라우팅 패턴 설계
컨텍스트 활용: request['app']을 통한 애플리케이션 컨텍스트 활용
로깅 전략: 구조화된 로깅으로 문제 해결 용이성 확보
graceful 종료: 모든 연결과 작업이 안전하게 종료되도록 처리
암호화 적용: 민감한 데이터에 적절한 암호화 적용
CSRF 보호: 상태 변경 요청에 CSRF 토큰 검증
CORS 설정: 적절한 CORS 정책으로 브라우저 보안 강화
API 문서화: OpenAPI 또는 유사한 도구로 API 문서화
비동기 패턴 이해: 블로킹 코드를 실행할 때 asyncio.to_thread() 활용

KR_aiohttp - somaz94/python-study GitHub Wiki

aiohttp

1️⃣ aiohttp 기초

주요 특징:

설치:

기본 서버 예제:

라우팅 시스템:

요청 및 응답 처리:

2️⃣ aiohttp 클라이언트

기본 사용법:

다양한 HTTP 메서드:

동시 요청 처리:

고급 클라이언트 기능:

1. 요청 타임아웃 설정:

2. 쿠키 및 세션 관리:

3. 요청 헤더 설정:

4. 프록시 설정:

5. SSL 인증서 검증:

3️⃣ 미들웨어와 핸들러

4️⃣ 데이터베이스 통합

5️⃣ WebSocket 구현

서버 측 WebSocket:

클라이언트 측 WebSocket:

실시간 채팅 애플리케이션 예제:

WebSocket 이벤트 관리:

6️⃣ 실용적인 예제

REST API 서비스:

파일 업로드 처리:

세션 관리:

7️⃣ aiohttp 애플리케이션 테스트

테스트 클라이언트:

pytest 픽스처 사용하기:

모킹과 의존성 주입:

WebSocket 테스트:

통합 테스트 작성:

비동기 테스트 팁:

8️⃣ aiohttp 애플리케이션 배포

gunicorn과 함께 사용하기:

Docker와 함께 배포하기:

프로덕션 환경 설정:

상태 모니터링 및 관리:

보안 강화:

스케일링 전략:

무중단 배포:

클라우드 환경에서의 배포:

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️