[Flask]갑자기 만들어보는 Flask 게시판

목차

 

[Flask]갑자기 만들어보는 Flask 서버 튜토리얼

 

약 40일 정도 시간이 지났지만, 뜬금없이 진도를 나가보려고 한다.

 

회사에서 플라스크와 웹플럭스를 이용한 데이터 파이프라인을 구축 중이라

 

한땀한땀 플라스트 앱을 만들기는 했으나 아무래도 주먹구구식으로 만든 거라

 

기초적인 이해가 더 필요하다고 느꼈기 때문이다.

 

기본적인 설정은 지난 글에서 했으니, 이번 글에선 바로 MVC 패턴을 적용해 기초적인 게시판을 만들어본다.

 

SQLAlchemy

 

먼저 터미널에 아래와 같은 명령어를 입력해 SQLAlchemy를 설치해준다.

source myenv/bin/activate

pip install Flask-SQLAlchemy

SQLAlchemy는 파이썬을 위한 ORM 라이브러리로, DB와의 상호작용을 좀 더 Pythonic하게 만들어준다고 한다.

 

예를 들자면 스프링 부트에서처럼 테이블을 클래스로, 레코드(로우)를 클래스 인스턴스로 표현할 수 있다.

 

추가로 이번 게시판 구현에서는 장고와 마찬가지로 SQLite를 사용할 계획이다.

 

Architecture

 

계속해서 MVC 패턴 적용을 위한 폴더 구조를 작성하자.

 

구현을 끝낸 트리구조는 대략 아래와 같은 모습이 된다.

 

 

이 글에선 게시판 글 등록 등의 로직 전에 회원 관련 모듈을 완성할 계획이다.

 

먼저, 폴더를 만들고 그 안에 비어있는 __init__.py를 만들어두고 시작하자.

 

db_config.py

 

가장 먼저 디비와의 연결 설정 모듈을 구성한다.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URI = "sqlite:///board_practice.db"
engine = create_engine(DATABASE_URI)
Session = sessionmaker(bind=engine)

Base = declarative_base()

한 줄씩 뜯어보자.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

첫 번째 import는 SQLAlchemy에서 디비 엔진을 생성하기 위한 create_engine을 가져온다.

 

두 번째 import는 ORM 모델을 생성하기 위한 declarative_base를 가져온다.

 

세 번째 import는 디비 세션을 관리하기 위한 sessionmaker를 가져온다.

DATABASE_URI = "sqlite:///board_practice.db"
engine = create_engine(DATABASE_URI)
Session = sessionmaker(bind=engine)

DATABASE_URI의 이름은 임의로 설정하면 된다. 나는 board_practice로 해주었다.

 

engine 줄은 설정한 디비 URI를 기반으로 엔진 객체를 생성한다.

 

Session 줄은 엔진에 바인딩된 세션 메이커를 생성한다. 이를 통해 디비와의 세션을 생성하고 관리할 수 있다.

Base = declarative_base()

모델 클래스를 생성할 때 상속받을 기본 클래스를 만든다.

 

앞으로 모든 모델 클래스는 위 클래스를 상속받아야 한다.

 

user_model.py

 

다음으로 사용자 정보를 담고 디비와 연결할 모델 파일을 구성한다.

 

여기서는 유저 아이디를 이메일로 대신하고, 비밀번호와 닉네임을 가진 간단한 사용자 모델을 구현한다.

from sqlalchemy import Column, Integer, String, Sequence
from ..config.db_config import Base


class User(Base):
    __tablename__ = "users"
    seq = Column(Integer, Sequence("user_seq"), primary_key=True)
    email = Column(String(50), unique=True)
    password = Column(String(50))
    nickname = Column(String(50))

    def __init__(self, email, password, nickname):
        self.email = email
        self.password = password
        self.nickname = nickname

    def to_dict(self):
        return {"email": self.email, "password": self.password, "nickname": self.nickname}

먼저 사용자 모듈에 유저 클래스를 구현했다.

 

코드의 의미를 뜯어보면 아래와 같다.

from sqlalchemy import Column, Integer, String, Sequence
from ..config.db_config import Base

첫 번째 import는 컬럼을 정의하기 위한 클래스와 함수를 가져온다.

 

두 번째 import는 위의 db_config에서 생성한 기본 클래스 Base를 가져온다.

 

위에도 언급했지만 Base는 SQLAlchemy를 사용하는 모든 모델 클래스가 상속받아야 하는 기본 클래스라고 보면 된다.

class User(Base):
    __tablename__ = "users"
    seq = Column(Integer, Sequence("user_seq"), primary_key=True)
    email = Column(String(50), unique=True)
    password = Column(String(50))
    nickname = Column(String(50))

이어서 Base 클래스를 상속받아 User 클래스를 정의한다.

 

먼저 테이블 이름을 정하고 각 컬럼의 요구사항에 맞춰 자료형을 명시적으로 설정한다.

    def __init__(self, email, password, nickname):
        self.email = email
        self.password = password
        self.nickname = nickname

위 함수는 User 클래스의 생성자 역할을 한다.

 

스프링에서와 마찬가지고 객체 생성시 이메일, 비밀번호, 닉네임을 초기화하는 기능을 한다.

    def to_dict(self):
        return {"email": self.email, "password": self.password, "nickname": self.nickname}

이 함수는 User 인스턴스의 속성을 딕셔너리로 반환하는 역할을 한다.

 

이를 통해 객체의 데이터를 JSON 등으로 쉽게 직렬화 할 수 있게 된다.

 

딕셔너리 형식은 return에 정의된 형태와 같다.

 

auth_service.py

 

계속해서 회원가입과 로그인 로직이 포함된 서비스 레이어를 작성하자.

 

먼저 아래와 같이 가상환경을 활성화 한 뒤 필요한 패키지를 설치한다.

source myenv/bin/activate
pip install PyJWT bcrypt email-validator

계속해서 아래와 같이 코드를 작성한다.

import jwt
import bcrypt
from datetime import datetime, timedelta
from email_validator import validate_email, EmailNotValidError
from ..models.user_model import User
from ..config.db_config import Session

SECRET_KEY = "thisisthesecretkeyforpracticingflask"


def register_user(email, password, pass_repeat, nickname):
    if password != pass_repeat:
        raise ValueError("Password and repeat password must match.")

    session = Session()
    
    existing_user = session.query(User).filter_by(email=email).first()
    if existing_user:
        raise ValueError("Email already registered.")

    try:
        validate_email(email)
    except EmailNotValidError as e:
        raise ValueError(str(e))

    hashed_password = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())

    new_user = User(email=email, password=hashed_password.decode("utf-8"), nickname=nickname)
    session.add(new_user)
    session.commit()


def login_user(email, password):
    session = Session()

    user = session.query(User).filter_by(email=email).first()

    if user:
        if bcrypt.checkpw(password.encode("utf-8"), user.password.encode("utf-8")):
            expiration_time = datetime.utcnow() + timedelta(days=1)
            payload = {"email": user.email, "exp": expiration_time}
            token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")

            return {"email": user.email, "token": token}
        else:
            raise ValueError("Incorrect Password")
    else:
        raise ValueError("User Not Found")

코드의 의미를 블록별로 뜯어보자.

import jwt
import bcrypt
from datetime import datetime, timedelta
from email_validator import validate_email, EmailNotValidError
from ..models.user_model import User
from ..config.db_config import Session

SECRET_KEY = "thisisthesecretkeyforpracticingflask"

먼저 필요한 모듈과 라이브러리를 임포트하고, 유저 모델과 DB 세션을 임포트한다.

 

이후엔 JWT 생성에 사용할 비밀 키를 설정한다.

def register_user(email, password, pass_repeat, nickname):
    if password != pass_repeat:
        raise ValueError("Password and repeat password must match.")

    session = Session()

    existing_user = session.query(User).filter_by(email=email).first()
    if existing_user:
        raise ValueError("Email already registered.")

    try:
        validate_email(email)
    except EmailNotValidError as e:
        raise ValueError(str(e))

    hashed_password = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())

    new_user = User(email=email, password=hashed_password.decode("utf-8"), nickname=nickname)
    session.add(new_user)
    session.commit()

위의 함수는 회원가입을 처리한다.

 

가장 먼저 입력된 비밀번호와 재입력한 비밀번호가 같은지 확인한 후, 다르다면 예외를 발생시킨다.

 

이후 SQLAlchemy 세션을 생성한 뒤 이미 가입된 메일이 있는지 확인한 후, 있다면 예외를 발생시킨다.

 

계속해서 try-except 블록을 사용해 이메일 유효성을 검사한다.

 

그 다음으로 비밀번호를 해싱해 hashed_password에 할당한 뒤,

 

이를 바탕으로 새로운 사용자 객체를 만들어 세션에 추가한 뒤 커밋으로 변경사항을 데이터베이스에 저장한다.

 

회원가입을 하면 바로 로그인을 하도록 구현하려다 이런 식으로 간단하게만 해보았다.

def login_user(email, password):
    session = Session()

    user = session.query(User).filter_by(email=email).first()

    if user:
        if bcrypt.checkpw(password.encode("utf-8"), user.password.encode("utf-8")):
            expiration_time = datetime.utcnow() + timedelta(days=1)
            payload = {"email": user.email, "exp": expiration_time}
            token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")

            return {"email": user.email, "token": token}
        else:
            raise ValueError("Incorrect Password")
    else:
        raise ValueError("User Not Found")

계속해서 로그인 로직을 처리하는 부분이다. 역시 세션을 생성하며 시작한다.

 

다음 줄에서는 디비에서 이메일을 이용해 사용자를 찾는다.

 

이어서 만약 사용자가 존재한다면 비밀번호를 검증하고, 검증 정보를 바탕으로 JWT를 발급해 이메일과 함께 반환한다.

 

나머지 부분은 예외 처리에 관한 부분이다.

 

auth_controller.py

 

계속해서 이 모든 것을 반영해 컨트롤러 파일을 구성한다. 노드 개발 때와는 다르게

 

데코레이터에서 라우팅 주소를 할당하는 부분이 스프링 MVC와 비슷해 흥미로웠다.

 

코드는 대략 아래와 같이 구현할 수 있다.

from flask import Blueprint, request, jsonify
from ..services.auth_service import register_user, login_user

auth_app = Blueprint("auth_app", __name__)


@auth_app.route("/register", methods=["POST"])
def register():
    try:
        data = request.json
        email = data["email"]
        password = data["password"]
        pass_repeat = data["pass_repeat"]
        nickname = data["nickname"]

        register_user(email, password, pass_repeat, nickname)

        return jsonify({"message": "User Registered Successfully"}), 201

    except ValueError as e:
        return jsonify({"message": str(e)}), 400


@auth_app.route("/login", methods=["POST"])
def login():
    try:
        data = request.json
        email = data["email"]
        password = data["password"]

        result = login_user(email, password)

        return jsonify(result), 200

    except ValueError as e:
        return jsonify({"message": str(e)}), 401

코드를 짧게 뜯어보자면

from flask import Blueprint, request, jsonify
from ..services.auth_service import register_user, login_user

auth_app = Blueprint("auth_app", __name__)

필요한 의존성을 가져온 뒤 Flask의 Bluprint를 이용해 auth_app을 생성한다.

 

이는 라우터의 모듈화에 필요한 작업이다.

@auth_app.route("/register", methods=["POST"])

회원가입을 처리하는 라우트를 설정하고, HTTP Method로 POST만 허용한다.

def register():
    try:
        data = request.json
        email = data["email"]
        password = data["password"]
        pass_repeat = data["pass_repeat"]
        nickname = data["nickname"]

        register_user(email, password, pass_repeat, nickname)

        return jsonify({"message": "User Registered Successfully"}), 201

    except ValueError as e:
        return jsonify({"message": str(e)}), 400

JSON으로 입력된 data에서 각각 값을 읽어 변수에 할당한 뒤 서비스 계층의 함수를 호출한다.

 

성공한다면 201 Created 코드와 함께 정해진 문자열을 출력하고,

 

실패한다면 400 Bad Request와 함께 에러메시지를 출력한다.

@auth_app.route("/login", methods=["POST"])
def login():
    try:
        data = request.json
        email = data["email"]
        password = data["password"]

        result = login_user(email, password)

        return jsonify(result), 200

    except ValueError as e:
        return jsonify({"message": str(e)}), 401

위 함수 블럭과 본질적으로 큰 차이가 없어 생략한다.

 

성공했을 때의 리턴값을 서비스 계층에서 받아 반환한다는 점이 특징이다.

 

app.py

 

마지막으로 이 모든 것을 모아 실행 파일을 구성한다.

 

이전 글의 app.py를 가져와 아래와 같이 구현한다.

from flask import Flask
from .controllers.auth_controller import auth_app
from .config.db_config import engine, Base
from .models.user_model import User

Base.metadata.create_all(engine)

app = Flask(__name__)

app.register_blueprint(auth_app, url_prefix="/api/auth")


@app.route("/")
def home():
    return "Hello, World!"


if __name__ == "__main__":
    app.run()

하던대로 코드를 부분별로 뜯어보자.

from flask import Flask
from .controllers.auth_controller import auth_app
from .config.db_config import engine, Base
from .models.user_model import User

필요한 모듈을 가져온다. 여기서 User를 가져오는 이유는 애플리케이션 실행시

 

User를 먼저 메모리에 올려 테이블을 생성하기 위한 것이다.

Base.metadata.create_all(engine)

app = Flask(__name__)

app.register_blueprint(auth_app, url_prefix="/api/auth")

계속해서 첫 줄에서는 메모리에 올라온 모델을 기반으로 테이블을 생성한다.

 

이때 테이블이 이미 생성되어 있는 경우엔 새로 생성하지 않고 넘어간다고 한다.

 

이어서 Flask앱을 초기화해 app에 할당하고,

 

컨트롤러 파일에서 정의한 'auth_app' Blueprint를 Flask 앱에 등록 및 접두사 설정을 해준다.

@app.route("/")
def home():
    return "Hello, World!"


if __name__ == "__main__":
    app.run()

이 부분은 지난 글에 있으니 생략한다.

 

이렇게 한 뒤에 터미널에서 flask run으로 서버를 올린 뒤 요청을 보내보면 아래와 같이 응답이 오는 것을 확인할 수 있다.