sqlalchemyのオブジェクトを辞書に変換し、reponseにjsonを返すAPIを作成したものを投稿します。
今回、WEBフレームワークは、「reposnder」を使っています。
２．ディレクトリ構成
├─mysite
│ alchemydb.py
│ jsonutils.py
│ models.py
│ run.py
│ sqlutils.py
│
└─templates
hello.html
layout.html
３．実装
###(1) alchemydb.py
データベースの初期化をする処理です。
alchemydb.py
# -*- coding: utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
import logging
Base = declarative_base()
RDB_PATH = 'postgresql+psycopg2://testuser:password@localhost:5432/flaskdb1'
ECHO_LOG = False
engine = create_engine(
RDB_PATH, echo=ECHO_LOG, pool_size=20, max_overflow=0
)
Session = sessionmaker(bind=engine)
###(2) models.py
sqlalchemyのmodelクラスを管理します。
models.py
# -*- coding: utf-8 -*-
from alchemydb import Base, engine
from sqlalchemy import Column, Integer, String, Text, text, DATETIME
from datetime import datetime
class User(Base):
__tablename__ = "users"
id = Column('id', Integer, primary_key = True)
username = Column('username', String(32))
mailaddress = Column('mailaddress', String(255))
password = Column('password', String(255))
role = Column('role', String(255))
created_at = Column('created_at', DATETIME, nullable=False, default=datetime.now)
updated_at = Column('updated_at', DATETIME, nullable=False, default=datetime.now, onupdate=datetime.now)
###(3) sqlutils.py
sqlalchemyのmodelオブジェクトを、辞書またはjsonに変換するためのユーティリティ。
sqlutils.py
import copy
import json
from jsonutils import json_converter
def alchemyrowtodict(obj):
entity = copy.deepcopy(obj) # ※２
objdict = entity.__dict__
if '_sa_instance_state' in objdict: # ※１
del objdict['_sa_instance_state']
return objdict
def alchemytodict(obj):
if isinstance(obj, list):
dicts=[]
for row in obj:
rowdict = alchemyrowtodict(row)
dicts.append(rowdict)
return dicts;
else:
return alchemyrowtodict(obj)
def alchemytojson(obj):
dictobj = alchemytodict(obj)
return json.dumps(dictobj, default=json_converter, indent=2 , ensure_ascii=False)
※１ sqlalchemyのmodelは、「_sa_instance_state」というプロパティがあり、この値をjavascriptに渡すとエラーになるので、「_sa_instance_state」を除いたデータを辞書に変換する必要があります。
※２ このとき、modelオブジェクトを直接編集しないように、cloneしています。
###(4) jsonutils.py
json.dumpsで、辞書をJsonに変換するときに、辞書のオブジェクトがdate, datetime型の場合に文字列に変換します。
jsonutils.py
from datetime import date, datetime
# date, datetimeを文字列に変換する
def json_converter(obj):
# 日付型は、文字列に変換
if isinstance(obj, datetime):
return obj.strftime("%Y/%m/%d %H:%M:%S")
elif isinstance(obj, date):
return obj.strftime("%Y/%m/%d")
# 上記以外はサポート対象外.
raise TypeError ("Type %s not serializable" % type(obj))
###(5) run.py
run.py
# -*- coding: utf-8 -*-
import responder
import traceback
from alchemydb import Session, engine
from models import User
from sqlutils import alchemytojson, alchemytodict
api = responder.API()
@api.route("/")
def hello_html(req, resp):
resp.html = api.template('hello.html')
@api.route("/api/user/{id}")
def users_json(req, resp, *, id):
session = Session()
try:
user = session.query(User).filter_by(id=id).first()
# mediaはdate, datetimeに対応していない？
# resp.media = alchemytodict(user)
resp.headers = {"Content-Type": "application/json; charset=utf-8"}
resp.content = alchemytojson(user)
except Exception:
traceback.print_exc()
resp.media ={"errmessage":"Error occured"}
finally:
session.close()
print(engine.pool.status())
@api.route("/api/users")
def users_json(req, resp):
session = Session()
try:
users = session.query(User).all()
resp.headers = {"Content-Type": "application/json; charset=utf-8"}
resp.content = alchemytojson(users)
except Exception:
traceback.print_exc()
resp.media ={"errmessage":"Error occured"}
finally:
session.close()
print(engine.pool.status())
if __name__ == '__main__':
api.run()
###(6) layout.html
テンプレートの共通レイアウト。
layout.html
<!doctype html>
<html>
<head>
<meta charset="utf-8"/>
<title>{{ title }}</title>
</head>
<body>
{% block content %}
<!-- ここにメインコンテンツを書く -->
{% endblock %}
</body>
</html>
###(7) hello.html
Rest APIを呼び出すための画面
hello.html
{% extends "layout.html" %}
{% block content %}
<h3>Responder sqlalchemy to json example</h3>
<ul>
<li><a href="/api/user/1">User（id=1）</a></li>
<li><a href="/api/users">User（list）</a></li>
</ul>
{% endblock %}
４．実行
cmd.prompt
(venv) C:\data\python\responder\mysite>python run.py
５．確認
http://127.0.0.1:5042/にブラウザにアクセスして、画面のURLをクリックしてみます。
- User（id=1）
{
"created_at": "2019/04/20 00:13:00",
"password": "admin",
"username": "password",
"updated_at": "2019/04/20 00:13:04",
"role": 0,
"mailaddress": "admin@localhost.com",
"id": 1
}
- User（list）
[
{
"created_at": "2019/04/20 00:13:00",
"password": "admin",
"username": "password",
"updated_at": "2019/04/20 00:13:04",
"role": 0,
"mailaddress": "admin@localhost.com",
"id": 1
},
{
"created_at": "2019/04/27 13:43:08",
"password": "guest",
"username": "password",
"updated_at": "2019/04/27 13:43:14",
"role": 9,
"mailaddress": "guest@localhost.com",
"id": 2
},
]
sqlalchemyのmodelにてこずりましたが、何とかjsonに変換することができました。