SQLAlchemy没有找到与postgres_fdw连接的Postgres表

Question

请原谅任何术语错别字，我对SQLite以外的数据库没有太多经验。我试图复制我在SQLite中所做的，我可以将数据库附加到第二个数据库并跨所有表进行查询。我没有使用SQLAlchemy和SQLite

我在Win7/54上使用SQLAlchemy 1.0.13, Postgres 9.5和Python 3.5.2(使用Anaconda)。我使用postgres_fdw连接了两个数据库(在本地主机上)，并从辅助数据库导入了一些表。我可以使用PgAdminIII中的SQL和使用psycopg2的Python成功地手动查询已连接的表。使用SQLAlchemy I've try:

# Same connection string info that psycopg2 used
engine = create_engine(conn_str, echo=True)
class TestTable(Base):
    __table__ = Table('test_table', Base.metadata,
                      autoload=True, autoload_with=engine)
    # Added this when I got the error the first time
    # test_id is a primary key in the secondary table
    Column('test_id', Integer, primary_key=True)

并得到错误:

sqlalchemy.exc.ArgumentError: Mapper Mapper|TestTable|test_table could not
assemble any primary key columns for mapped table 'test_table'

Then I try:

insp = reflection.Inspector.from_engine(engine)
print(insp.get_table_names())

和附加的表没有列出(主数据库中的表会显示出来)。有什么方法可以完成我想要完成的事情吗?

Answer 1

为了映射表，SQLAlchemy需要至少有一个列表示为主键列。这并不意味着该列在数据库中实际上必须是主键列，尽管这是一个好主意。根据从外部模式导入表的方式，表可能没有主键约束的表示，也可能没有任何其他约束。您可以通过在 Table实例(不在映射的类体中)中重写反射的主键列来解决这个问题，或者更好地告诉映射器哪些列包含候选键:

engine = create_engine(conn_str, echo=True)
test_table = Table('test_table', Base.metadata,
                   autoload=True, autoload_with=engine)
class TestTable(Base):
    __table__ = test_table
    __mapper_args__ = {
        'primary_key': (test_table.c.test_id, )  # candidate key columns
    }

使用PGInspector.get_foreign_table_names()方法检查外表名:

print(insp.get_foreign_table_names())

Answer 2

以@ilja的兄弟姐妹回答为基础。

当使用SQLAlchemy自动映射特性从现有数据库模式自动生成映射的类和关系时，我发现__mapper_args__解决方案不能创建模型。

手动定义私钥的替代方法将正确地使automap能够创建您的模型。

from sqlalchemy import Column, create_engine, Text
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.schema import Table
Base = automap_base()
engine = create_engine(conn_str, convert_unicode=True)
pk = Column('uid', Text, primary_key=True)
test_table = Table(
    'test_table', Base.metadata, pk, autoload=True, autoload_with=engine
)
# Inspect postgres schema
Base.prepare(engine, reflect=True)
print(dict(Base.classes))
print(test_table)