背景

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