본 문서는 REST API 시리즈 중 4편입니다. 1편에서는 REST API의 기본 개념과 설계 원칙을, 2편에서는 Spring Boot를, 3편에서는 Nest.JS를 활용한 구현 방법을 다루었으며, 이번 편에서는 다양한 플랫폼에서의 REST API 구현 방법을 비교 설명합니다.
1. 다양한 플랫폼 비교
여러 프레임워크에서의 REST API 구현을 비교하기 전에, 각 플랫폼의 주요 특징을 살펴보겠습니다.
프레임워크 언어 타입 특징 적합한 상황
| Django REST Framework | Python | 풀스택 | 배터리 포함 방식, 관리자 인터페이스, ORM | 대규모 프로젝트, 복잡한 비즈니스 로직 |
| Flask | Python | 마이크로 | 경량, 확장성, 유연성 | 작은 프로젝트, 마이크로서비스, API |
| Express.js | JavaScript | 미니멀 | 유연성, 미들웨어 | 빠른 개발, Node.js 생태계 활용 |
| FastAPI | Python | 고성능 | 타입 힌트, 자동 문서화, 비동기 지원 | 고성능 API, 현대적 Python 프로젝트 |
2. Django REST Framework
Django REST Framework(DRF)는 Django 웹 프레임워크 위에 구축된 강력한 도구 세트로, 웹 API를 쉽게 구축할 수 있게 해줍니다.
2.1 설치 및 설정
# 가상환경 생성
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# Django 및 DRF 설치
pip install django djangorestframework
# 프로젝트 생성
django-admin startproject django_rest_demo
cd django_rest_demo
# 앱 생성
python manage.py startapp users
2.2 settings.py 설정
# django_rest_demo/settings.py
INSTALLED_APPS = [
# Django 앱
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
# 써드파티 앱
'rest_framework',
'rest_framework.authtoken',
# 로컬 앱
'users',
]
REST_FRAMEWORK = {
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.TokenAuthentication',
'rest_framework.authentication.SessionAuthentication',
],
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 10
}
2.3 모델 정의
# users/models.py
from django.db import models
from django.contrib.auth.models import AbstractUser
from django.utils.translation import gettext_lazy as _
class User(AbstractUser):
email = models.EmailField(_('email address'), unique=True)
phone = models.CharField(max_length=20, null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
USERNAME_FIELD = 'email'
REQUIRED_FIELDS = ['username']
def __str__(self):
return self.email
2.4 Serializer 정의
# users/serializers.py
from rest_framework import serializers
from .models import User
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ('id', 'username', 'email', 'phone', 'created_at', 'updated_at')
read_only_fields = ('id', 'created_at', 'updated_at')
class UserCreateSerializer(serializers.ModelSerializer):
password = serializers.CharField(write_only=True)
class Meta:
model = User
fields = ('id', 'username', 'email', 'phone', 'password')
extra_kwargs = {'password': {'write_only': True}}
def create(self, validated_data):
user = User.objects.create_user(
username=validated_data['username'],
email=validated_data['email'],
password=validated_data['password'],
phone=validated_data.get('phone')
)
return user
2.5 ViewSet 정의
# users/views.py
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import User
from .serializers import UserSerializer, UserCreateSerializer
class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer
permission_classes = [permissions.IsAuthenticated]
def get_serializer_class(self):
if self.action == 'create':
return UserCreateSerializer
return UserSerializer
def get_permissions(self):
if self.action == 'create':
return [permissions.AllowAny()]
return super().get_permissions()
@action(detail=False, methods=['get'])
def me(self, request):
serializer = self.get_serializer(request.user)
return Response(serializer.data)
@action(detail=False, methods=['get'], url_path='search/(?P<username>[\w.@+-]+)')
def search(self, request, username=None):
users = self.queryset.filter(username__icontains=username)
page = self.paginate_queryset(users)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(users, many=True)
return Response(serializer.data)
2.6 URL 설정
# users/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import UserViewSet
router = DefaultRouter()
router.register('users', UserViewSet)
urlpatterns = [
path('', include(router.urls)),
]
# django_rest_demo/urls.py
from django.contrib import admin
from django.urls import path, include
from rest_framework.authtoken.views import obtain_auth_token
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', include('users.urls')),
path('api/auth/', obtain_auth_token),
]
2.7 인증 구현
# users/auth.py
from rest_framework.authtoken.views import ObtainAuthToken
from rest_framework.authtoken.models import Token
from rest_framework.response import Response
class CustomAuthToken(ObtainAuthToken):
def post(self, request, *args, **kwargs):
serializer = self.serializer_class(data=request.data,
context={'request': request})
serializer.is_valid(raise_exception=True)
user = serializer.validated_data['user']
token, created = Token.objects.get_or_create(user=user)
return Response({
'token': token.key,
'user_id': user.pk,
'email': user.email
})
# django_rest_demo/urls.py에 추가
from users.auth import CustomAuthToken
urlpatterns = [
# 기존 URL 패턴들
path('api/auth/token/', CustomAuthToken.as_view()),
]
2.8 테스트 작성
# users/tests.py
from django.urls import reverse
from rest_framework.test import APITestCase
from rest_framework import status
from .models import User
class UserAPITestCase(APITestCase):
def setUp(self):
self.user = User.objects.create_user(
username='testuser',
email='test@example.com',
password='password123'
)
self.client.force_authenticate(user=self.user)
def test_get_user_list(self):
url = reverse('user-list')
response = self.client.get(url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data['results']), 1)
def test_create_user(self):
url = reverse('user-list')
data = {
'username': 'newuser',
'email': 'new@example.com',
'password': 'newpassword123',
'phone': '1234567890'
}
self.client.force_authenticate(user=None) # 로그아웃
response = self.client.post(url, data)
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(User.objects.count(), 2)
self.assertEqual(User.objects.get(email='new@example.com').username, 'newuser')
3. Flask와 Flask-RESTful
Flask는 Python의 마이크로 웹 프레임워크로, 핵심 기능을 단순하게 유지하면서도 확장성을 제공합니다. Flask-RESTful은 Flask 위에 API를 구축하기 위한 확장 모듈입니다.
3.1 설치 및 프로젝트 구조
# 가상환경 생성
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 의존성 설치
pip install flask flask-restful flask-sqlalchemy flask-migrate flask-jwt-extended marshmallow
# 프로젝트 구조
mkdir flask_rest_demo
cd flask_rest_demo
mkdir app
touch app/__init__.py
touch app/models.py
touch app/resources.py
touch app/schemas.py
touch config.py
touch run.py
3.2 설정 파일
# config.py
import os
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'dev_key')
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'sqlite:///app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY', 'jwt_dev_key')
JWT_ACCESS_TOKEN_EXPIRES = 3600 # 1시간
3.3 애플리케이션 초기화
# app/__init__.py
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_jwt_extended import JWTManager
from config import Config
db = SQLAlchemy()
migrate = Migrate()
jwt = JWTManager()
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
from app.resources import UserResource, UserListResource, UserLoginResource
api = Api(app, prefix='/api')
api.add_resource(UserListResource, '/users')
api.add_resource(UserResource, '/users/<int:user_id>')
api.add_resource(UserLoginResource, '/auth/login')
@app.route('/')
def home():
return {'message': 'Welcome to Flask REST API'}
return app
3.4 모델 정의
# app/models.py
from app import db
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), index=True, unique=True)
email = db.Column(db.String(120), index=True, unique=True)
phone = db.Column(db.String(20), nullable=True)
password_hash = db.Column(db.String(128))
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
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)
def __repr__(self):
return f'<User {self.username}>'
3.5 스키마 정의 (Marshmallow)
# app/schemas.py
from marshmallow import Schema, fields, validate, validates, ValidationError
from app.models import User
class UserSchema(Schema):
id = fields.Int(dump_only=True)
username = fields.Str(required=True, validate=validate.Length(min=2, max=50))
email = fields.Email(required=True)
phone = fields.Str(validate=validate.Regexp(r'^[0-9]{10,15}$'), allow_none=True)
password = fields.Str(load_only=True, required=True, validate=validate.Length(min=8))
created_at = fields.DateTime(dump_only=True)
updated_at = fields.DateTime(dump_only=True)
@validates('email')
def validate_email(self, value):
if User.query.filter_by(email=value).first():
raise ValidationError('Email already exists.')
@validates('username')
def validate_username(self, value):
if User.query.filter_by(username=value).first():
raise ValidationError('Username already exists.')
class UserUpdateSchema(Schema):
username = fields.Str(validate=validate.Length(min=2, max=50))
email = fields.Email()
phone = fields.Str(validate=validate.Regexp(r'^[0-9]{10,15}$'), allow_none=True)
password = fields.Str(load_only=True, validate=validate.Length(min=8))
class UserLoginSchema(Schema):
email = fields.Email(required=True)
password = fields.Str(required=True, load_only=True)
3.6 리소스 정의
# app/resources.py
from flask import request
from flask_restful import Resource
from flask_jwt_extended import (
create_access_token, jwt_required, get_jwt_identity
)
from marshmallow import ValidationError
from app import db
from app.models import User
from app.schemas import UserSchema, UserUpdateSchema, UserLoginSchema
user_schema = UserSchema()
users_schema = UserSchema(many=True)
user_update_schema = UserUpdateSchema()
user_login_schema = UserLoginSchema()
class UserListResource(Resource):
@jwt_required()
def get(self):
users = User.query.all()
return users_schema.dump(users)
def post(self):
try:
data = user_schema.load(request.json)
except ValidationError as err:
return {"errors": err.messages}, 400
user = User(
username=data['username'],
email=data['email'],
phone=data.get('phone')
)
user.set_password(data['password'])
db.session.add(user)
db.session.commit()
return user_schema.dump(user), 201
class UserResource(Resource):
@jwt_required()
def get(self, user_id):
user = User.query.get_or_404(user_id)
return user_schema.dump(user)
@jwt_required()
def patch(self, user_id):
user = User.query.get_or_404(user_id)
# 현재 로그인한 사용자가 자신의 정보만 수정할 수 있도록
current_user_id = get_jwt_identity()
if current_user_id != user.id:
return {"error": "You can only modify your own profile"}, 403
try:
data = user_update_schema.load(request.json)
except ValidationError as err:
return {"errors": err.messages}, 400
if 'username' in data:
user.username = data['username']
if 'email' in data:
user.email = data['email']
if 'phone' in data:
user.phone = data['phone']
if 'password' in data:
user.set_password(data['password'])
db.session.commit()
return user_schema.dump(user)
@jwt_required()
def delete(self, user_id):
user = User.query.get_or_404(user_id)
# 현재 로그인한 사용자가 자신의 계정만 삭제할 수 있도록
current_user_id = get_jwt_identity()
if current_user_id != user.id:
return {"error": "You can only delete your own account"}, 403
db.session.delete(user)
db.session.commit()
return "", 204
class UserLoginResource(Resource):
def post(self):
try:
data = user_login_schema.load(request.json)
except ValidationError as err:
return {"errors": err.messages}, 400
user = User.query.filter_by(email=data['email']).first()
if user and user.check_password(data['password']):
access_token = create_access_token(identity=user.id)
return {
"access_token": access_token,
"user": user_schema.dump(user)
}
return {"error": "Invalid email or password"}, 401
3.7 실행 파일
# run.py
from app import create_app, db
app = create_app()
if __name__ == '__main__':
app.run(debug=True)
3.8 데이터베이스 초기화
# 환경 변수 설정
export FLASK_APP=run.py
# 데이터베이스 초기화
flask db init
flask db migrate -m "Initial migration"
flask db upgrade
# 서버 실행
flask run
4. Express.js
Express.js는 Node.js를 위한 빠르고 간결한 웹 프레임워크로, REST API 구축에 널리 사용됩니다.
4.1 프로젝트 설정
# 프로젝트 초기화
mkdir express_rest_demo
cd express_rest_demo
npm init -y
# 의존성 설치
npm install express mongoose bcryptjs jsonwebtoken dotenv cors helmet morgan
npm install --save-dev nodemon
# 프로젝트 구조
mkdir src
mkdir src/models
mkdir src/controllers
mkdir src/routes
mkdir src/middleware
touch src/app.js
touch src/server.js
touch .env
4.2 환경 설정 (.env)
PORT=3000
NODE_ENV=development
MONGODB_URI=mongodb://localhost:27017/express_rest_demo
JWT_SECRET=your_jwt_secret
JWT_EXPIRES_IN=1h
4.3 서버 설정
// src/app.js
const express = require('express');
const mongoose = require('mongoose');
const cors = require('cors');
const helmet = require('helmet');
const morgan = require('morgan');
const dotenv = require('dotenv');
// 라우트 import
const userRoutes = require('./routes/users');
const authRoutes = require('./routes/auth');
// 환경 변수 로드
dotenv.config();
// Express 앱 생성
const app = express();
// 미들웨어
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
app.use(cors());
app.use(helmet());
app.use(morgan('dev'));
// 라우트
app.use('/api/users', userRoutes);
app.use('/api/auth', authRoutes);
// 메인 라우트
app.get('/', (req, res) => {
res.json({ message: 'Welcome to Express REST API' });
});
// 에러 핸들링 미들웨어
app.use((err, req, res, next) => {
const statusCode = err.statusCode || 500;
res.status(statusCode).json({
error: {
message: err.message || 'Internal Server Error',
...(process.env.NODE_ENV === 'development' && { stack: err.stack })
}
});
});
module.exports = app;
// src/server.js
const app = require('./app');
const mongoose = require('mongoose');
// 데이터베이스 연결
mongoose.connect(process.env.MONGODB_URI)
.then(() => {
console.log('Connected to MongoDB');
// 서버 시작
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
console.log(`Server running on port ${PORT}`);
});
})
.catch(err => {
console.error('Failed to connect to MongoDB', err);
process.exit(1);
});
4.4 모델 정의
// src/models/user.js
const mongoose = require('mongoose');
const bcrypt = require('bcryptjs');
const userSchema = new mongoose.Schema({
username: {
type: String,
required: true,
unique: true,
trim: true,
minlength: 2,
maxlength: 50
},
email: {
type: String,
required: true,
unique: true,
trim: true,
lowercase: true,
match: [/^\S+@\S+\.\S+$/, 'Please use a valid email address']
},
password: {
type: String,
required: true,
minlength: 8,
select: false // 쿼리 시 password 필드 제외
},
phone: {
type: String,
match: [/^[0-9]{10,15}$/, 'Please use a valid phone number'],
default: null
}
}, {
timestamps: true // createdAt, updatedAt 자동 생성
});
// 비밀번호 해싱 미들웨어
userSchema.pre('save', async function(next) {
if (!this.isModified('password')) return next();
try {
const salt = await bcrypt.genSalt(10);
this.password = await bcrypt.hash(this.password, salt);
next();
} catch (error) {
next(error);
}
});
// 비밀번호 검증 메서드
userSchema.methods.comparePassword = async function(candidatePassword) {
return await bcrypt.compare(candidatePassword, this.password);
};
const User = mongoose.model('User', userSchema);
module.exports = User;
4.5 컨트롤러 구현
// src/controllers/user.controller.js
const User = require('../models/user');
// 모든 사용자 조회
exports.getAllUsers = async (req, res, next) => {
try {
const users = await User.find();
res.status(200).json(users);
} catch (error) {
next(error);
}
};
// 특정 사용자 조회
exports.getUserById = async (req, res, next) => {
try {
const user = await User.findById(req.params.id);
if (!user) {
return res.status(404).json({ message: 'User not found' });
}
res.status(200).json(user);
} catch (error) {
next(error);
}
};
// 사용자 생성
exports.createUser = async (req, res, next) => {
try {
const { username, email, password, phone } = req.body;
// 이메일 중복 확인
const existingEmail = await User.findOne({ email });
if (existingEmail) {
return res.status(409).json({ message: 'Email already exists' });
}
// 사용자명 중복 확인
const existingUsername = await User.findOne({ username });
if (existingUsername) {
return res.status(409).json({ message: 'Username already exists' });
}
const user = new User({
username,
email,
password,
phone
});
await user.save();
// 비밀번호 제외하고 응답
const userResponse = user.toObject();
delete userResponse.password;
res.status(201).json(userResponse);
} catch (error) {
next(error);
}
};
// 사용자 정보 수정
exports.updateUser = async (req, res, next) => {
try {
const { username, email, phone, password } = req.body;
const userId = req.params.id;
// 권한 검사 (자신의 정보만 수정 가능)
if (req.user.id !== userId) {
return res.status(403).json({ message: 'You can only update your own profile' });
}
// 이메일 중복 확인 (변경된 경우)
if (email) {
const existingEmail = await User.findOne({ email, _id: { $ne: userId } });
if (existingEmail) {
return res.status(409).json({ message: 'Email already exists' });
}
}
// 사용자명 중복 확인 (변경된 경우)
if (username) {
const existingUsername = await User.findOne({ username, _id: { $ne: userId } });
if (existingUsername) {
return res.status(409).json({ message: 'Username already exists' });
}
}
let user = await User.findById(userId);
if (!user) {
return res.status(404).json({ message: 'User not found' });
}
// 필드 업데이트
if (username) user.username = username;
if (email) user.email = email;
if (phone !== undefined) user.phone = phone;
if (password) user.password = password;
await user.save();
// 비밀번호 제외하고 응답
const userResponse = user.toObject();
delete userResponse.password;
res.status(200).json(userResponse);
} catch (error) {
next(error);
}
};
// 사용자 삭제
exports.deleteUser = async (req, res, next) => {
try {
const userId = req.params.id;
// 권한 검사 (자신의 계정만 삭제 가능)
if (req.user.id !== userId) {
return res.status(403).json({ message: 'You can only delete your own account' });
}
const user = await User.findByIdAndDelete(userId);
if (!user) {
return res.status(404).json({ message: 'User not found' });
}
res.status(204).end();
} catch (error) {
next(error);
}
};
// 이름으로 사용자 검색
exports.searchUsersByName = async (req, res, next) => {
try {
const username = req.query.username;
if (!username) {
return res.status(400).json({ message: 'Username query parameter is required' });
}
const users = await User.find({
username: { $regex: username, $options: 'i' }
});
res.status(200).json(users);
} catch (error) {
next(error);
}
};
// src/controllers/auth.controller.js
const User = require('../models/user');
const jwt = require('jsonwebtoken');
// 로그인
exports.login = async (req, res, next) => {
try {
const { email, password } = req.body;
// 이메일 확인
const user = await User.findOne({ email }).select('+password');
if (!user) {
return res.status(401).json({ message: 'Invalid email or password' });
}
// 비밀번호 확인
const isMatch = await user.comparePassword(password);
if (!isMatch) {
return res.status(401).json({ message: 'Invalid email or password' });
}
// JWT 토큰 생성
const token = jwt.sign(
{ id: user._id },
process.env.JWT_SECRET,
{ expiresIn: process.env.JWT_EXPIRES_IN }
);
// 비밀번호 제외하고 응답
const userResponse = user.toObject();
delete userResponse.password;
res.status(200).json({
token,
user: userResponse
});
} catch (error) {
next(error);
}
};
// 현재 사용자 정보 조회
exports.getMe = async (req, res, next) => {
try {
const user = await User.findById(req.user.id);
res.status(200).json(user);
} catch (error) {
next(error);
}
};
4.6 미들웨어 정의
// src/middleware/auth.js
const jwt = require('jsonwebtoken');
const User = require('../models/user');
exports.protect = async (req, res, next) => {
try {
let token;
// Authorization 헤더에서 토큰 추출
if (req.headers.authorization && req.headers.authorization.startsWith('Bearer')) {
token = req.headers.authorization.split(' ')[1];
}
if (!token) {
return res.status(401).json({ message: 'Not authorized, no token' });
}
// 토큰 검증
const decoded = jwt.verify(token, process.env.JWT_SECRET);
// 사용자 정보 가져오기
const user = await User.findById(decoded.id);
if (!user) {
return res.status(401).json({ message: 'Not authorized, user not found' });
}
// 요청 객체에 사용자 정보 추가
req.user = { id: user._id.toString() };
next();
} catch (error) {
if (error.name === 'JsonWebTokenError') {
return res.status(401).json({ message: 'Not authorized, invalid token' });
}
if (error.name === 'TokenExpiredError') {
return res.status(401).json({ message: 'Not authorized, token expired' });
}
next(error);
}
};
4.7 라우트 정의
// src/routes/users.js
const express = require('express');
const {
getAllUsers,
getUserById,
createUser,
updateUser,
deleteUser,
searchUsersByName
} = require('../controllers/user.controller');
const { protect } = require('../middleware/auth');
const router = express.Router();
router.post('/', createUser);
router.get('/', protect, getAllUsers);
router.get('/search', protect, searchUsersByName);
router.get('/:id', protect, getUserById);
router.patch('/:id', protect, updateUser);
router.delete('/:id', protect, deleteUser);
module.exports = router;
// src/routes/auth.js
const express = require('express');
const { login, getMe } = require('../controllers/auth.controller');
const { protect } = require('../middleware/auth');
const router = express.Router();
router.post('/login', login);
router.get('/me', protect, getMe);
module.exports = router;
4.8 package.json 스크립트 설정
"scripts": {
"start": "node src/server.js",
"dev": "nodemon src/server.js",
"test": "echo \"Error: no test specified\" && exit 1"
}
5. FastAPI
FastAPI는 Python 3.6+ 기반의 현대적이고 빠른 웹 프레임워크로, 자동 API 문서화와 타입 힌트를 활용한 데이터 검증을 제공합니다.
5.1 설치 및 프로젝트 구조
# 가상환경 생성
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 의존성 설치
pip install fastapi uvicorn sqlalchemy pydantic python-jose[cryptography] passlib[bcrypt] python-multipart
# 프로젝트 구조
mkdir fastapi_rest_demo
cd fastapi_rest_demo
mkdir app
mkdir app/api
mkdir app/api/endpoints
mkdir app/core
mkdir app/crud
mkdir app/db
mkdir app/models
mkdir app/schemas
5.2 설정 파일
# app/core/config.py
from pydantic import BaseSettings
from typing import Optional, Dict, Any
class Settings(BaseSettings):
PROJECT_NAME: str = "FastAPI REST Demo"
API_V1_STR: str = "/api/v1"
SECRET_KEY: str = "your-secret-key" # 실무에서는 .env 파일에서 로드
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
DATABASE_URL: str = "sqlite:///./test.db"
class Config:
env_file = ".env"
settings = Settings()
5.3 데이터베이스 설정
# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(
settings.DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
5.4 모델 정의
# app/models/user.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from app.db.session import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
phone = Column(String, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
5.5 스키마 정의 (Pydantic)
# app/schemas/user.py
from typing import Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr, validator, constr
# 기본 사용자 스키마
class UserBase(BaseModel):
username: Optional[str] = None
email: Optional[EmailStr] = None
phone: Optional[str] = None
@validator('phone')
def validate_phone(cls, v):
if v is not None and not v.isdigit():
raise ValueError('Phone must contain only digits')
if v is not None and not (10 <= len(v) <= 15):
raise ValueError('Phone must be between 10 and 15 digits')
return v
# 사용자 생성 스키마
class UserCreate(UserBase):
username: str
email: EmailStr
password: constr(min_length=8)
# 사용자 업데이트 스키마
class UserUpdate(UserBase):
password: Optional[constr(min_length=8)] = None
# 데이터베이스에서 반환되는 사용자 스키마
class User(UserBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
# 인증용 스키마
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class UserLogin(BaseModel):
email: EmailStr
password: str
5.6 CRUD 작업
# app/crud/user.py
from sqlalchemy.orm import Session
from passlib.context import CryptContext
from typing import Optional, List, Dict, Any
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_user_by_email(db: Session, email: str) -> Optional[User]:
return db.query(User).filter(User.email == email).first()
def get_user_by_username(db: Session, username: str) -> Optional[User]:
return db.query(User).filter(User.username == username).first()
def get_user(db: Session, user_id: int) -> Optional[User]:
return db.query(User).filter(User.id == user_id).first()
def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
return db.query(User).offset(skip).limit(limit).all()
def search_users_by_username(db: Session, username: str, skip: int = 0, limit: int = 100) -> List[User]:
return db.query(User).filter(User.username.ilike(f"%{username}%")).offset(skip).limit(limit).all()
def create_user(db: Session, user: UserCreate) -> User:
hashed_password = get_password_hash(user.password)
db_user = User(
username=user.username,
email=user.email,
hashed_password=hashed_password,
phone=user.phone
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def update_user(db: Session, user_id: int, user: UserUpdate) -> Optional[User]:
db_user = get_user(db, user_id)
if not db_user:
return None
update_data = user.dict(exclude_unset=True)
if 'password' in update_data:
update_data['hashed_password'] = get_password_hash(update_data.pop('password'))
for field, value in update_data.items():
setattr(db_user, field, value)
db.commit()
db.refresh(db_user)
return db_user
def delete_user(db: Session, user_id: int) -> bool:
db_user = get_user(db, user_id)
if not db_user:
return False
db.delete(db_user)
db.commit()
return True
def authenticate_user(db: Session, email: str, password: str) -> Optional[User]:
user = get_user_by_email(db, email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
5.7 보안 및 인증
# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import jwt
from app.core.config import settings
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
# app/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.orm import Session
from typing import Optional
from app.core.config import settings
from app.db.session import get_db
from app.models.user import User
from app.schemas.user import TokenData
from app.crud.user import get_user
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
user_id: Optional[int] = payload.get("sub")
if user_id is None:
raise credentials_exception
token_data = TokenData(user_id=user_id)
except JWTError:
raise credentials_exception
user = get_user(db, user_id=token_data.user_id)
if user is None:
raise credentials_exception
return user
5.8 API 엔드포인트
# app/api/endpoints/auth.py
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.core.config import settings
from app.core.security import create_access_token
from app.crud.user import authenticate_user
from app.db.session import get_db
from app.schemas.user import Token, User, UserLogin
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests
"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
data={"sub": str(user.id)}, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/login/email", response_model=dict)
def login_with_email(
login_data: UserLogin, db: Session = Depends(get_db)
) -> Any:
"""
Login with email and password
"""
user = authenticate_user(db, login_data.email, login_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
data={"sub": str(user.id)}, expires_delta=access_token_expires
),
"token_type": "bearer",
"user": {
"id": user.id,
"username": user.username,
"email": user.email,
}
}
@router.get("/me", response_model=User)
def read_users_me(current_user: User = Depends(get_current_user)) -> Any:
"""
Get current user.
"""
return current_user
# app/api/endpoints/users.py
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.crud.user import (
create_user, get_user, get_users, update_user, delete_user,
get_user_by_email, get_user_by_username, search_users_by_username
)
from app.db.session import get_db
from app.schemas.user import User, UserCreate, UserUpdate
router = APIRouter()
@router.get("/", response_model=List[User])
def read_users(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
username: Optional[str] = Query(None, description="Filter by username"),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve users.
"""
if username:
users = search_users_by_username(db, username=username, skip=skip, limit=limit)
else:
users = get_users(db, skip=skip, limit=limit)
return users
@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
def create_new_user(
user_in: UserCreate, db: Session = Depends(get_db)
) -> Any:
"""
Create new user.
"""
user = get_user_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="A user with this email already exists",
)
user = get_user_by_username(db, username=user_in.username)
if user:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="A user with this username already exists",
)
return create_user(db, user_in)
@router.get("/{user_id}", response_model=User)
def read_user(
user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
) -> Any:
"""
Get a specific user by id.
"""
user = get_user(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
return user
@router.patch("/{user_id}", response_model=User)
def update_user_info(
user_id: int, user_in: UserUpdate, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
) -> Any:
"""
Update a user.
"""
if current_user.id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only update your own profile",
)
user = get_user(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
# Check for email uniqueness if email is being updated
if user_in.email and user_in.email != user.email:
if get_user_by_email(db, email=user_in.email):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="A user with this email already exists",
)
# Check for username uniqueness if username is being updated
if user_in.username and user_in.username != user.username:
if get_user_by_username(db, username=user_in.username):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="A user with this username already exists",
)
user = update_user(db, user_id=user_id, user=user_in)
return user
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user_account(
user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
) -> Any:
"""
Delete a user.
"""
if current_user.id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You can only delete your own account",
)
user = get_user(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
success = delete_user(db, user_id=user_id)
if not success:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to delete user",
)
return None
5.9 API 라우터 설정
# app/api/api.py
from fastapi import APIRouter
from app.api.endpoints import users, auth
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
5.10 애플리케이션 초기화
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.api import api_router
from app.core.config import settings
from app.db.session import Base, engine
# 데이터베이스 테이블 생성
Base.metadata.create_all(bind=engine)
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url=f"{settings.API_V1_STR}/openapi.json"
)
# CORS 설정
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 실제 환경에서는 특정 도메인으로 제한해야 함
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/")
def root():
return {"message": "Welcome to FastAPI REST API"}
5.11 서버 실행
# 개발 서버 실행
uvicorn app.main:app --reload
6. 프레임워크 비교 분석
각 프레임워크의 특징과 장단점을 비교하여 프로젝트에 적합한 프레임워크를 선택하는 데 도움이 되도록 해보겠습니다.
6.1 코드 복잡성 및 개발 속도
프레임워크 복잡성 개발 속도 초기 설정 난이도
| Django REST Framework | 중간 | 빠름 | 낮음 (많은 기능이 내장됨) |
| Flask | 낮음 | 빠름 | 낮음 (단순함) |
| Express.js | 낮음 | 빠름 | 낮음 (유연함) |
| FastAPI | 낮음 | 매우 빠름 | 낮음 (자동화 기능) |
| Spring Boot | 높음 | 중간 | 높음 (많은 설정 필요) |
| Nest.JS | 높음 | 중간 | 중간 (구조화된 접근) |
6.2 성능 및 확장성
프레임워크 성능 확장성 비고
| Django REST Framework | 중간 | 높음 | ORM으로 인한 오버헤드 |
| Flask | 높음 | 중간 | 확장이 필요한 경우 추가 설정 필요 |
| Express.js | 매우 높음 | 높음 | 이벤트 루프 기반 |
| FastAPI | 매우 높음 | 높음 | Starlette, Pydantic의 고성능 |
| Spring Boot | 높음 | 매우 높음 | JVM 기반 안정성 |
| Nest.JS | 높음 | 높음 | Express.js 또는 Fastify 기반 |
6.3 개발자 경험 및 생태계
프레임워크 개발자 경험 문서화 생태계 커뮤니티
| Django REST Framework | 우수 | 우수 | 풍부 | 대규모 |
| Flask | 우수 | 우수 | 중간 | 대규모 |
| Express.js | 우수 | 우수 | 매우 풍부 | 매우 대규모 |
| FastAPI | 우수 | 우수 | 성장 중 | 성장 중 |
| Spring Boot | 우수 | 우수 | 매우 풍부 | 매우 대규모 |
| Nest.JS | 우수 | 우수 | 성장 중 | 성장 중 |
6.4 사용 사례 및 권장 상황
프레임워크 권장 사용 사례
| Django REST Framework | 관리자 패널이 필요한 경우, 기존 Django 프로젝트에 API 추가, 빠르게 개발해야 하는 복잡한 비즈니스 로직 |
| Flask | 간단한 API, 마이크로서비스, 프로토타입, 특정 요구에 맞는 커스텀 솔루션 |
| Express.js | 실시간 애플리케이션, 높은 동시성이 필요한 경우, JavaScript/Node.js 생태계 활용 |
| FastAPI | 고성능 API, 자동 문서화가 중요한 경우, 타입 힌트를 활용한 현대적인 Python 개발 |
| Spring Boot | 엔터프라이즈급 애플리케이션, 복잡한 비즈니스 로직, Java/Kotlin 생태계 선호, 마이크로서비스 아키텍처 |
| Nest.JS | TypeScript 기반 백엔드, Angular 개발자, 엔터프라이즈급 기능이 필요한 Node.js 애플리케이션 |
7. 프레임워크 선택 가이드
REST API를 개발할 때 프레임워크 선택을 위한 가이드라인:
7.1 프로젝트 요구사항 파악
- 프로젝트 규모: 소규모, 중규모, 대규모 프로젝트인지
- 성능 요구사항: 동시성, 지연 시간, 처리량 등
- 팀 역량: 팀이 이미 알고 있는 언어와 프레임워크
- 개발 일정: 개발 속도와 시간 제약
- 확장성 요구사항: 향후 확장 계획
- 생태계 요구사항: 필요한 라이브러리와 도구
7.2 프레임워크 선택 결정 트리
시작
|
├─ 기존 프로젝트에 API 추가?
| ├─ Yes → 기존 프레임워크와 호환되는 솔루션 선택
| └─ No → 계속
|
├─ 개발 속도가 가장 중요?
| ├─ Yes → 팀이 알고 있는 언어 기반 프레임워크 선택
| | (Django REST Framework, Flask, Express.js)
| └─ No → 계속
|
├─ 성능이 가장 중요?
| ├─ Yes → FastAPI, Express.js
| └─ No → 계속
|
├─ 타입 안전성과 엔터프라이즈 기능 필요?
| ├─ Yes → Spring Boot, Nest.JS
| └─ No → 계속
|
├─ 자동 문서화 필요?
| ├─ Yes → FastAPI, Swagger 지원 프레임워크
| └─ No → 계속
|
├─ 많은 내장 기능 선호?
| ├─ Yes → Django REST Framework, Spring Boot
| └─ No → 계속
|
└─ 유연성과 간단함 선호?
├─ Yes → Flask, Express.js
└─ No → 가장 적합한 프레임워크 재평가
8. 결론
이 문서에서는 다양한 프레임워크를 통한 REST API 구현 방법을 살펴보았습니다. 각 프레임워크는 고유한 장점과 특징을 가지고 있으며, 프로젝트의 요구사항과 개발 팀의 역량에 따라 최적의 선택이 달라질 수 있습니다.
REST API 설계 및 구현의 핵심 원칙은 프레임워크에 관계없이 적용됩니다:
- 자원 중심 설계
- 적절한 HTTP 메서드 사용
- 일관된 URL 구조
- 적절한 상태 코드 사용
- 명확한 오류 처리
- 보안 고려사항
프레임워크 선택 시 고려해야 할 주요 사항:
- 개발 언어 및 생태계: 팀의 기존 역량과 프로젝트 통합 요구사항
- 성능과 확장성: 애플리케이션의 부하 및 성장 예측
- 개발 속도와 유지보수성: 프로젝트 일정 및 장기적 관리 계획
- 커뮤니티 및 지원: 문제 해결과 지식 공유를 위한 자원
모든 프레임워크는 그 자체로 훌륭한 도구이며, 어떤 프레임워크가 "최고"라는 것은 없습니다. 중요한 것은 프로젝트와 팀에 가장 적합한 도구를 선택하는 것입니다.
9. 후속 시리즈 안내
본 문서는 REST API 구현 시리즈의 마지막 편입니다. 시리즈의 주제는 다음과 같습니다:
시리즈 제목 주요 내용
| 1편 | REST API의 기본 개념과 설계 원칙 | REST 아키텍처, HTTP 메서드, 상태 코드, 설계 모범 사례 등 |
| 2편 | Spring Boot에서의 REST API 작성 예시 | Java 기반 Spring Boot 프레임워크를 이용한 REST API 구현 |
| 3편 | Nest.JS에서의 REST API 작성 예시 | TypeScript 기반 Nest.JS 프레임워크를 이용한 REST API 구현 |
| 4편 | 그 외 플랫폼에서의 REST API 작성 예시 | Django, Flask, Express.js, FastAPI 등 다양한 프레임워크에서의 구현 |
각 편에서는 해당 프레임워크의 특성을 살린 REST API 설계 및 구현 방법, 보안 처리, 테스트 방법 등을 실제 코드 예제와 함께 제공했습니다.
10. 참고 자료
- Django REST Framework 공식 문서
- Flask 공식 문서
- Flask-RESTful 공식 문서
- Express.js 공식 문서
- FastAPI 공식 문서
- Spring Boot 공식 문서
- Nest.JS 공식 문서
- REST API 설계 모범 사례
- HTTP 상태 코드
11. 마무리
이 시리즈를 통해 REST API의 기본 개념부터 다양한 프레임워크에서의 구현 방법까지 살펴보았습니다. REST API는 현대 웹 개발에서 필수적인 요소로, 클라이언트와 서버 간의 효율적인 통신을 가능하게 합니다.
각 프레임워크는 고유한 특성과 장점을 가지고 있으며, 이를 이해하고 프로젝트에 맞는 도구를 선택하는 것이 중요합니다. Django REST Framework와 같은 배터리 포함 접근 방식은 빠른 개발을, Flask와 Express.js와 같은 경량 프레임워크는 유연성을, FastAPI와 같은 현대적 프레임워크는 고성능과 개발자 경험을, Spring Boot와 Nest.JS와 같은 엔터프라이즈급 프레임워크는 확장성과 타입 안전성을 제공합니다.
REST API 개발은 기술적 구현을 넘어 좋은 설계와 사용자 경험을 고려하는 것이 중요합니다. 이 시리즈에서 다룬 설계 원칙과 모범 사례를 적용하여, 각 프레임워크의 강점을 활용한다면 견고하고 효율적인 API를 구축할 수 있을 것입니다.
마지막으로, 웹 개발은 계속해서 진화하고 있으며, 새로운 도구와 접근 방식이 등장하고 있습니다. GraphQL, gRPC와 같은 대안적인 API 설계 방식도 특정 사용 사례에서 고려해볼 가치가 있습니다. 개발자로서 계속해서 학습하고, 새로운 기술과 방법론을 탐색하는 것이 중요합니다.
이 시리즈가 여러분의 REST API 개발 여정에 도움이 되었기를 바랍니다.
감사합니다.
'Study' 카테고리의 다른 글
| Nest.JS에서의 REST API 작성 예시 (3/4) (0) | 2025.04.26 |
|---|---|
| Spring Boot에서의 REST API 작성 예시 (2/4) (0) | 2025.04.26 |
| REST API 란 무엇인가? (1/4) (0) | 2025.04.26 |
| Spring JPA에서 PostgreSQL JSONB 타입 활용하기 (1) | 2025.04.21 |