Flask sqlalchemy 예시

단순 PostgreSQL db read/write 예시

이 경우는 data model을 만들어 줘야 함.

# coding: utf-8
from sqlalchemy import Column, DateTime, Integer, String
from sqlalchemy.schema import FetchedValue
from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()


class Barcode(db.Model):
    __tablename__ = 'barcode'

    code_seq = db.Column(db.Integer, primary_key=True, unique=True, server_default=db.FetchedValue())
    barcode = db.Column(db.String(50), nullable=False)
    ref_num = db.Column(db.String(30))
    pname = db.Column(db.String(100))
    size = db.Column(db.Integer)
    updatedate = db.Column(db.DateTime)

 

app.py에서는 위 model ( from moduel.localmodels import Barcode ) 부분을 추가해야 모델을 읽고 쑬수 있다.

from flask import Flask, render_template, request
import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session
import datetime, json

app = Flask(__name__)
url = 'postgresql://{}:{}@{}:{}/{}'.format('erp', 'erp', '127.0.0.1', 5432, 'erpdb')
engine = sqlalchemy.create_engine(url)
Session = scoped_session(sessionmaker(bind=engine))
s = Session()

def json_default(value):
    if isinstance(value, datetime.date):
        return value.strftime('%Y/%m/%d')
    raise TypeError('not JSON serializable')

@app.route('/find/<bcode>')
def find(bcode):
    res = []
    results = s.execute('SELECT * FROM barcode WHERE barcode = :bcode', {'bcode': bcode})
    for row in results:
        res.append(dict(row))
    print(res)
    return json.dumps(res, default=json_default)

@app.route('/exec/<query>')
def exec(query):
    try:
        res = []
        results = s.execute(query)
        for row in results:
            res.append(dict(row))
        print(res)
        return json.dumps(res, default=json_default)
    except:
        s.rollback()
        raise
    finally:
        s.commit()

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

 

이렇게 Model을 새로 작성시 번거럽다 Tool을 쓰면 Model을 자동생성이 가능하다.

 

flask-sqlacodegen "postgresql://erp:erp@127.0.0.1:5432/erpdb"  --flask > localmodels.py

하지만, 아예 모델을 신경 쓰고 싶지 않다면.. 아래 처럼 engine을 사용해야 한다.

 

from flask import Flask, render_template, request
import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session
import datetime, json

app = Flask(__name__)
url = 'postgresql://{}:{}@{}:{}/{}'.format('erp', 'erp', '127.0.0.1', 5432, 'erpdb')
engine = sqlalchemy.create_engine(url)
Session = scoped_session(sessionmaker(bind=engine))
s = Session()

def json_default(value):
    if isinstance(value, datetime.date):
        return value.strftime('%Y/%m/%d')
    raise TypeError('not JSON serializable')

@app.route('/find/<bcode>')
def find(bcode):
    res = []
    results = s.execute('SELECT * FROM barcode WHERE barcode = :bcode', {'bcode': bcode})
    for row in results:
        res.append(dict(row))

    print(res)
    return json.dumps(res, default=json_default)


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