Skip to content

KR_Bottle

Jump to bottom
somaz edited this page Apr 22, 2025 · 5 revisions

Python Bottle 개념 정리

1️⃣ Bottle 기초

Bottle은 단일 파일로 구성된 간단하고 가벼운 WSGI 웹 프레임워크이다.

from bottle import Bottle, run, response, request

app = Bottle()

@app.route('/')
def hello():
    return 'Hello World!'

@app.route('/api/users/<id:int>')
def get_user(id):
    response.content_type = 'application/json'
    return {
        'id': id,
        'name': f'User {id}',
        'email': f'user{id}@example.com'
    }

@app.route('/api/users', method='POST')
def create_user():
    data = request.json
    # 사용자 생성 로직
    return {'status': 'created', 'data': data}

if __name__ == '__main__':
    run(app, host='localhost', port=8080, debug=True)

✅ 특징:

  • 단일 파일 구성
  • 간단한 라우팅 시스템
  • JSON 응답 지원
  • 의존성 없는 독립적 실행
  • WSGI 호환성

2️⃣ 라우팅과 HTTP 메서드

Bottle의 라우팅 시스템을 활용하여 다양한 HTTP 메서드를 처리하는 방법이다.

from bottle import route, get, post, put, delete, request

@route('/hello/<name>')
def hello(name):
    return f'Hello {name}!'

@get('/api/items')
def get_items():
    return {'items': ['item1', 'item2', 'item3']}

@post('/api/items')
def create_item():
    data = request.json
    # 아이템 생성 로직
    return {'status': 'created'}

@put('/api/items/<id:int>')
def update_item(id):
    data = request.json
    # 아이템 업데이트 로직
    return {'status': 'updated'}

@delete('/api/items/<id:int>')
def delete_item(id):
    # 아이템 삭제 로직
    return {'status': 'deleted'}

✅ 특징:

  • 다양한 HTTP 메서드 지원
  • URL 패턴 매칭
  • 데코레이터 기반 라우팅
  • 타입 변환 지원
  • 경로 매개변수 추출

3️⃣ 템플릿과 정적 파일

Bottle에서 템플릿 엔진을 사용하고 정적 파일을 제공하는 방법이다.

from bottle import route, template, static_file

@route('/static/<filename:path>')
def serve_static(filename):
    return static_file(filename, root='./static')

@route('/hello/<name>')
def hello_template(name):
    return template('hello_template', name=name)

# views/hello_template.tpl
"""
<!DOCTYPE html>
<html>
<head>
    <title>Hello {{name}}</title>
</head>
<body>
    <h1>Hello {{name}}!</h1>
</body>
</html>
"""

✅ 특징:

  • 템플릿 엔진 내장
  • 정적 파일 서빙
  • 간단한 템플릿 문법
  • MIME 타입 자동 감지
  • 조건부 HTTP 지원

4️⃣ 데이터베이스 통합

Bottle과 데이터베이스를 연동하여 데이터를 저장하고 조회하는 방법이다.

import sqlite3
from bottle import route, post, request, response

def db_connection():
    return sqlite3.connect('database.db')

@route('/users')
def get_users():
    conn = db_connection()
    c = conn.cursor()
    
    try:
        c.execute('SELECT * FROM users')
        users = [
            {'id': row[0], 'name': row[1], 'email': row[2]}
            for row in c.fetchall()
        ]
        return {'users': users}
    finally:
        conn.close()

@post('/users')
def create_user():
    data = request.json
    conn = db_connection()
    c = conn.cursor()
    
    try:
        c.execute(
            'INSERT INTO users (name, email) VALUES (?, ?)',
            (data['name'], data['email'])
        )
        conn.commit()
        return {'id': c.lastrowid}
    finally:
        conn.close()

✅ 특징:

  • 다양한 DB 지원
  • 커넥션 관리
  • 트랜잭션 처리
  • SQL 인젝션 방지
  • ORM 라이브러리 통합 가능

5️⃣ 미들웨어와 플러그인

요청/응답 처리 전후에 추가 로직을 실행할 수 있는 미들웨어와 플러그인 시스템이다.

from bottle import hook, request, response, install

@hook('before_request')
def setup_request():
    request.db = db_connection()

@hook('after_request')
def cleanup_request():
    if hasattr(request, 'db'):
        request.db.close()

class AuthPlugin:
    def apply(self, callback, route):
        def wrapper(*args, **kwargs):
            token = request.headers.get('Authorization')
            if not token:
                response.status = 401
                return {'error': '인증이 필요합니다'}
            
            # 토큰 검증 로직
            return callback(*args, **kwargs)
        return wrapper

app = Bottle()
app.install(AuthPlugin())

✅ 특징:

  • 미들웨어 시스템
  • 플러그인 아키텍처
  • 인증 처리 지원
  • 요청/응답 수정 가능
  • 기능 모듈화

6️⃣ 실용적인 예제

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

RESTful API 구현:

from bottle import Bottle, route, request, response

app = Bottle()

class UserAPI:
    def __init__(self):
        self.users = {}
    
    @app.route('/api/users', method='GET')
    def get_users(self):
        return {'users': list(self.users.values())}
    
    @app.route('/api/users/<id:int>', method='GET')
    def get_user(self, id):
        user = self.users.get(id)
        if not user:
            response.status = 404
            return {'error': '사용자를 찾을 수 없습니다'}
        return user
    
    @app.route('/api/users', method='POST')
    def create_user(self):
        data = request.json
        user_id = len(self.users) + 1
        user = {
            'id': user_id,
            'name': data['name'],
            'email': data['email']
        }
        self.users[user_id] = user
        response.status = 201
        return user

api = UserAPI()

파일 업로드 처리:

import os
from bottle import route, request, static_file

@route('/upload', method='POST')
def upload_file():
    upload = request.files.get('upload')
    if not upload:
        return {'error': '파일이 없습니다'}
    
    name, ext = os.path.splitext(upload.filename)
    if ext not in ('.png', '.jpg', '.jpeg'):
        return {'error': '지원하지 않는 파일 형식입니다'}
    
    save_path = f"uploads/{upload.filename}"
    upload.save(save_path)
    return {'filename': upload.filename}

@route('/uploads/<filename:path>')
def serve_upload(filename):
    return static_file(filename, root='./uploads')

세션 관리:

from bottle import route, request, response
import uuid
from datetime import datetime

class Session:
    def __init__(self):
        self.sessions = {}
    
    def create_session(self, user_id):
        session_id = str(uuid.uuid4())
        self.sessions[session_id] = {
            'user_id': user_id,
            'created_at': datetime.now()
        }
        return session_id
    
    def get_session(self, session_id):
        return self.sessions.get(session_id)
    
    def delete_session(self, session_id):
        self.sessions.pop(session_id, None)

session_manager = Session()

@route('/login', method='POST')
def login():
    data = request.json
    # 사용자 인증 로직
    def authenticate_user(data):
        # 실제 구현에서는 DB 확인 등의 로직
        if data.get('username') == 'admin' and data.get('password') == 'password':
            return 1
        return None
        
    user_id = authenticate_user(data)
    if user_id:
        session_id = session_manager.create_session(user_id)
        response.set_cookie('session_id', session_id)
        return {'status': 'logged in'}
    
    response.status = 401
    return {'error': '인증 실패'}

에러 처리:

from bottle import error, response

@error(404)
def error404(error):
    return {
        'error': '페이지를 찾을 수 없습니다',
        'status': 404
    }

@error(500)
def error500(error):
    return {
        'error': '서버 오류가 발생했습니다',
        'status': 500
    }

def handle_errors(fn):
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except Exception as e:
            response.status = 500
            return {'error': str(e)}
    return wrapper

@route('/api/protected')
@handle_errors
def protected_route():
    # 예외가 발생할 수 있는 코드
    result = 1 / 0  # 의도적인 예외
    return {'data': result}

비동기 작업 처리:

import asyncio
import threading
from bottle import route, run

# 백그라운드 작업을 위한 이벤트 루프
background_loop = asyncio.new_event_loop()

def run_async_task(coroutine):
    """비동기 작업을 백그라운드에서 실행"""
    asyncio.set_event_loop(background_loop)
    background_loop.run_until_complete(coroutine)

async def long_running_task(task_id):
    """시간이 오래 걸리는 작업 시뮬레이션"""
    print(f"작업 {task_id} 시작")
    await asyncio.sleep(5)  # I/O 작업 시뮬레이션
    print(f"작업 {task_id} 완료")
    return f"작업 {task_id} 결과"

@route('/async-task/<task_id>')
def start_async_task(task_id):
    # 백그라운드에서 비동기 작업 실행
    thread = threading.Thread(
        target=run_async_task,
        args=(long_running_task(task_id),)
    )
    thread.daemon = True
    thread.start()
    
    return {'status': 'task_started', 'task_id': task_id}

✅ 특징:

  • RESTful API 설계
  • 파일 업로드 처리
  • 세션 기반 인증
  • 종합적인 에러 처리
  • 비동기 작업 처리
  • 미들웨어 활용
  • 데이터 유효성 검증
  • 보안 고려사항

7️⃣ 배포와 확장

Bottle 애플리케이션을 프로덕션 환경에 배포하고 확장하는 방법이다.

WSGI 서버 배포:

# wsgi.py (gunicorn, uWSGI 등과 함께 사용)
from bottle import Bottle

app = Bottle()

@app.route('/')
def index():
    return "Hello from Production!"

# 추가 라우트 정의
# ...

# WSGI 호환성을 위한 애플리케이션 객체
application = app

# 로컬 테스트용
if __name__ == '__main__':
    from bottle import run
    run(app, host='localhost', port=8080)

애플리케이션 구조화:

# bottle_app/
# ├── app.py
# ├── routes/
# │   ├── __init__.py
# │   ├── user_routes.py
# │   └── product_routes.py
# ├── models/
# │   ├── __init__.py
# │   ├── user.py
# │   └── product.py
# ├── views/
# │   ├── base.tpl
# │   └── index.tpl
# └── static/
#     ├── css/
#     └── js/

# routes/user_routes.py
from bottle import Bottle, request, response

user_app = Bottle()

@user_app.route('/users')
def get_users():
    # 사용자 목록 반환
    pass

@user_app.route('/users/<id:int>')
def get_user(id):
    # 특정 사용자 반환
    pass

# app.py
from bottle import Bottle
from routes.user_routes import user_app
from routes.product_routes import product_app

app = Bottle()

# 하위 애플리케이션 마운트
app.mount('/api', user_app)
app.mount('/api', product_app)

# 기본 라우트
@app.route('/')
def index():
    return "Welcome to the main application"

성능 최적화:

from bottle import route, response
import time
import gzip
import functools

# 응답 캐싱
cache = {}

def cached(timeout=5 * 60):
    """간단한 메모리 캐싱 데코레이터"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # 캐시 항목 확인
            if key in cache:
                timestamp, value = cache[key]
                if time.time() - timestamp < timeout:
                    return value
            
            # 캐시 미스 또는 만료
            result = func(*args, **kwargs)
            cache[key] = (time.time(), result)
            return result
        return wrapper
    return decorator

# gzip 압축
def enable_gzip(func):
    """응답을 gzip으로 압축하는 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        
        # 문자열 응답만 처리
        if isinstance(result, str):
            response.headers['Content-Encoding'] = 'gzip'
            return gzip.compress(result.encode('utf-8'))
        
        return result
    return wrapper

@route('/heavy-operation')
@cached(timeout=60)
@enable_gzip
def heavy_operation():
    # 시간이 오래 걸리는 작업 시뮬레이션
    time.sleep(2)
    return "대량의 데이터 " * 1000  # 큰 응답

✅ 특징:

  • WSGI 서버 통합
  • 모듈화된 애플리케이션 구조
  • 캐싱 전략
  • 응답 압축
  • 로드 밸런싱 고려
  • 마이크로서비스 아키텍처
  • 데이터베이스 연결 풀링
  • 보안 강화

8️⃣ 보안 고려사항

Bottle 애플리케이션의 보안을 강화하는 방법이다.

CSRF 보호:

import uuid
from bottle import request, response, abort

class CSRFProtection:
    def __init__(self, app):
        self.app = app
        self.app.add_hook('before_request', self.before_request)
        self.tokens = {}  # 사용자 세션별 토큰 저장
    
    def before_request(self):
        # 읽기 전용 메서드는 검사 제외
        if request.method in ('GET', 'HEAD', 'OPTIONS'):
            return
        
        # CSRF 토큰 검사
        session_id = request.cookies.get('session_id')
        if not session_id:
            abort(403, "세션이 없습니다")
        
        token = request.headers.get('X-CSRF-Token')
        if not token or token != self.tokens.get(session_id):
            abort(403, "CSRF 토큰이 유효하지 않습니다")
    
    def generate_token(self, session_id):
        """새 CSRF 토큰 생성 및 저장"""
        token = str(uuid.uuid4())
        self.tokens[session_id] = token
        return token

# 애플리케이션에 CSRF 보호 추가
app = Bottle()
csrf = CSRFProtection(app)

@app.route('/get-csrf-token')
def get_csrf_token():
    session_id = request.cookies.get('session_id')
    if not session_id:
        # 세션이 없는 경우 생성
        session_id = str(uuid.uuid4())
        response.set_cookie('session_id', session_id)
    
    token = csrf.generate_token(session_id)
    return {'csrf_token': token}

입력 유효성 검사:

from bottle import route, request, abort
import re
import html

def validate_email(email):
    """이메일 형식 검증"""
    pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

def sanitize_input(text):
    """HTML 이스케이프 처리"""
    return html.escape(text)

@route('/register', method='POST')
def register():
    data = request.json
    
    # 필수 필드 확인
    if not all(k in data for k in ('username', 'email', 'password')):
        abort(400, "필수 필드가 누락되었습니다")
    
    # 이메일 유효성 검사
    if not validate_email(data['email']):
        abort(400, "유효하지 않은 이메일 형식입니다")
    
    # 비밀번호 강도 검사
    if len(data['password']) < 8:
        abort(400, "비밀번호는 최소 8자 이상이어야 합니다")
    
    # HTML 이스케이프 처리
    username = sanitize_input(data['username'])
    
    # 사용자 등록 로직
    # ...
    
    return {'status': 'success'}

헤더 보안:

from bottle import hook, response

@hook('after_request')
def add_security_headers():
    """보안 관련 HTTP 헤더 추가"""
    headers = response.headers
    
    # XSS 방지
    headers['X-XSS-Protection'] = '1; mode=block'
    
    # 클릭재킹 방지
    headers['X-Frame-Options'] = 'DENY'
    
    # MIME 스니핑 방지
    headers['X-Content-Type-Options'] = 'nosniff'
    
    # HTTPS 강제
    headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains'
    
    # 콘텐츠 보안 정책
    headers['Content-Security-Policy'] = "default-src 'self'"

✅ 특징:

  • CSRF 방어 매커니즘
  • 입력 데이터 검증
  • 보안 헤더 설정
  • XSS 방지
  • SQL 인젝션 방지
  • 세션 보안
  • 암호화 통신
  • 인증 및 권한 제어

주요 팁

✅ 모범 사례:

  • 모듈화 구조: 큰 애플리케이션은 기능별로 분리하여 구성
  • 컨텍스트 관리자: 리소스 관리에 with 문 활용
  • 적절한 HTTP 상태 코드: 응답에 올바른 상태 코드 사용
  • 설정 분리: 개발/테스트/프로덕션 환경별 설정 관리
  • 라우트 그룹화: 관련 기능은 하위 애플리케이션으로 구성
  • 보안 우선: 입력 검증, CSRF 보호, 세션 관리 철저히
  • 응답 최적화: 압축, 캐싱으로 성능 향상
  • 데이터베이스 연결 관리: 커넥션 풀 또는 컨텍스트 관리자 사용
  • 템플릿 상속: 공통 레이아웃은 베이스 템플릿 활용
  • 정적 파일 캐싱: 적절한 캐시 헤더 설정
  • WSGI 서버 활용: 프로덕션에서는 gunicorn, uWSGI 등 사용
  • 에러 처리: 사용자 친화적인 에러 메시지 제공
  • 로깅 구현: 디버깅 및 모니터링을 위한 로깅 설정
  • 성능 측정: 애플리케이션 지연 요소 모니터링
  • API 설계: 일관된 엔드포인트와 응답 형식 유지

Clone this wiki locally