동아리 친구가 공부하고 있는걸 보고 좋아보여서 따라산 책.
한국어로 나온 백엔드 개발관련 도서는 대부분 자바+스프링쪽에 몰려있어서 파이썬으로 구현하는 백엔드에 대해 이렇게 너무 가볍지도 어렵지도 않은 정도로 잘 설명해주는 책은 찾기가 어려웠다....
그런데 이 책을 읽으면서 정말 너무 마음에 들어서 그런 아쉬움이 어느정도 해소되는 느낌이 들었다!
이 책은 큰 틀에서 현대 웹 시스템 구조, HTTP의 구조와 핵심요소, 미니터(트위터와 비슷한 기능이다.) 기능 개발, 데이터베이스 연결 및 사용, 인증 구현, unit test, aws, API 아키텍쳐, 파일 업로드 엔드포인트 등의 내용을 다루고 있다.
그 중 내가 가장 인상깊게 봤던 파트는 인증구현과 유닛테스트, API 아키텍쳐(코드 아키텍쳐) 부분인데, 그 중 특히 재밌었던 내용을 잠깐 소개하고자 한다.
10장, API 아키텍쳐
1. 아키텍쳐 개선 전 코드
이전장까지 만들었던 서비스는 미니터 서비스로, 회원가입/로그인 기능같은 간단한 유저 관련 기능과 팔로우/언팔로우/트위터 글 쓰기/타임라인(내글+내가 팔로우한 사람들의 글 모아보기) 기능과 같은 트위터 관련 기능들을 포함하고 있다.
이 서비스들은 app.py라는 하나의 파일에 다음과 같이 작성되어 있었다.
코드의 자세한 설명은 아래 코드 사이사이에 주석으로 첨부했다!
import jwt
import bcrypt
from flask import Flask, request, jsonify, current_app, Response, g
from flask.json import JSONEncoder
from sqlalchemy import create_engine, text
from datetime import datetime, timedelta
from functools import wraps
from flask_cors import CORS
## Default JSON encoder는 set를 JSON으로 변환할 수 없다.
## 그러므로 커스텀 엔코더를 작성해서 set을 list로 변환하여
## JSON으로 변환 가능하게 해주어야 한다.
class CustomJSONEncoder(JSONEncoder):
def default(self, obj):
if isinstance(obj, set):
return list(obj)
return JSONEncoder.default(self, obj)
## 데이터베이스에 접근해서 정보를 등록/수정/삭제하거나 가져오는 부분이다.
## (sqlalchemy의 text를 import하여 sql문을 실행하도록 함.)
def get_user(user_id):
user = current_app.database.execute(text("""
SELECT
id,
name,
email,
profile
FROM users
WHERE id = :user_id
"""), {
'user_id' : user_id
}).fetchone()
return {
'id' : user['id'],
'name' : user['name'],
'email' : user['email'],
'profile' : user['profile']
} if user else None
def insert_user(user):
return current_app.database.execute(text("""
INSERT INTO users (
name,
email,
profile,
hashed_password
) VALUES (
:name,
:email,
:profile,
:password
)
"""), user).lastrowid
def insert_tweet(user_tweet):
return current_app.database.execute(text("""
INSERT INTO tweets (
user_id,
tweet
) VALUES (
:id,
:tweet
)
"""), user_tweet).rowcount
def insert_follow(user_follow):
return current_app.database.execute(text("""
INSERT INTO users_follow_list (
user_id,
follow_user_id
) VALUES (
:id,
:follow
)
"""), user_follow).rowcount
def insert_unfollow(user_unfollow):
return current_app.database.execute(text("""
DELETE FROM users_follow_list
WHERE user_id = :id
AND follow_user_id = :unfollow
"""), user_unfollow).rowcount
def get_timeline(user_id):
timeline = current_app.database.execute(text("""
SELECT
t.user_id,
t.tweet
FROM tweets t
LEFT JOIN users_follow_list ufl ON ufl.user_id = :user_id
WHERE t.user_id = :user_id
OR t.user_id = ufl.follow_user_id
"""), {
'user_id' : user_id
}).fetchall()
return [{
'user_id' : tweet['user_id'],
'tweet' : tweet['tweet']
} for tweet in timeline]
def get_user_id_and_password(email):
row = current_app.database.execute(text("""
SELECT
id,
hashed_password
FROM users
WHERE email = :email
"""), {'email' : email}).fetchone()
return {
'id' : row['id'],
'hashed_password' : row['hashed_password']
} if row else None
## 특정 함수가 로그인을 요구하도록 하는 데코레이터 함수이다.
## 이 데코레이터가 달린 해당 함수를 실행시키기 위해서는 로그인이 강제되도록 한다.
def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
access_token = request.headers.get('Authorization')
if access_token is not None:
try:
payload = jwt.decode(access_token, current_app.config['JWT_SECRET_KEY'], 'HS256')
except jwt.InvalidTokenError:
payload = None
if payload is None: return Response(status=401)
user_id = payload['user_id']
g.user_id = user_id
g.user = get_user(user_id) if user_id else None
else:
return Response(status = 401)
return f(*args, **kwargs)
return decorated_function
## import한 Flask 클래스를 객체화시켜서 app이라는 변수에 저장한다.
## flask의 route 데코레이터를 사용하여 엔드포인트를 등록한다. (본격적인 기능들 구현)
## 데이터베이스는 config 파일에서 설정한 데이터베이스를 사용하기 때문에,
## 테스트를 진행할때는 test config 파일을 사용하여야 한다.
def create_app(test_config = None):
app = Flask(__name__)
CORS(app)
app.json_encoder = CustomJSONEncoder
if test_config is None:
app.config.from_pyfile("config.py")
else:
app.config.update(test_config)
database = create_engine(app.config['DB_URL'], encoding = 'utf-8', max_overflow = 0)
app.database = database
@app.route("/ping", methods=['GET'])
def ping():
return "pong"
@app.route("/sign-up", methods=['POST'])
def sign_up():
new_user = request.json
new_user['password'] = bcrypt.hashpw(
new_user['password'].encode('UTF-8'),
bcrypt.gensalt()
)
new_user_id = insert_user(new_user)
new_user = get_user(new_user_id)
return jsonify(new_user)
@app.route('/login', methods=['POST'])
def login():
credential = request.json
email = credential['email']
password = credential['password']
user_credential = get_user_id_and_password(email)
if user_credential and bcrypt.checkpw(password.encode('UTF-8'), user_credential['hashed_password'].encode('UTF-8')):
user_id = user_credential['id']
payload = {
'user_id' : user_id,
'exp' : datetime.utcnow() + timedelta(seconds = 60 * 60 * 24)
}
token = jwt.encode(payload, app.config['JWT_SECRET_KEY'], 'HS256')
return jsonify({
'user_id' : user_id,
'access_token' : token.decode('UTF-8')
})
else:
return '', 401
@app.route('/tweet', methods=['POST'])
@login_required
def tweet():
user_tweet = request.json
user_tweet['id'] = g.user_id
tweet = user_tweet['tweet']
if len(tweet) > 300:
return '300자를 초과했습니다', 400
insert_tweet(user_tweet)
return '', 200
@app.route('/follow', methods=['POST'])
@login_required
def follow():
payload = request.json
payload['id'] = g.user_id
insert_follow(payload)
return '', 200
@app.route('/unfollow', methods=['POST'])
@login_required
def unfollow():
payload = request.json
payload['id'] = g.user_id
insert_unfollow(payload)
return '', 200
@app.route('/timeline/<int:user_id>', methods=['GET'])
def timeline(user_id):
return jsonify({
'user_id' : user_id,
'timeline' : get_timeline(user_id)
})
@app.route('/timeline', methods=['GET'])
@login_required
def user_timeline():
user_id = g.user_id
return jsonify({
'user_id' : user_id,
'timeline' : get_timeline(user_id)
})
return app
2. 코드 아키텍쳐를 구상할 때 염두에 두어야 하는 요소들
- 확장성(Extensiblity)
- 재사용성(Reusablity)
- 보수 유지 가능성(Maintability)
- 가독성(Readability)
- 테스트가능성(Testability)
3. 레이어드 패턴
Multi-tier 아키텍처 패턴이라고도 하는 레이어드 아키텍쳐는 코드를 각 역할에 따라 독립된 모듈로 나누어 구성하는 패턴이다. 그리고 각 모듈이 서로의 의존도에 따라 층층히 쌓듯 연결되어서 전체의 시스템을 구현하는 구조다.
각 시스템마다 구조가 다를 수 있지만 일반적으로는 다음과 같은 3개의 레이어가 존재한다.
| presentation layer | 클라이언트 시스템과 직접적으로 연결되는 부분. 백엔드 api에서는 엔드포인트에 해당한다. 그러므로 presentation layer에서는 api의 엔드포인트를 정의하고 단순히 전송된 HTTP요청들을 읽어들이는 로직을 구현한다. |
| business layer | 실제 시스템이 구현해야하는 비즈니스 로직을 구현하는 부분. 예를 들어 트위터가 300자가 넘으면 create을 거부해야 하는 로직 등이 이 business layer에서 구현된다. |
| persistence layer | 데이터베이스와 관련된 로직을 구현하는 부분. business layer에서 필요한 데이터 생성, 수정, 읽기 등을 처리하여 실제 데이터베이스에서 데이터를 CRUD하는 역할을 한다. |
이러한 레이어드 아키텍처의 핵심 요소는 바로 단방향 의존성이다. 각각의 레이어는 오직 자기보다 하위에 있는 레이어에만 의존한다. (presentation layer는 business layer에 의존하고, business layer는 persistence layer에만 의존한다. 그 반대 관계에 대해서는 완전히 독립적이다.)
또 다른 핵심요소는 "separation of concerns"이다. 각 레이어의 역할이 명확하다는 뜻이다.
예를 들어 presentation layer에서는 비즈니스 로직을 전혀 구현하지 않으므로, 로직 처리를 위해서는 presentation에서 business layer의 코드를 호출해서 처리해야 한다.
그렇기 때문에 이러한 레이어드 아키텍처 구조로 코드를 구현하면 확장성이 높아진다.
또 각 레이어의 역할과 구조가 명확하므로 코드의 가독성 또한 높아지고, 코드를 재사용하거나 테스트 코드를 구현하기에도 편리해진다. (유닛테스트도 각 레이어별로 해줘야 한다.)
4. 레이어드 아키텍처 적용 후 코드
이제 레이어드 아키텍처를 적용해서 위에서 확인한 코드를 개선해보자.
모든 기능을 다 넣으면 코드가 좀 길어져서 아키텍처를 확인하기 복잡할것같아서, 트위터기능을 제외한 유저기능부분의 코드만 각 레이어별로 떼어서 가져와봤다.
* view/__init__.py (presentation layer)
presentation layer에서는 위에서 설명한 것과 같이 엔드포인트를 정의하고, request(HTTP 요청)에서 필요한 인풋 데이터를 받은 후, business layer의 코드(여기서는 service 디렉터리에 있는 코드들)를 호출해서 비즈니스 로직을 실행시킨 후 그 결괏값으로 response(HTTP 응답)를 리턴해준다.
import jwt
from flask import request, jsonify, current_app, Response, g
from flask.json import JSONEncoder
from functools import wraps
class CustomJSONEncoder(JSONEncoder):
def default(self, obj):
if isinstance(obj, set):
return list(obj)
return JSONEncoder.default(self, obj)
def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
access_token = request.headers.get('Authorization')
if access_token is not None:
try:
payload = jwt.decode(access_token, current_app.config['JWT_SECRET_KEY'], 'HS256')
except jwt.InvalidTokenError:
payload = None
if payload is None: return Response(status=401)
user_id = payload['user_id']
g.user_id = user_id
else:
return Response(status = 401)
return f(*args, **kwargs)
return decorated_function
def create_endpoints(app, services):
app.json_encoder = CustomJSONEncoder
user_service = services.user_service
tweet_service = services.tweet_service
@app.route("/ping", methods=['GET'])
def ping():
return "pong"
@app.route("/sign-up", methods=['POST'])
def sign_up():
new_user = request.json
new_user = user_service.create_new_user(new_user)
return jsonify(new_user)
@app.route('/login', methods=['POST'])
def login():
credential = request.json
authorized = user_service.login(credential)
if authorized:
user_credential = user_service.get_user_id_and_password(credential['email'])
user_id = user_credential['id']
token = user_service.generate_access_token(user_id)
return jsonify({
'user_id' : user_id,
'access_token' : token
})
else:
return '', 401
* service/user_service.py (business layer)
business layer에서는 비즈니스 로직을 구현한다. user_service모듈에서는 유저 관련 기능만 구현하므로 회원가입, 로그인, 인증 토큰발행, 팔로우, 언팔로우 등의 로직을 포함한다.
비즈니스 로직 외의 것 (데이터베이스에 새로운 유저를 추가하거나, 팔로우 유저리스트에 유저를 넣었다 뺐다 하는 것 등등)은 persistence layer에서 처리하므로 여기에서는 구현하지 않는다.
import jwt
import bcrypt
from datetime import datetime, timedelta
class UserService:
def __init__(self, user_dao, config):
self.user_dao = user_dao
self.config = config
def create_new_user(self, new_user):
new_user['password'] = bcrypt.hashpw(
new_user['password'].encode('UTF-8'),
bcrypt.gensalt()
)
new_user_id = self.user_dao.insert_user(new_user)
return new_user_id
def login(self, credential):
email = credential['email']
password = credential['password']
user_credential = self.user_dao.get_user_id_and_password(email)
authorized = user_credential and bcrypt.checkpw(password.encode('UTF-8'), user_credential['hashed_password'].encode('UTF-8'))
return authorized
def generate_access_token(self, user_id):
payload = {
'user_id' : user_id,
'exp' : datetime.utcnow() + timedelta(seconds = 60 * 60 * 24)
}
token = jwt.encode(payload, self.config.JWT_SECRET_KEY, 'HS256')
return token.decode('UTF-8')
def follow(self, user_id, follow_id):
return self.user_dao.insert_follow(user_id, follow_id)
def unfollow(self, user_id, unfollow_id):
return self.user_dao.insert_unfollow(user_id, unfollow_id)
def get_user_id_and_password(self, email):
return self.user_dao.get_user_id_and_password(email)
* model/user_dao.py (persistence layer)
이제 마지막으로 데이터베이스에 접근하는 persistence layer를 구현한다.
아래 파일은 user_dao.py라는 모듈인데 여기서 dao는 data access objects의 약자로, 데이터베이스의 상세한 사항을 노출시키지 않으면서 특정 데이터의 일부 동작을 제공해주게 된다.
from sqlalchemy import text
class UserDao:
def __init__(self, database):
self.db = database
def insert_user(self, user):
return self.db.execute(text("""
INSERT INTO users (
name,
email,
profile,
hashed_password
) VALUES (
:name,
:email,
:profile,
:password
)
"""), user).lastrowid
def get_user_id_and_password(self, email):
row = self.db.execute(text("""
SELECT
id,
hashed_password
FROM users
WHERE email = :email
"""), {'email' : email}).fetchone()
return {
'id' : row['id'],
'hashed_password' : row['hashed_password']
} if row else None
def insert_follow(self, user_id, follow_id):
return self.db.execute(text("""
INSERT INTO users_follow_list (
user_id,
follow_user_id
) VALUES (
:id,
:follow
)
"""), {
'id' : user_id,
'follow' : follow_id
}).rowcount
def insert_unfollow(self, user_id, unfollow_id):
return self.db.execute(text("""
DELETE FROM users_follow_list
WHERE user_id = :id
AND follow_user_id = :unfollow
"""), {
'id' : user_id,
'unfollow' : unfollow_id
}).rowcount이렇게 api 레이어드 아키텍처에 대해 살펴보면서, 왜 코드의 구조를 체계적으로 구현하는 것이 중요한지 좀 더 구체적으로 배울 수 있었다.
무엇보다 처음과 같이 한덩어리로 되어있는 코드에서는 데이터를 받아와서 프론트로 다시 넘겨주는 부분(presentation layer)과 비즈니스 로직을 처리하는 부분(business layer), 데이터베이스에 접근하는 부분(persistence layer)이 모두 한묶음으로 되어있어서 코드를 읽고 파악하기가 어렵고 시간이 많이 걸렸었는데,
이렇게 각 레이어 별로 코드가 분리되고 나니 각 파일들에서 코드들이 하는 역할이 보다 명확하고 간결해져서 코드를 보고 이해하기가 훨씬 수월해졌던 점이 너무 신기하고 재미있었다.
곧 학교에서 시작하는 프로젝트가 있는데, 거기에도 이 레이어드 아키텍처를 적용해서 작업해보고 싶다는 생각이 들었다.