LoginSignup
4

More than 3 years have passed since last update.

postgres × python (+ heroku)

Last updated at Posted at 2019-08-07

背景

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

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
4