8.2 웹 개발 API 사용법

파이썬을 사용하여 웹 개발 및 API를 구축하려면 Flask 또는 Django와 같은 웹 프레임워크를 사용합니다. 여기에서는 Flask를 사용하여 간단한 RESTful API를 구축하는 방법을 설명하겠습니다.

먼저 Flask를 설치해야 합니다. 다음 명령어를 사용하여 설치합니다.

pip install Flask

API를 구축하기 위해 간단한 Flask 애플리케이션을 작성해 봅시다. app.py라는 파일을 만들고 다음 코드를 작성합니다.

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/test', methods=['GET'])
def test():
    return jsonify({'message': 'Hello, World!'})

if __name__ == '__main__':
    app.run(debug=True)

이 코드는 간단한 Flask 애플리케이션을 만들고, /api/test 엔드포인트를 생성합니다. GET 요청이 이 엔드포인트로 전송되면 "Hello, World!"라는 메시지를 JSON 형식으로 반환합니다.

애플리케이션을 실행하려면 터미널에서 app.py 파일을 실행합니다.

python app.py

웹 브라우저에서 http://localhost:5000/api/test 주소로 이동하거나 curl과 같은 도구를 사용하여 API를 테스트할 수 있습니다.
이제, 간단한 POST 요청을 처리하는 엔드포인트를 추가해 봅시다. 예를 들어, 클라이언트로부터 JSON 데이터를 받아 합계를 반환하는 엔드포인트를 만들 수 있습니다.

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/test', methods=['GET'])
def test():
    return jsonify({'message': 'Hello, World!'})

@app.route('/api/sum', methods=['POST'])
def sum_numbers():
    data = request.get_json()
    numbers = data.get('numbers', [])
    result = sum(numbers)
    return jsonify({'result': result})

if __name__ == '__main__':
    app.run(debug=True)

이 엔드포인트는 /api/sum이며, POST 요청을 처리합니다. JSON 데이터를 받아 합계를 계산한 후 결과를 반환합니다.

이렇게 Flask를 사용하여 간단한 API를 만들 수 있습니다. 실제 프로젝트에서는 데이터베이스와의 연동, 인증 및 권한 관리, 에러 처리 등 추가적인 작업이 필요합니다.

또한, API 문서를 작성하고 유지 관리하는 것이 좋습니다. Swagger/OpenAPI와 같은 도구를 사용하여 API 문서를 자동화할 수 있습니다. 이를 통해 API의 사용법을 명확하게 전달하고, 클라이언트 개발자들이 API를 쉽게 이해하고 테스트할 수 있습니다.

Flask 애플리케이션을 확장하고 싶다면 다음과 같은 추가 작업을 고려해볼 수 있습니다.

1. 데이터베이스 연동: SQLAlchemy와 같은 ORM(Object Relational Mapper)을 사용하여 Flask 애플리케이션과 데이터베이스를 연동합니다. 이를 통해 데이터를 저장하고, 조회하고, 수정하고, 삭제하는 작업을 쉽게 수행할 수 있습니다.

 

예시: SQLAlchemy 설치 및 초기 설정

pip install Flask-SQLAlchemy

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    age = db.Column(db.Integer, nullable=False)

    def __repr__(self):
        return f'<User {self.name}>'

데이터베이스 및 테이블을 생성하려면 다음 코드를 실행합니다.

db.create_all()

이제 CRUD 작업을 수행하는 엔드포인트를 작성해 보겠습니다.

from flask import request

@app.route('/users', methods=['POST'])
def create_user():
    name = request.json['name']
    age = request.json['age']
    new_user = User(name=name, age=age)
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'message': 'User created'}), 201

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify({'id': user.id, 'name': user.name, 'age': user.age})

@app.route('/users', methods=['GET'])
def get_all_users():
    users = User.query.all()
    return jsonify([{'id': user.id, 'name': user.name, 'age': user.age} for user in users])

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get_or_404(user_id)
    user.name = request.json['name']
    user.age = request.json['age']
    db.session.commit()
    return jsonify({'message': 'User updated'})

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get_or_404(user_id)
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'User deleted'})

if __name__ == '__main__':
    app.run(debug=True)

이제 실행하고 있는 Flask 애플리케이션에 요청을 보내 사용자를 생성하고, 조회하고, 수정하고, 삭제할 수 있습니다. 예를 들어, Postman이나 curl과 같은 도구를 사용하여 요청을 보낼 수 있습니다.

 

2. 인증 및 권한 관리: Flask-JWT-Extended는 JSON Web Tokens(JWT)를 사용하여 Flask 애플리케이션에서 인증 및 권한 부여를 쉽게 구현할 수 있는 확장 모듈입니다. 아래에 설치 및 사용법을 설명하겠습니다.

 

Flask-JWT-Extended 설치

pip install Flask-JWT-Extended

 

예제를 위한 기본 설정 및 모델 생성

Flask 애플리케이션 및 사용자 모델을 설정하겠습니다. 이 예제에서는 간단한 사용자 모델을 사용하며, 패스워드를 안전하게 저장하기 위해 Werkzeug 라이브러리의 generate_password_hash 및 check_password_hash 함수를 사용합니다.

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
app.config['SECRET_KEY'] = 'super-secret-key'
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)

    def set_password(self, password):
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        return check_password_hash(self.password_hash, password)

db.create_all()

 

Flask-JWT-Extended 설정 및 라우트 생성

Flask-JWT-Extended를 설정하고, 사용자 인증 및 보호된 라우트를 생성하겠습니다.

from flask import jsonify, request
from flask_jwt_extended import (
    JWTManager, jwt_required, create_access_token, get_jwt_identity
)

app.config['JWT_SECRET_KEY'] = 'jwt-secret-key'
jwt = JWTManager(app)

@app.route('/login', methods=['POST'])
def login():
    username = request.json.get('username', None)
    password = request.json.get('password', None)
    
    user = User.query.filter_by(username=username).first()
    if user is None or not user.check_password(password):
        return jsonify({'message': 'Invalid username or password'}), 401

    access_token = create_access_token(identity=username)
    return jsonify(access_token=access_token)

@app.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    current_user = get_jwt_identity()
    return jsonify(logged_in_as=current_user), 200

 

테스트를 위한 사용자 생성

user = User(username='testuser')
user.set_password('testpassword')
db.session.add(user)
db.session.commit()

이제 애플리케이션을 실행하고 /login 엔드포인트에 유효한 사용자 이름과 패스워드를 전송하여 액세스 토큰을 받을 수 있습니다. 이 토큰을 사용하여 보호된 엔드포인트 /protected에 액세스할 수 있습니다.

참고: 이 예제에서는 간단한 사용자 모델 및 인증 방식을 사용했습니다. 실제 프로덕션 환경에서는 보안을 강화하고, 다양한 인증 및 권한 부여 기능을 구현해야 합니다. Flask-JWT-Extended의 기능 중 일부를 소개하겠습니다.

 

토큰 만료 시간 설정
토큰의 만료 시간을 설정할 수 있습니다. 예를 들어, 액세스 토큰의 만료 시간을 30분으로 설정하려면 다음과 같이 작성합니다.

from datetime import timedelta

app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(minutes=30)

 

토큰 새로 고침
토큰을 새로 고침하여 만료 시간을 연장할 수 있습니다. 먼저 리프레시 토큰의 만료 시간을 설정하고, 엔드포인트를 생성합니다.

from flask_jwt_extended import (
    create_refresh_token, jwt_refresh_token_required, get_jwt_identity
)

app.config['JWT_REFRESH_TOKEN_EXPIRES'] = timedelta(days=30)

# 로그인 라우트 수정
@app.route('/login', methods=['POST'])
def login():
    # ...
    refresh_token = create_refresh_token(identity=username)
    return jsonify(access_token=access_token, refresh_token=refresh_token)

@app.route('/refresh', methods=['POST'])
@jwt_refresh_token_required
def refresh():
    current_user = get_jwt_identity()
    new_access_token = create_access_token(identity=current_user)
    return jsonify(access_token=new_access_token)

사용자 권한에 따른 엔드포인트 보호
사용자 권한에 따라 엔드포인트에 대한 액세스를 제한할 수 있습니다. 예를 들어, 관리자 권한이 있는 사용자만 액세스할 수 있는 엔드포인트를 만들고 싶다면 다음과 같이 작성할 수 있습니다.

from functools import wraps

def admin_required(fn):
    @wraps(fn)
    @jwt_required
    def wrapper(*args, **kwargs):
        current_user = get_jwt_identity()
        user = User.query.filter_by(username=current_user).first()
        if user is None or not user.is_admin:
            return jsonify({'message': 'Admins only!'}), 403
        return fn(*args, **kwargs)
    return wrapper

@app.route('/admin', methods=['GET'])
@admin_required
def admin():
    return jsonify({'message': 'Welcome, admin!'})

 

3. 에러 처리: Flask에서는 @app.errorhandler 데코레이터를 사용하여 특정 에러 코드에 대한 에러 처리를 구현할 수 있습니다. 이를 통해 사용자에게 적절한 에러 메시지와 함께 에러 코드를 반환할 수 있습니다.

 

예시: 404 Not Found 에러 처리

@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Not Found'}), 404

4. API 버전 관리: 엔드포인트 URL에 버전을 포함시켜 API 버전 관리를 할 수 있습니다. 예를 들어, /api/v1/test와 같은 형식으로 엔드포인트를 구성할 수 있습니다. 이렇게 하면 이후 API 변경 사항을 새 버전으로 관리하면서 기존 클라이언트와의 호환성을 유지할 수 있습니다.

 

5. 이 외 Flask 애플리케이션을 확장하고 최적화하기 위한 다양한 기능이 있으며, 필요에 따라 적절한 확장 모듈을 사용하여 원하는 기능을 구현할 수 있습니다. Flask의 경우 커뮤니티에서 다양한 확장 모듈을 제공하고 있으므로, 필요한 기능을쉽게 찾아서 사용할 수 있습니다. 여기에는 다음과 같은 확장 모듈들이 포함됩니다.

 

Flask-CORS는 Flask 애플리케이션에 Cross-Origin Resource Sharing(CORS)를 쉽게 적용할 수 있도록 도와주는 확장입니다. CORS는 보안을 위해 웹 브라우저에서 제한하는 기본적인 정책인 동일 출처 정책(Same-Origin Policy)을 완화하거나 특정 요청에 대해 허용하는 기술입니다.

 

Flask-CORS 설치

pip install Flask-CORS

 

Flask 애플리케이션에 CORS 적용
Flask 애플리케이션을 생성한 후 Flask-CORS를 사용하여 CORS를 설정합니다.

from flask import Flask
from flask_cors import CORS

app = Flask(__name__)
CORS(app)

이렇게 하면 모든 라우트에 대해 모든 출처에서의 요청이 허용됩니다. 더 세밀한 제어를 위해 개별 라우트 또는 애플리케이션 전체에 대해 다양한 옵션을 설정할 수 있습니다.

CORS 옵션 설정
예를 들어, 애플리케이션 전체에서 특정 출처에서의 요청만 허용하려면 다음과 같이 작성할 수 있습니다.

CORS(app, origins=['http://example.com', 'https://another.example.com'])

 

개별 라우트에 대해 CORS 정책을 적용하려면 데코레이터를 사용할 수 있습니다.

from flask import jsonify
from flask_cors import cross_origin

@app.route('/cors_route')
@cross_origin(origins=['http://example.com'])
def cors_route():
    return jsonify({"message": "This route has CORS applied"})

 

다른 설정 옵션을 사용하려면 다음과 같이 작성할 수 있습니다.

CORS(app, resources={
    r'/api/*': {
        'origins': ['http://example.com'],
        'methods': ['GET', 'POST'],
        'allow_headers': ['Content-Type', 'Authorization']
    }
})

이렇게 하면 /api/* 패턴에 일치하는 라우트에 대해 http://example.com에서 오는 GET 및 POST 요청이 허용되며, 'Content-Type' 및 'Authorization' 헤더가 포함된 요청도 허용됩니다.

 

Flask-Caching은 Flask 애플리케이션에서 캐싱을 쉽게 구현할 수 있는 확장 모듈입니다. 캐싱을 사용하면 데이터 또는 결과를 일시적으로 저장하여 반복된 요청에 대해 빠르게 응답할 수 있습니다.

참고: Flask-Cache는 오래된 확장이며, 현재는 Flask-Caching을 사용하는 것이 좋습니다. 이 예제에서는 Flask-Caching을 사용합니다.

Flask-Caching 설치

pip install Flask-Caching

Flask 애플리케이션에 캐싱 적용
Flask 애플리케이션을 생성한 후 Flask-Caching을 사용하여 캐싱을 설정합니다.

from flask import Flask
from flask_caching import Cache

app = Flask(__name__)
app.config['CACHE_TYPE'] = 'simple'  # 기본 캐시 유형 설정
cache = Cache(app)

캐싱 데코레이터 사용
@cache.cached() 데코레이터를 사용하여 라우트에 캐싱을 적용할 수 있습니다.

from flask import jsonify

@app.route('/expensive_operation')
@cache.cached(timeout=60)  # 캐시 만료 시간을 60초로 설정
def expensive_operation():
    # 시간이 오래 걸리는 작업 수행
    result = perform_expensive_operation()
    return jsonify(result)

 

기타 캐싱 기능
또한 Flask-Caching은 다양한 캐싱 기능을 제공합니다. 예를 들어, 특정 키를 사용하여 캐시를 저장하고 검색하거나 캐시 항목을 제거할 수 있습니다.

@app.route('/store_cache')
def store_cache():
    cache.set('key', 'value', timeout=60)
    return 'Cache stored.'

@app.route('/retrieve_cache')
def retrieve_cache():
    value = cache.get('key')
    if value is None:
        return 'Cache not found.', 404
    return value

@app.route('/delete_cache')
def delete_cache():
    cache.delete('key')
    return 'Cache deleted.'

사용 가능한 캐시 유형
Flask-Caching은 여러 가지 캐시 유형을 지원합니다. 예를 들어, Memcached 또는 Redis와 같은 외부 캐시 시스템을 사용할 수 있습니다. 캐시 유형을 변경하려면 app.config['CACHE_TYPE'] 값을 수정하고 필요한 설정을 추가하세요.

# Memcached를 사용하는 경우
app.config['CACHE_TYPE'] = 'memcached'
app.config['CACHE_MEMCACHED_SERVERS'] = ['localhost:11211']

# Redis를 사용하는 경우
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_URL'] = 'redis://localhost:6379'

 

조건부 캐싱
특정 조건에 따라 캐싱을 적용하려면 @cache.cached() 데코레이터의 unless 매개변수를 사용하여 조건을 지정할 수 있습니다.

def should_skip_cache():
    # 캐싱을 건너뛸지 여부를 결정하는 로직
    return check_some_condition()

@app.route('/conditional_cache')
@cache.cached(timeout=60, unless=should_skip_cache)
def conditional_cache():
    result = perform_expensive_operation()
    return jsonify(result)

캐시 버전 관리
캐시 항목의 버전 관리를 사용하려면 @cache.memoize() 데코레이터를 사용하세요. 이 데코레이터를 사용하면 함수 결과를 캐시하고, 버전 번호를 증가시켜 캐시를 무효화할 수 있습니다.

@app.route('/versioned_cache/<int:param>')
@cache.memoize()
def versioned_cache(param):
    # 시간이 오래 걸리는 작업 수행
    result = perform_expensive_operation(param)
    return jsonify(result)

@app.route('/invalidate_cache/<int:param>')
def invalidate_cache(param):
    cache.delete_memoized(versioned_cache, param)
    return 'Cache invalidated.'

 

사용자 정의 캐시 키
기본적으로 Flask-Caching은 요청 경로를 기반으로 캐시 키를 생성합니다. 사용자 정의 캐시 키를 사용하려면 @cache.cached() 데코레이터의 key_prefix 매개변수를 사용하세요.

def custom_cache_key():
    # 사용자 정의 캐시 키 생성 로직
    return 'my_custom_key:' + get_some_data()

@app.route('/custom_key_cache')
@cache.cached(timeout=60, key_prefix=custom_cache_key)
def custom_key_cache():
    result = perform_expensive_operation()
    return jsonify(result)

 

마지막으로, 배포 시 고려해야 할 사항 중 하나는 WSGI(Web Server Gateway Interface) 서버를 사용하는 것입니다. Flask의 기본 개발 서버는 성능과 안정성 면에서 제한이 있으므로, Gunicorn, uWSGI 등의 WSGI 서버를 사용하여 배포하는 것이 좋습니다. 또한, 웹 서버(Nginx, Apache 등)와 함께 사용하여 정적 파일을 제공하고, 요청을 로드 밸런싱하는 등의 작업을 수행할 수 있습니다.