背景
postgresとPythonで連携するノウハウ
ポスグレリンク集
日本PostgreSQLユーザ会
PythonからPostgreSQLに接続する方法
import
pythonから使うためのツールはこんな感じです。
from flask import Flask, render_template, g
from hamlish_jinja import HamlishExtension
from werkzeug import ImmutableDict
from flask_sqlalchemy import SQLAlchemy
requirements.txt
Flask==1.0.0
line-bot-sdk==1.5.0
requests
psutil
psycopg2
flask-sqlalchemy
hamlish_jinja
werkzeug
テーブル生成
テーブル生成用のpythonスクリプトを作成
import os
from flask import Flask, render_template, g
from hamlish_jinja import HamlishExtension
from werkzeug import ImmutableDict
from flask_sqlalchemy import SQLAlchemy
class FlaskWithHamlish(Flask):
jinja_options = ImmutableDict(
extensions=[HamlishExtension]
)
app = FlaskWithHamlish(__name__)
db_uri = os.environ.get('DATABASE_URL')
app.config['SQLALCHEMY_DATABASE_URI'] = db_uri
db = SQLAlchemy(app)
class Entry(db.Model):
# テーブル名を定義
__tablename__ = "entries"
# カラムを定義
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(), nullable=False)
body = db.Column(db.String(), nullable=False)
テーブルを生成する
## 編集したファイルをデプロイ
$ git add . ; git commit -m "fix" ; git push
## herokuでテーブル作成
$ heroku run python -a [app名]
>>> # テーブル定義されたファイルを実行
>>> from app import db
>>> db.create_all()
SQLへの接続テストを行う
>>> # 必要ライブラリのインポート
>>> import os, psycopg2
>>>
>>> # DBヘ接続
>>> dsn = os.environ.get('DATABASE_URL')
>>> conn = psycopg2.connect(dsn)
>>> cur = conn.cursor()
>>>
>>> #insert文実行(適宜読み替えてください)
>>> cur.execute('BEGIN')
>>> cur.execute("insert into entries values(1,'title', 'body', 'date:XX:YY:ZZ')")
>>> cur.execute("insert into entries values(2,'title', 'body', 'date:XX:YY:ZZ')")
>>>
>>> # データを取得する
>>> cur.execute('SELECT * FROM entries')
>>> for row in cur:
... print(row[0])
>>> cur.execute('COMMIT')
>>> exit()
接続用のflaskアプリ作成
main.py
import os
import json
import psycopg2
from flask import Flask, request, abort, jsonify
app = Flask(__name__)
@app.route("/api/db/test", methods=['GET'])
def db_test():
dsn = os.environ.get('DATABASE_URL')
conn = psycopg2.connect(dsn)
cur = conn.cursor()
cur.execute('SELECT * FROM dzed')
res = {}
for i in cur:
res[i[0]] = i[1]
return jsonify(res)
if __name__ == "__main__":
port = int(os.getenv("PORT", 5000))
app.run(host="0.0.0.0", port=port)
REST API
REST APIのテンプレート
api.py
from flask import Blueprint, request, abort, jsonify
from models import db, User
api = Blueprint('api', __name__, url_prefix='/api')
@api.route('/users', methods=['GET'])
def list_user():
q_limit = request.args.get('limit', default=-1, type=int)
q_offset = request.args.get('offset', default=0, type=int)
if q_limit == -1:
users = User.query.all()
else:
users = User.query.offset(q_offset).limit(q_limit)
return jsonify({'users': [user.to_dict() for user in users]})
@api.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id=None):
user = User.query.filter_by(id=user_id).first()
return jsonify(user.to_dict())
@api.route('/users', methods=['POST'])
def post_user():
payload = request.json
name = payload.get('name')
age = payload.get('age')
user = User(name, age)
db.session.add(user)
db.session.commit()
response = jsonify(user.to_dict())
response.headers['Location'] = '/api/users/%d' % user.id
return response, 201
@api.route('/users/<user_id>', methods=['PUT'])
def put_user(user_id):
user = User.query.filter_by(id=user_id).first()
if not user:
abort(404, {'code': 'Not found', 'message': 'user not found'})
payload = request.json
user.name = payload.get('name')
user.age = payload.get('age')
db.session.commit()
return jsonify(user.to_dict())
@api.route('/users/<user_id>', methods=['DELETE'])
def delete_user(user_id):
user = User.query.filter_by(id=user_id).first()
if not user:
abort(404, {'code': 'Not found', 'message': 'user not found'})
db.session.delete(user)
db.session.commit()
return jsonify(None), 204
# エラーハンドリング
@api.errorhandler(400)
@api.errorhandler(404)
def error_handler(error):
return jsonify({'error': {
'code': error.description['code'],
'message': error.description['message']
}}), error.code