코드 구조의 중요성
레이어드 패턴
unit test
View Unit Test
레이어트 아키텍처 적용하기
1. view (presentation layer)
import jwt
from flask import request, jsonify, current_app, Response, g
from flask.json import JSONEncoder
from functools import wraps
## 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)
#########################################################
# Decorators
#########################################################
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
@app.route('/tweet', methods=['POST'])
@login_required
def tweet():
user_tweet = request.json
tweet = user_tweet['tweet']
user_id = g.user_id
result = tweet_service.tweet(user_id, tweet)
if result is None:
return '300자를 초과했습니다', 400
return '', 200
@app.route('/follow', methods=['POST'])
@login_required
def follow():
payload = request.json
user_id = g.user_id
follow_id = payload['follow']
user_service.follow(user_id, follow_id)
return '', 200
@app.route('/unfollow', methods=['POST'])
@login_required
def unfollow():
payload = request.json
user_id = g.user_id
unfollow_id = payload['unfollow']
user_service.unfollow(user_id, unfollow_id)
return '', 200
@app.route('/timeline/<int:user_id>', methods=['GET'])
def timeline(user_id):
timeline = tweet_service.get_timeline(user_id)
return jsonify({
'user_id' : user_id,
'timeline' : timeline
})
@app.route('/timeline', methods=['GET'])
@login_required
def user_timeline():
timeline = tweet_service.get_timeline(g.user_id)
return jsonify({
'user_id' : user_id,
'timeline' : timeline
})2. service (business layer)
from .user_service import UserService
from .tweet_service import TweetService
__all__ = [
'UserService',
'TweetService'
]- User service
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)- Tweet Service
class TweetService:
def __init__(self, tweet_dao):
self.tweet_dao = tweet_dao
def tweet(self, user_id, tweet):
if len(tweet) > 300:
return None
return self.tweet_dao.insert_tweet(user_id, tweet)
def get_timeline(self, user_id):
return self.tweet_dao.get_timeline(user_id)3. model (persistence layer)
from .user_dao import UserDao
from .tweet_dao import TweetDao
__all__ = [
'UserDao',
'TweetDao'
]- User Dao
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- tweet dao
from sqlalchemy import text
class TweetDao:
def __init__(self, database):
self.db = database
def insert_tweet(self, user_id, tweet):
return self.db.execute(text("""
INSERT INTO tweets (
user_id,
tweet
) VALUES (
:id,
:tweet
)
"""), {
'id' : user_id,
'tweet' : tweet
}).rowcount
def get_timeline(self, user_id):
timeline = self.db.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]app.py 파일
import config
from flask import Flask
from sqlalchemy import create_engine
from flask_cors import CORS
from model import UserDao, TweetDao
from service import UserService, TweetService
from view import create_endpoints
class Services:
pass
################################
# Create App
################################
def create_app(test_config = None):
app = Flask(__name__)
CORS(app)
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)
## Persistenace Layer
user_dao = UserDao(database)
tweet_dao = TweetDao(database)
## Business Layer
services = Services
services.user_service = UserService(user_dao, config)
services.tweet_service = TweetService(tweet_dao)
## 엔드포인트들을 생성
create_endpoints(app, services)
return app