Python SQL Alchemy 多个数据库 - 绑定 Automap_Base

Question

我正在使用 SQL Alchemy，但正在努力将文档中的信息结构化到我的项目中。我有两个数据库，我的第一个将用于存储来自 python 应用程序的所有新信息。第二个数据库（在本例中为 DB1）是我需要从中访问信息的现有数据库。使用 SQLAlchemy 创建这种结构的正确方法是什么？

我对多个数据库使用了建议的 BINDS 方法。这似乎有效。

class BaseConfig(object):
     SECRET_KEY = "SO_SECURE"
     DEBUG = True
     SQLALCHEMY_DATABASE_URI = 'mssql+pyodbc://sa:funpassword@(localdb)\MSSQLLocalDB/Testing?driver=SQL+Server+Native+Client+11.0'
     SQLALCHEMY_BINDS = {
         'DB1': 'mssql+pyodbc://sa:$funpassword@ProdDB/DB1?driver=SQL+Server+Native+Client+11.0'
     }
     SQLALCHEMY_TRACK_MODIFICATIONS = True

此配置似乎可以正常工作，因为我可以使用以下代码在这两个数据库中创建新模型。 （这样做只是为了确认我能够连接到两者）。 db 是我的 index.py 文件中的 SqlAlchemy(app) 初始化。

from index import app, db

#Test Writing To Default DB
class Test(db.Model):
     id = db.Column(db.Integer(), primary_key=True)
     employeeNum = db.Column(db.String(255), unique=False)
     job = db.Column(db.String(255))

    def __init__(self, employeeNum, job):
         self.employeeNum = employeeNum
         self.job = job

# Test Writing To DB1
class Test(db.Model):
     __bind_key__ = 'DB1' 
     id = db.Column(db.Integer(), primary_key=True)
     employeeNum = db.Column(db.String(255), unique=False)
     job = db.Column(db.String(255))

    def __init__(self, employeeNum, job):
         self.employeeNum = employeeNum
         self.job = job

我使用 SQLAlchemy 文档中的 table 和 automap_base 尝试了许多组合，但这似乎不起作用。当我尝试映射现有表时，我不确定如何使用 DB1 的 bind_key。

from index import app, db

def loadSession():
    Table('employee', db.metadata, Column('emp_num', Text, primary_key=True))
    Base = automap_base(metadata=metadata)
    Base.prepare()
    employee = Base.classes.employee
    emp = db.session.query(employee).all()
    for i in emp:
        print(i.data)

感谢您提供的任何帮助。

Answer 1

对于具有现有架构和数据的 DB1，您可以使用反射将表映射到 sqlalchemy。

对于您的示例，它可能看起来像这样：

from sqlalchemy.ext.automap import automap_base
from sqlalchemy import MetaData
from index import app, db

def loadSession():
    # get a db engine object for DB1 configuration
    engine = db.get_engine(bind='DB1')

    # create a empty db Metadata object and bind it to DB1
    database = MetaData(bind=engine)

    # load the DB1 table/column structure into the Metadata object. This is quite a heavy operation and should optimally be done once on app setup and then the created automap_base can be stored / reused on demand until app shutdown
    database.reflect(
        bind=engine, views=True, autoload_replace=False
    )
    # create a python class structure out of the db metadata structure
    auto_base = automap_base(metadata=database)
    auto_base.prepare()

    # create a db session for DB1
    db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
    # get the reflected employees class from DB1. The employee table must exist in the database
    employee = auto_base.classes.employee

    # query DB1 with the new session object for all employee records
    emp = db_session.query(employee).all()
    for i in emp:
        print(i.data)

    # the following could also be substituted for the query without creating a session object:
    for entry in employee.query.all():
        print(entry)