[AWS] Python Flask + AWS AuroraDB 연동하기

RDS(Relational Database Service)

AWS 환경에서 관계형 데이터베이스를 제공하는 서비스이다. PostgreSQL, MySQL, MariaDB, ORACLE 등 다양한 벤더의 데이터베이스를 마이그레이션하거나 복제하는 것을 지원한다.

본 포스트에서는 AWS RDS(AuroraDB)와 Flask의 웹 애플리케이션을 연동하는 것을 목표로 한다.

개발 스택은 다음과 같다.

[개발 환경]
Python 3.7.3
Flask
[AWS 서비스]
AWS Codecommit(CI/CD)
Elastic Beanstalk(앱 배포)
AuroraDB

1. RDS 시작하기

Create database

베어메탈에 OS를 구성하고 DB 엔진을 설치하고 운영에 걸리는 시간에 비하면 빠른 시간 내에 DB의 설정이 완료된다. 다만, DB instance class를 설정할 때 과도하게 CPU 스펙과 메모리 양을 높은 것을 선택하면, DB는 해당 스펙으로 24시간 동작하게 되므로 요금이 과도 청구될 수 있으니 용도에 맞는 클래스를 선택하여 사용하도록 한다.

연결 설정이다.

현재는 개발환경이며 외부에서 DB 접속 도구를 이용한 접속이 필요하므로 Public access에 허용으로 두었다. 이 설정은 변경이 불가능하므로 운영 DB의 경우 No를 선택하여 VPC 내부에 위치한 EC2에서만 접속이 가능하도록 설정한다.

VPC 보안 그룹 설정까지 마치면 DB 설정이 끝나게 된다.

설정을 마치면 DB가 동작중인 것을 확인할 수 있다. 외부 접속을 위한 엔드포인트를 확인하기 위해 DB 식별자 명을 클릭하여 접속한다.

Endpoint 확인

2. DB 연결 확인

DB 접속 도구로는 DBeaver를 사용한다. 새 연결을 생성한다.

연결 테스트

Server Host : DB 생성시 확인했던 엔드포인트 주소 입력
Port : 설정한 포트(기본포트 : 3306)

Authentication
Username : 기본 DB 사용자 명
Password : 패스워드

3. DB & TABLE 생성

CREATE DATABASE sys default CHARACTER SET UTF8;
use sys;

sys DB를 생성하고 사용한다.

CREATE TABLE users(
	id INT PRIMARY KEY AUTO_INCREMENT,
	name VARCHAR(32) NOT NULL,
	email VARCHAR(100),
	username VARCHAR(50),
	password VARCHAR(100)
);

회원가입한 사용자 정보를 담을 테이블을 간단하게 생성했다. DB 준비과정은 끝이 났다.

4. Flask 웹 애플리케이션

from flask import Flask, render_template, flash, redirect, url_for, session, request, logging
from flask_mysqldb import MySQL

SQLALCHEMY_ECHO = True
application = Flask(__name__)
application.secret_key='this_is_not_real_secret'
application.config['SESSION_TYPE'] = 'memcached'
application.config['SECRET_KEY'] = 'super secret key'

# Config MySQL
application.config['MYSQL_HOST'] = 'auro-database-1-instance-1.aaaaaaaaaa.us-west-1.rds.amazonaws.com'
application.config['MYSQL_USER'] = 'admin'
application.config['MYSQL_PASSWORD'] = 'this_is_not_real_password'
application.config['MYSQL_DB'] = 'sys'
application.config['MYSQL_CURSORCLASS'] = 'DictCursor'
# init MYSQL
mysql = MySQL(application)

MySQL을 사용하기 위한 설정 값들을 준비한다. DB에 연결하기 위한 HOST 정보, 사용자 및 패스워드, DB정보까지 입력해 주면 SQL에 연결하기 위한 준비는 완료되었다.

@application.route('/')
def index():
    return render_template('home.html')
    
# Register Form Class
class RegisterForm(Form):
    name = StringField('Name', [validators.Length(min=1, max=50)])
    username = StringField('Username', [validators.Length(min=4, max=25)])
    email = StringField('Email', [validators.Length(min=6, max=50)])
    password = PasswordField('Password', [
        validators.DataRequired(),
        validators.EqualTo('confirm', message='Passwords do not match')
    ])
    confirm = PasswordField('Confirm Password')


# User Register
@application.route('/register', methods=['GET', 'POST'])
def register():
    form = RegisterForm(request.form)
    if request.method == 'POST' and form.validate():
        name = form.name.data
        email = form.email.data
        username = form.username.data
        password = sha256_crypt.encrypt(str(form.password.data))

        # Create cursor
        cur = mysql.connection.cursor()

        # Execute query
        cur.execute("INSERT INTO users(name, email, username, password) VALUES(%s, %s, %s, %s)", (name, email, username, password))

        # Commit to DB
        mysql.connection.commit()

        # Close connection
        cur.close()

        flash('You are now registered and can log in', 'success')

        return redirect(url_for('login'))
    return render_template('register.html', form=form)


# User login
@application.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # Get Form Fields
        username = request.form['username']
        password_candidate = request.form['password']

        # Create cursor
        cur = mysql.connection.cursor()
        cmd = "SELECT * FROM users WHERE username = '%s'" % username
        result = cur.execute(cmd)
        print(result)
        # Get user by username
        # SAFE TRANSACTION
        # result = cur.execute("SELECT * FROM users WHERE username = %s", [username])

        if result > 0:
            # Get stored hash
            data = cur.fetchone()
            password = data['password']

            # Compare Passwords
            if sha256_crypt.verify(password_candidate, password):
                # Passed
                session['logged_in'] = True
                session['username'] = username

                flash('You are now logged in', 'success')
                return redirect(url_for('dashboard'))
            else:
                error = 'Invalid login'
                return render_template('login.html', error=error)
            # Close connection
            cur.close()
        else:
            error = 'Username not found'
            return render_template('login.html', error=error)

    return render_template('login.html')

# Check if user logged in
def is_logged_in(f):
    @wraps(f)
    def wrap(*args, **kwargs):
        if 'logged_in' in session:
            return f(*args, **kwargs)
        else:
            flash('Unauthorized, Please login', 'danger')
            return redirect(url_for('login'))
    return wrap

# Logout
@application.route('/logout')
@is_logged_in
def logout():
    session.clear()
    flash('You are now logged out', 'success')
    return redirect(url_for('login'))

회원가입, 로그인, 로그아웃 기능을 하는 함수들이다.

회원가입 폼

로그인 이후 화면