이전 블로그에서 알아보았듯이 Layered Architecture는 3가지 레이어로 나눌 수 있다.
이번에는 우리가 만들었던 Minitter를 레이어드 패턴 아키텍처로 구성하고자 한다
Layered Architecture
레이어드 아키텍쳐 구조는 다음과 같이 구성하면 된다.
$ mkdir {view,service,model}
$ tree
api
|_ view
|_ service
|_ model
|_ app.pymkdir: 디렉토리를 생성하는 명령어로{ }괄호 안에 생성하고자 하는 디렉토리 여러개를 넣어 생성할 수 있다.
view(presentation layer)
- view.py에서 presentation layer에 대한 내용을 구현한다.
- API의 엔드포인트에 대한 부분을 작성한다.
- 많은 내용이 필요하지 않음으로
__init__.py를 만들어서 API의 엔드포인트를 구현해도 좋다.
def create_endpoints(app, service): # 1)
user_service = service.user_service # 2)
@app.route("/sign-up", methods=['POST']
def sign_up():
new_user = request.json
new_user_id = user_service.create_new_user(new_user) # 3)
new_user = user_service.get_user(new_user_id) # 4)
return jsonify(new_user)1) create_endpoints()라는 함수를 만들어서 이 함수 안에 엔드포인트를 생성한다.
create_endpoints()함수는 2개의 인자를 받는데, 먼저 Flask 클래스를 생성해서 만든 app을 인자로 받고,
business layer의 코드를 가지고 있는 service를 인자로 받는다.
2) service인자에서 UserService 객체를 user_service변수에 읽어 들인다.
UserService 클래스는 사용자 관련 비즈니스 로직을 구현하는 클래스이다.
3) user_service의 create_new_user함수를 호출해서 사용자를 생성한다.
즉, business layer의 코드로 위임해서 비즈니스 로직을 해결하는 것이다.
4) 3)에서 생성한 새로운 사용자의 아이디를 사용하여 user_service의 get_user함수를 호출해서 사용자 정보를 읽어 들인다.
service(business layer)
- service.py에서 business layer에 대한 내용을 구현한다.
- Minitter의 API의 경우 business layer를 크게 2개로 나눌 수 있다. (user, tweet)
- user : UserService- tweet : TweetService
- business layer는 persistance layer에 의존하고 있다.
- UserService 클래스는 Persistance layer의
UserDao에 의존한다. - DAO는 여기서 Data Access Object의 약어로, 데이터 접속을 담당하는 객체이다.
class UserService: # 1)
def __init__(self, user_dao): # 2)
self.user_dao = user_dao
def create_new_user(self, new_user): # 3)
new_user['password'] = bcrypt.hashpw(new_user['password'].encode('UTF-8'), # 4)
bcyrpt.gensalt()
)
new_user_id = self.user_dao.insert_user(new_user) # 5)
return new_user_id1) UserService 클래스를 정의한다.
UserService 클래스가 business layer의 사용자 관련부분을 담당한다.
2) UserService의 __init__메소드에 user_dao를 인자로 받는다.
user_dao가 persistance layer의 사용자 부분을 담당하는 객체다. 즉, users 테이블을 담당하는 persistnace layer이다.
3) create_new_user()함수를 통해 새로 생성된 사용자를 처리하는 비즈니스 로직을 구현한다.
여기서 new_user인자는 persistance layer를 통해 받는다.
4) presentation layer를 통해 받는 new_user의 비밀번호를 단방향 암호화한다.
5) 4)에서 비밀번호를 암호화한 사용자 정보를 user_dao를 통해 persistance layer에 넘겨주어서 데이터베이스에 사용자를 생성하도록 한다.
model(persistance layer)
- model.py에서 persistnace layer에 대한 내용을 구현한다.
- service.py와 마찬가지로 model.py에서도 user, tweet에 대한 두가지 DAO 클래스를 구현한다.
- user : UserDao- twwet : TweetDao
UserDao클래스는UserService에서만 사용하고, 마찬가지로TweetDao는TweetService에서만 사용한다.
class UserDao: # 1)
def __init__(self, database): # 2)
self.db = database
def insert_user(self, user):
return self.db.execute(text(""" # 3)
INSERT INTO users (
name,
email,
profile,
hashed_password
) VALUES (
:name,
:email,
:profile,
:password
)
"""), user).lastrowid1) UserDao 클래스를 정의한다.
2) UserDao 클래스의 __init__ 메소드에 database를 인자로 받는다.
database는 sqlalchemy의 create_engine 함수를 통해서 생성한 Engine 객체다.
클래스 안에서 직접 데이터베이스 연결을 생성하는 것이 아니라 외부에서 dependency로 받는 이유는 몇가지 있지만, 주된 이유는 unit test를 구현하기 위해서 이다.
3) 2)dptj dependency로 받은 database를 통하여 데이터베이스에 사용자 데이터를 INSERT 한다.
전체연결과 app.py
- service, view, model을 모두 구현하고 app.py를 통해 모두 연결을 해줘야한다.
- 각 레이어의 폴더를 통해 파일과 클래스 및 메소드를 import 할 수 있지만, 각 폴더에
__init__.py를 생성하여 간단하게 import 할 수 있다
# service/__init__.py
from .user_service import UserService
from .tweet_service import TweetService
__all__ = [ # 1)
'UserService',
'TweetService'
]1) __all__에 UserService와 TweetService를 지정해주어서 service모듈에서 한번에 둘다 import 할 수 있도록 해준다.
- app.py에서는 각 Layer의 클래스나 모듈을 import하고, Flask 어플리케이션을 생성하여 주는 역할을 한다.
# app.py
import config
from flask import Flask
from sqlalchemy import create_engine
from flask_cors import CORS
from model import UserDao, TweetDao # 1)
from service import UserService, TweetService # 2)
from view import create_endpoints # 3)
class Services: # 4)
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) # 5)
tweet_dao = TweetDao(database)
## Business Layer
services = Services # 6)
services.user_service = UserService(user_dao, config)
services.tweet_service = TweetService(tweet_dao)
## 엔드포인트들을 생성
create_endpoints(app, services) # 7)
return app # 8)1) persistance layer인 DAO 클래스들을 import한다.
2) business layer인 service 클래스들을 import한다.
3) presentation layer인 view 모듈의 create_endpoints함수를 import한다.
4) service 클래스들을 담고 있는 클래스이다.6)business layer를 생성할때, 7)엔드포인들을 생성할때 사용한다.
5) persistance layer인 DAO 클래스들 부터 생성한다.
6) business layer인 service 클래스들을 생성한다.
4)의 Service클래스를 service에 저장한다.
5)에서 생성한 DAO 클래스들을 인자로 넘겨준다. business layer와 persistance layer의 의존관계를 볼 수 있다.
7) presentation layer에 해당하는 view의 엔드포인트들을 생성한다.
6)에서 생성된 service 클래스들을 인자로 넘겨준다. presentaion layer와 business layer의 의존관계를 볼 수 있다.
8) persistance layer, business layer, presentation layer가 모두 연결된 Flask 어플리케이션을 리턴해준다.
전체 코드
# views/__init__.py
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
})# serivce/user_service.py
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
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