Flask-SQLAlchemyで複数のデータベースに接続するときの備忘録

概要

Flask-SQLAlchemyで複数のデータベースに接続するときの覚え書き。

データベースの構築

docker-composeでPostgreSQLを建てる

pgdb:
    image: postgres:latest
    ports: ["5555:5432"]
    environment: 
    - POSTGRES_DB=test
    - POSTGRES_PASSWORD=password
    - TZ=Asia/Tokyo
    command: ["postgres", "-c", "log_statement=all", "-c", "log_destination=stderr"]

１つのデータベースに接続する場合

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import datetime


app = Flask(__name__,template_folder='../templates')
app.config['SQLALCHEMY_DATABASE_URI']= 'postgresql://postgres:password@localhost:5432/test'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO']=True

db = SQLAlchemy(app)

class test(db.Model):
    __tablename__ = 'test'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    test_model = db.Column(db.Text, nullable=True)
    test_vendor = db.Column(db.Text, nullable=True)
    test_serial_number = db.Column(db.Text, nullable=True)
    test_url = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

class test2(db.Model):
    __tablename__ = 'test2'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    test2_id = db.Column(db.Integer, nullable=True)
    test2_connector_id = db.Column(db.Integer, nullable=True)
    test2_ip_address = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

with app.app_context():
    db.drop_all()
    db.create_all()

複数のデータベースに接続する場合

app = Flask(__name__,template_folder='../templates')
app.config['SQLALCHEMY_DATABASE_URI']= 'postgresql://postgres:password@localhost:5432/test'
app.config['SQLALCHEMY_BINDS'] = {
    'db0': app.config['SQLALCHEMY_DATABASE_URI'],
    'db1': 'postgresql://postgres:password@localhost:5555/test'
}
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO']=True

db = SQLAlchemy(app)

class test(db.Model):
    __tablename__ = 'test'
    __bind_key__ = 'db0'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    test_model = db.Column(db.Text, nullable=True)
    test_vendor = db.Column(db.Text, nullable=True)
    test_serial_number = db.Column(db.Text, nullable=True)
    test_url = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

class test2(db.Model):
    __tablename__ = 'test2'
    __bind_key__ = 'db1'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    test2_id = db.Column(db.Integer, nullable=True)
    test2_connector_id = db.Column(db.Integer, nullable=True)
    test2_ip_address = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.datetime.now)
    updated_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)

with app.app_context():
    db.drop_all()
    db.create_all()
    """"もしくは"""
    # db.create_all(bind=['db0'])
    # db.create_all(bind=['db1'])

備考

複数のデータベースからそれぞれの値を取り出すことはできたが、他データベース間はリレーションが貼れないらしくJOINはどう頑張ってもできなかった。。。