2017-10-21

flask / sqlalchemy / pandas / tips

SQLAlchemy Query를 Pandas DataFrame로 만들기

이번 글은 기존 DB를 Flask-SQLAlchemy ORM Model로 사용하기를 보고 오시면 좀더 빠르게 실 프로젝트에 적용이 가능합니다.

들어가며

전체 예시를 보시려면 TL;DR를 참고하세요.

DB에 있는 정보를 파이썬 코드 속에서 SQL raw Query를 통해 정보를 가져오는 아래와 같은 코드의 형태는 대다수의 언어에서 지원합니다.

import sqlite3

# 굳이 sqlite3이 아닌 다른 MySQL와 같은 DB의 connect를 이뤄도 상관없습니다.
# 여기서는 파이썬 파일과 같은 위치에 blog.sqlite3 파일이 있다고 가정합니다.
conn = sqlite3.connect("blog.sqlite3") 
cur = conn.cursor()
cur.execute("select * from post where id < 10;")

위와 같은 형식으로 코드를 사용할 경우 웹이 이루어지는 과정 중 2~3번째 과정인 “SQL쿼리 요청하기”와 “데이터 받기”라는 부분을 수동으로 처리해 줘야 하는 부분이 있습니다.

이런 경우 파이썬 파일이더라도 한 파일 안에 두개의 언어를 사용하게 되는 셈입니다. (python와 SQL)

만약 여러분이 Pandas DataFrame객체를 DB에서 가져와 만들려면 이런 문제가 생깁니다.

DB에 연결을 구성해야 함
가져온 데이터를 데이터 타입에 맞춰 파이썬이 이해하는 형태로 변환
정리한 데이터를 Pandas로 불러오기

음, 보기만 해도 상당히 귀찮네요.

설치하기

우선 필요한 패키지들을 먼저 설치해 줍시다.

1
2
3

pip install flask
pip install Flask-SQLAlchemy
pip install pandas

Pandas로 SQL요청하기

Pandas에서는 이런 귀찮은 점을 보완해 주기 위해 read_sql_query라는 함수를 제공합니다. 위 코드를 조금 바꿔봅시다.

import sqlite3
import pandas as pd # NoQA

conn = sqlite3.connect("blog.sqlite3")
# 이 부분을 삭제
# cur = conn.cursor()
# cur.execute("select * from post where id < 10;")

# 아래 부분을 추가
df = pd.read_sql_query("select * from post where id < 10;", conn)
# df는 이제 Pandas Dataframe 객체

단순하게 DB 커넥션, 그리고 read_sql_query만으로 SQL Query를 바로 Pandas DataFrame 객체로 받아왔습니다. 이제 데이터를 수정하고 가공하는 처리는 Pandas에게 맡기면 되겠군요!

하지만, 여전히 우리는 SQL을 짜고있어요. 복잡한 쿼리라면 몰라도, 단순한 쿼리를 이렇게까지 할 필요가 있을까요?

SQLAlchemy 모델 이용하기

Flask를 사용할때 많이 쓰는 SQLAlchemy는 ORM으로 수많은 DB를 파이썬만으로 제어하도록 도와줍니다. 그리고 이 점이 우리가 SQL을 SQLAlchemy를 통해 바로 만들 수 있도록 도와줍니다.

NOTE: 이번 글에서는 Flask-SQLAlchemy 패키지를 사용합니다. SQLAlchemy와는 약간 다르게 동작할 수도 있습니다.

모델 클래스 만들기

모델 클래스를 기존 DB를 참조해 만드는 것은 기존 DB를 Flask-SQLAlchemy ORM Model로 사용하기 를 참고하세요.

예제 모델: Post

블로그에서 자주 쓸 법한 Post라는 이름의 모델 클래스를 하나 만들어 봅시다.

우선 SQLAlchemy를 flask_sqlalchemy에서 import 해옵시다. 그리고 Flask도 가져와 app을 만들어 줍시다. 그리고 db객체를 만들어줍시다.

from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
import pandas as pd

app = Flask(__name__)
# 현재 경로의 blog.sqlite3을 불러오기
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.sqlite3'
db = SQLAlchemy(app)

class Post(db.Model):
    __tablename__ = 'post'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    content = db.Column(db.Text)
    pub_date = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

자, 이제 여러분은 blog.sqlite3파일 안에 post라는 테이블에 값들을 넣거나 뺄 수 있게 되었습니다.

루트 View 만들기

여러분이 app.run( ) 으로 Flask 개발 서버를 띄웠을 때 첫 화면(‘/‘ URL에서) 실행될 View 함수(post_all)를 만들어줍시다.

from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
import pandas as pd
import json

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.sqlite3'
db = SQLAlchemy(app)

class Post(db.Model):
    __tablename__ = 'post'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    content = db.Column(db.Text)
    pub_date = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

# 아래 줄을 추가해 줍시다.
# List post which id is less then 10
@app.route('/')
def post_all():
    df = pd.read_sql_query("select * from post where id < 10;", db.session.bind).to_json()
    return jsonify(json.loads(df))

자, 분명히 ORM을 쓰는데도 아직 SQL 쿼리를 쓰고있네요! SQL쿼리문을 지워버립시다!

queryset 객체를 만들기

우리는 Post라는 모델을 만들어줬으니 이제 Post객체의 .query와 .filter()를 통해 객체들을 가져와 봅시다.

우선 queryset라는 이름에 넣어줍시다. 그리고 Pandas의 read_sql(유의: read_sql_query가 아닙니다.)에 queryset의 내용과 세션을 넘겨줘 DataFrame 객체로 만들어줍시다.

from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
import pandas as pd
import json

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.sqlite3'
db = SQLAlchemy(app)

class Post(db.Model):
    __tablename__ = 'post'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    content = db.Column(db.Text)
    pub_date = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

# List post which id is less then 10
@app.route('/')
def post_all():
    # 이 줄은 지우고,
    # df = pd.read_sql_query("select * from post where id < 10;", db.session.bind).to_json()
    # 아래 두줄을 추가해주세요.
    queryset = Post.query.filter(Post.id < 10)  # SQLAlchemy가 만들어준 쿼리, 하지만 .all()이 없어 실행되지는 않음
    df = pd.read_sql(queryset.statement, queryset.session.bind)  # 진짜로 쿼리가 실행되고 DataFrame이 만들어짐
    return jsonify(json.loads(df).to_json())

자, 위와 같이 코드를 짜 주면 이제 SQLAlchemy ORM와 Pandas의 read_sql을 통해 df이 DataFrame 객체로 자연스럽게 가져오게 됩니다.

정리하기

여러분이 Pandas를 사용해 데이터를 분석하거나 정제하려 할 때 웹앱으로 Flask를 사용하고 ORM을 이용한다면, 굳이 SQL Query를 직접 만드는 대신 이처럼 Pandas와 SQLAlchemy의 강력한 조합을 이용해 보세요. 조금 더 효율적인 시스템 활용을 고려한 파이썬 프로그램이 나올거에요!

TL;DR

아래 코드와 같이 모델을 만들고 db 객체를 만든 뒤 pandas의 read_sql을 사용하면 됩니다.

from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
import pandas as pd
import json

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///blog.sqlite3'
db = SQLAlchemy(app)

class Post(db.Model):
    __tablename__ = 'post'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100))
    content = db.Column(db.Text)
    pub_date = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)

# List post which id is less then 10
@app.route('/')
def post_all():
    queryset = Post.query.filter(Post.id < 10)  # SQLAlchemy가 만들어준 쿼리, 하지만 .all()이 없어 실행되지는 않음
    df = pd.read_sql(queryset.statement, queryset.session.bind)  # 진짜로 쿼리가 실행되고 DataFrame이 만들어짐
    return jsonify(json.loads(df).to_json())

2017-10-20

flask / sqlalchemy

기존 DB를 Flask-SQLAlchemy ORM Model로 사용하기

본 게시글에서는 MySQL/Sqlite을 예제로 하고있지만, Flask-SQLAlchemy가 지원하는 다른 DB에서도 사용 가능합니다.

들어가며

Flask로 웹 개발 진행 시 SQLAlchemy(Flask-SQLAlchemy)를 사용해 ORM구조를 구성할 때 데이터를 저장할 DB의 구조를 직접 확인하며 진행하는 것은 상당히 귀찮고 어려운 일입니다.

Django에는 내장된 inspectdb라는 명령어를 통해 Django와 일치하는 DB Model구조를 만들어주지만 SQLAlchemy 자체에 내장된 automap은 우리가 상상하는 모델 구조를 바로 만들어주지는 않습니다.

따라서 다른 패키지를 고려해볼 필요가 있습니다.

flask-sqlacodegen

flask-sqlacodegen은 기존 DB를 Flask-SQLAlchemy에서 사용하는 Model 형식으로 변환해 보여주는 패키지입니다. 기존 sqlacodegen에서 포크해 Flask-SQLAlchemy에 맞게 기본 설정이 갖추어져있어 편리합니다.

설치하기

설치는 pip로 간단하게 진행해 주세요.

글쓰는 시점 최신버전은 1.1.6.1입니다.

글쓴것과 같은 버전으로 설치하려면 flask-sqlacodegen==1.1.6.1 로 설치해 주세요.

# 최신 버전 설치하기
pip install flask-sqlacodegen
# 글쓴 시점과 같게 설치하려면
# pip install flask-sqlacodegen==1.1.6.1

설치가 완료되면 명령줄에서 flask-sqlacodegen라는 명령어를 사용할 수 있습니다.

주의: sqlacodegen이 이미 깔려있다면 다른 가상환경(virtuale / venv)를 만드시고 진행해 주세요. sqlacodegen이 깔려있으면 --flask이 동작하지 않습니다.

DB 구조 뜯어내기

flask-sqlacodegen는 sqlacodegen과 거의 동일한 문법을 사용합니다.(포크를 뜬 프로젝트니까요!)

flask-sqlacodegen 명령어로 DB를 지정하면 구조를 알 수 있습니다.

SQLite의 경우

1	flask-sqlacodegen "sqlite:///db.sqlite3" --flask > models.py # 상대경로, 현재 위치의 db.sqlite3파일

SQLite는 로컬에 있는 DB의 위치를 지정하면 됩니다.

위 명령어를 실행하면 models.py파일 안에 db.sqlite3 DB의 모델이 정리됩니다.

NOTE: Sqlite의 파일을 지정할 경우 “sqlite://“가 아닌 “sqlite:///“ 로 /를 3번 써주셔야 상대경로로 지정 가능하며, “sqlite:////“로 /를 4번 써주셔야 절대경로로 지정이 가능합니다.

mysql 서버의 경우

1	flask-sqlacodegen "mysql://username:password@DB_IP/DB_NAME" --flask > models.py

MySQL의 경우 mysql에 접속하는 방식 그대로 사용자 이름, 비밀번호, IP(혹은 HOST도메인), DB이름을 넣어준 뒤 진행해주면 됩니다.

NOTE: mysql은 “mydql://“ 로 /가 2번입니다.

NOTE: mysql에 연결하려면 pip패키지 중 mysqlclient가 설치되어있어야 합니다.
설치가 되어있지 않으면 아래와 같이 ModuleNotFoundError가 발생합니다.

MAC에서 진행 중 혹시 mysqlclient설치 중 아래와 같은 에러가 발생한다면

아래 명령어를 실행해 xcode cli developer tool와 openssl을 설치해주신 후 mysqlclient를 설치해 주세요.

xcode-select --install
brew install openssl
export LIBRARY_PATH=$LIBRARY_PATH:/usr/local/opt/openssl/lib/
pip install mysqlclient

실행결과

아래 결과는 장고 프로젝트를 생성하고 첫 migrate를 진행할 때 생기는 예시 db.sqlite3파일을 flask-sqlacodegen을 사용한 결과입니다.

Index, PK등을 잘 잡아주고 있는 모습을 볼 수 있습니다.

# models.py 파일
# coding: utf-8
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Index, Integer, String, Table, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql.sqltypes import NullType
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class AuthGroup(db.Model):
    __tablename__ = 'auth_group'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)


class AuthGroupPermission(db.Model):
    __tablename__ = 'auth_group_permissions'
    __table_args__ = (
        db.Index('auth_group_permissions_group_id_permission_id_0cd325b0_uniq', 'group_id', 'permission_id'),
    )

    id = db.Column(db.Integer, primary_key=True)
    group_id = db.Column(db.ForeignKey('auth_group.id'), nullable=False, index=True)
    permission_id = db.Column(db.ForeignKey('auth_permission.id'), nullable=False, index=True)

    group = db.relationship('AuthGroup', primaryjoin='AuthGroupPermission.group_id == AuthGroup.id', backref='auth_group_permissions')
    permission = db.relationship('AuthPermission', primaryjoin='AuthGroupPermission.permission_id == AuthPermission.id', backref='auth_group_permissions')


class AuthPermission(db.Model):
    __tablename__ = 'auth_permission'
    __table_args__ = (
        db.Index('auth_permission_content_type_id_codename_01ab375a_uniq', 'content_type_id', 'codename'),
    )

    id = db.Column(db.Integer, primary_key=True)
    content_type_id = db.Column(db.ForeignKey('django_content_type.id'), nullable=False, index=True)
    codename = db.Column(db.String(100), nullable=False)
    name = db.Column(db.String(255), nullable=False)

    content_type = db.relationship('DjangoContentType', primaryjoin='AuthPermission.content_type_id == DjangoContentType.id', backref='auth_permissions')


class AuthUser(db.Model):
    __tablename__ = 'auth_user'

    id = db.Column(db.Integer, primary_key=True)
    password = db.Column(db.String(128), nullable=False)
    last_login = db.Column(db.DateTime)
    is_superuser = db.Column(db.Boolean, nullable=False)
    first_name = db.Column(db.String(30), nullable=False)
    last_name = db.Column(db.String(30), nullable=False)
    email = db.Column(db.String(254), nullable=False)
    is_staff = db.Column(db.Boolean, nullable=False)
    is_active = db.Column(db.Boolean, nullable=False)
    date_joined = db.Column(db.DateTime, nullable=False)
    username = db.Column(db.String(150), nullable=False)


class AuthUserGroup(db.Model):
    __tablename__ = 'auth_user_groups'
    __table_args__ = (
        db.Index('auth_user_groups_user_id_group_id_94350c0c_uniq', 'user_id', 'group_id'),
    )

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.ForeignKey('auth_user.id'), nullable=False, index=True)
    group_id = db.Column(db.ForeignKey('auth_group.id'), nullable=False, index=True)

    group = db.relationship('AuthGroup', primaryjoin='AuthUserGroup.group_id == AuthGroup.id', backref='auth_user_groups')
    user = db.relationship('AuthUser', primaryjoin='AuthUserGroup.user_id == AuthUser.id', backref='auth_user_groups')


class AuthUserUserPermission(db.Model):
    __tablename__ = 'auth_user_user_permissions'
    __table_args__ = (
        db.Index('auth_user_user_permissions_user_id_permission_id_14a6b632_uniq', 'user_id', 'permission_id'),
    )

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.ForeignKey('auth_user.id'), nullable=False, index=True)
    permission_id = db.Column(db.ForeignKey('auth_permission.id'), nullable=False, index=True)

    permission = db.relationship('AuthPermission', primaryjoin='AuthUserUserPermission.permission_id == AuthPermission.id', backref='auth_user_user_permissions')
    user = db.relationship('AuthUser', primaryjoin='AuthUserUserPermission.user_id == AuthUser.id', backref='auth_user_user_permissions')


class DjangoAdminLog(db.Model):
    __tablename__ = 'django_admin_log'

    id = db.Column(db.Integer, primary_key=True)
    object_id = db.Column(db.Text)
    object_repr = db.Column(db.String(200), nullable=False)
    action_flag = db.Column(db.Integer, nullable=False)
    change_message = db.Column(db.Text, nullable=False)
    content_type_id = db.Column(db.ForeignKey('django_content_type.id'), index=True)
    user_id = db.Column(db.ForeignKey('auth_user.id'), nullable=False, index=True)
    action_time = db.Column(db.DateTime, nullable=False)

    content_type = db.relationship('DjangoContentType', primaryjoin='DjangoAdminLog.content_type_id == DjangoContentType.id', backref='django_admin_logs')
    user = db.relationship('AuthUser', primaryjoin='DjangoAdminLog.user_id == AuthUser.id', backref='django_admin_logs')


class DjangoContentType(db.Model):
    __tablename__ = 'django_content_type'
    __table_args__ = (
        db.Index('django_content_type_app_label_model_76bd3d3b_uniq', 'app_label', 'model'),
    )

    id = db.Column(db.Integer, primary_key=True)
    app_label = db.Column(db.String(100), nullable=False)
    model = db.Column(db.String(100), nullable=False)


class DjangoMigration(db.Model):
    __tablename__ = 'django_migrations'

    id = db.Column(db.Integer, primary_key=True)
    app = db.Column(db.String(255), nullable=False)
    name = db.Column(db.String(255), nullable=False)
    applied = db.Column(db.DateTime, nullable=False)


class DjangoSession(db.Model):
    __tablename__ = 'django_session'

    session_key = db.Column(db.String(40), primary_key=True)
    session_data = db.Column(db.Text, nullable=False)
    expire_date = db.Column(db.DateTime, nullable=False, index=True)


t_sqlite_sequence = db.Table(
    'sqlite_sequence',
    db.Column('name', db.NullType),
    db.Column('seq', db.NullType)
)

Flask의 app에 덧붙이기

이렇게 만들어진 model은 다른 Extension과 동일하게 Flask app에 붙일 수 있습니다.

app.py라는 파일을 하나 만들고 아래 내용으로 채워주세요.

# app.py (models.py와 같은 위치)
from flask import Flask

import models # models.py파일을 가져옵시다.

def create_app():
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = "mysql://username:password@DB_IP/DB_NAME"
    models.db.init_app(app)
    return app

if __name__=='__main__':
    app = create_app()
    app.run()

앞서 만들어준 models.py파일을 가져와 create_app 함수를 통해 app을 lazy_loading해주는 과정을 통해 진행해 줄 수 있습니다.

마치며

기존에 사용하던 DB를 Flask와 SqlAlchemy를 통해 ORM으로 이용해 좀 더 빠른 개발이 가능하다는 것은 큰 이점입니다. ORM에서 DB 생성을 하지 않더라도 이미 있는 DB를 ORM으로 관리하고 Flask 프로젝트에 바로 가져다 쓸 수 있다는 점이 좀 더 빠른 프로젝트 진행에 도움이 될거랍니다.