13. SQLAlchemy를 이용하여 API와 DB 연결하기

  본격적으로 miniter API의 회원가입, tweet, , follow, timeline 엔드포인트들을 SQLAlchemy를 이용하여 DB에 연결해 보도록 하겠다. 이로써 데이터를 저장하고 불러오는 것이 가능해 진다. 

 

  먼저 앞서 작성한 app.py 파일을 수정하여 config.py 파일의 설정 정보를 읽어들여 DB에 연결하도록 한다.

 

from flask      import Flask, request, jsonify, current_app
from flask.json import JSONEncoder
from sqlalchemy import create_engine, text 


class CustomJSONEncoder(JSONEncoder):
    def default(self, obj):
        if isinstance(obj, set):
            return list(obj)

        return JSONEncoder.default(self, obj)


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.excute(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.excute(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
    """), {
        'user_id' : user_id
    }).fetchall()

    return [{
        'user_id' : tweet['user_id'],
        'tweet'   : tweet['tweet']
    } for tweet in timeline]

def create_app(test_config = None):
    app = Flask(__name__)

    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_id = insert_user(new_user)
        new_user    = get_user(new_user_id)

        return jsonify(new_user)
    
    @app.route("/tweet", methods=['POST'])
    def tweet():
        user_tweet = request.json
        tweet      = user_tweet['tweet']

        if len(tweet) > 300:
            return '300자를 초과했습니다.', 400

        insert_tweet(user_tweet)

        return '', 200

    @app.route("/follow", methods=['POST'])
    def follow():
        payload = request.json
        insert_follow(payload)

        return '', 200

    @app.route("/unfollow", methods=['POST'])
    def unfollow():
        payload = request.json
        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)
        })
    
    return app

 

  아래쪽에서 app.route 로 구현한 엔드포인트들이 각각 필요로 하는 함수들을 위쪽의 def 함수를 호출하는 형태이다. 

예를들어 sign-up 엔드포인트에서 new_user_id 는 위쪽의 insert_user 를 호출하는 식이다.

 

  다른말로 'return create_app' function을 Flask 가 자동으로 인식해서 Flask instance 를 코드상에서 찾아 실행하는 것이다.

 

  이제 제대로 작동하는지 회원가입, tweet, follow, timeline에 HTTP 요청을 보내 테스트 해본다. 먼저 아래의 명령어로 작성한 코드를 활성화 시킨다. 

 

FLASK_ENV=development FKAS_APP=app.py flask run

 

  그 후 HTTP 요청을 보내본다. 

 

● 회원가입

 

● tweet

 

 

● follow

 

 

● timeline