SQLAlchemy

SQLAlchemy

DB와 서버를 연동시키는 파이썬 라이브러리이며
관계형 데이터베이스의 테이블들을 파이썬 클래스로 표현하고 저장, 읽기, 업데이트 등을 가능하게 해주는 ORM

sqlalchemy와 mysql용 DB를 사용하기 위한 api인 mysql-connector-python 설치

SQLAlchemy를 사용하여 API와 데이터베이스 연결

config.py로 새로운 파일 생성 및 데이터 베이스의 연결 정보 저장

db = {
    "user": "root", #데이터 베이스에 접속할 아이디
    "password": "wlsdn12450", # 유저 비밀번호
    "host": "localhost", #접속할 데이터베이스 주소, 외부 서버에 설치되어있다면 해당 서버 주소
    "port": 3306,#접속할 데이터베이스 포트 넘버, 주로 3306 포트
    "database": "fmms" #실제 사용할 데이터 페이스 이름
}
DB_URL = f"mysql+mysqlconnector://{db['user']}:{db['password']}@{db['host']}:{db['port']}/{db['database']}?charset=utf8"

config(설정) 파일을 따로 만드는 이유

• 설정 정보를 따로 관리함으로써 민감한 개인 접속 정보 노출 최소화
• 각 환경과 설정에 맍는 설정 파일 적용 가능
  - .gitignore 파일에 config 파일을 지정해놓음으로써 config.py 파일이 git 레포지토리에 포함되지 않아 개인정보 노출을 막고, 각 개발 호스트 혹은 서버에 맞는 config.py를 생성하도록 함으로써 각 환경에 적합한 설정 적용 가능

sqlalchemy의 create_engine을 사용하여 데이터베이스와 연결

from sqlalchemy import create_engine, text
#플라스크가 create_app 이라는 이름의 함수를 자동으로 팩토리함수로 인식해서 해당 함수를 통해 flask 실행
def create_app(test_config=None):  
    app = Flask(__name__)

    if not test_config:
        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)
    #생성된 engine 객체를 flask객체에 저장함으로써 create_app 함수 외부에서도 데이터베이스 사용 가능
    app.database = database  

    return app  

회원가입 엔드포인트

1. current_app을 import, current_app은 create_app function에서 생선한 app을 create_app function 외부에서도 사용할 수 있게 해줌.

2. INSERT SQL 구문을 DB에 실행하고, 생성된 사용자의 id를 lastrowid를 통해 읽어들이는 function

3. SELECT SQL 구문을 DB에 실행, DB에서 paramter로 전달된 사용자 id에 레코드 (record)를 function

4. sign-up endpoint를 2, 3에서 정의한 function을 이용하여 정의

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

...

def insert_user(user):  # 2
    return current_app.database.execute(text("""
    INSERT INTO users (name, email, profile, hashed_password) VALUES (:name, :email, :profile, :password)
    """), user).lastrowid


def get_user(user_id):  # 3
    user = current_app.database.execute(text("""
    SELECT id, name, email, profile FROM users WHERE id =:user_id
    """), {"user_id": user_id}).fetchone()
    return dict(id=user["id"], name=user["name"], email=user["email"], profile=user["profile"])


def create_app(test_config=None):

    ...

    @app.route("/sign-up", methods=["POST"])  # 4
    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)

    return app  # Flask instance를 return

서버에서 테스트

http -v POST http://localhost:5000/sign-up \
name=김보섭 \
email=bsk0130@gmail.com \
password=test1234 \
profile="aisolab, self-motivated deep learning engineer"

실행결과화면

tweet 엔드포인트

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

...

def insert_tweet(user_tweet):  # 2
    current_app.database.execute(text("""
    INSERT INTO tweets (user_id, tweet) VALUES (:id, :tweet)
    """), user_tweet).rowcount

...

def create_app(test_config=None):

    ...

    @app.route("/tweet", methods=["POST"])  # 3
    def tweet():
        user_tweet = request.json
        tweet = user_tweet["tweet"]

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

        insert_tweet(user_tweet)

        return "", 200

    return app  # Flask instance를 return

실행 결과

timeline 엔드포인트

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

...

def get_timeline(user_id):  # 2
    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 create_app(test_config=None):

    ...

    @app.route("/timeline/<int:user_id>", methods=["GET"])  # 3
    def timeline(user_id):
        return jsonify(dict(user_id=user_id, timeline=get_timeline(user_id)))

    return app  # Flask instance를 return

실행결과

follow, unfollow 엔드포인트

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

...

def insert_follow(user_follow):
    current_app.database.execute(text("""
    INSERT INTO users_follow_list (user_id, follow_user_id) VALUES (:id, :follow)
    """), user_follow)


def insert_unfollow(user_unfollow):
    current_app.database.execute(text("""
    DELETE FROM users_follow_list
    WHERE user_id =:id
    AND follow_user_id = :unfollow
    """), user_unfollow)

...

def create_app(test_config=None):

    ...

    @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

    return app  # Flask instance를 return

실행결과