如何推迟对没有主键的视图/表的反射？

Question

我有一个视图

xyz.MyTable

，我想为其构建模型。它没有唯一的列，但列

col1

和

col2

的组合保证是唯一的。

我也无法在声明时访问引擎，因此我需要使用延迟反射。这是我的尝试：

import sqlalchemy
import sqlalchemy.ext.declarative

Base = sqlalchemy.orm.declarative_base()
class MyTable(sqlalchemy.ext.declarative.DeferredReflection, Base):
    __tablename__ = 'MyTable'
    __table__ = sqlalchemy.Table(__tablename__, Base.metadata, schema='xyz', autoload=True)
    __mapper_args__ = {'primary_key': [__table__.c.col1, __table__.c.col2]} 

但是，这会导致以下结果：

键错误：'col1'

这是有道理的，因为在反射发生之前，

col1

不能是一个对象。但是如何声明主键列呢？请注意，我无法控制数据库，也无法对视图进行任何更改。

更新：

根据 python_user 的建议，我尝试使用字符串作为列名称：

import sqlalchemy
from sqlalchemy import Column
from sqlalchemy.types import DATE, INTEGER, VARCHAR

from sqlalchemy.ext.declarative import declarative_base, DeferredReflection

Base = declarative_base(cls=DeferredReflection)

class VHistSnapPnL(Base):
    __tablename__ = "MyTable"

    __mapper_args__ = {
        "primary_key": [
            'col1',
            'col2'
        ]
    }

Base.prepare(engine)

但是这会导致属性错误：

---------------------------------------------------------------------------
AttributeError                            Traceback (most recent call last)
Cell In[10], line 6
      2 from sqlalchemy.ext.automap import automap_base
      4 # Base = automap_base()
      5 # reflect the tables
----> 6 Base.prepare(engine)

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/ext/declarative/extensions.py:400, in DeferredReflection.prepare(cls, engine)
    398 for thingy in to_map:
    399     cls._sa_decl_prepare(thingy.local_table, insp)
--> 400     thingy.map()
    401     mapper = thingy.cls.__mapper__
    402     metadata = mapper.class_.metadata

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/orm/decl_base.py:1127, in _DeferredMapperConfig.map(self, mapper_kw)
   1125 def map(self, mapper_kw=util.EMPTY_DICT):
   1126     self._configs.pop(self._cls, None)
-> 1127     return super(_DeferredMapperConfig, self).map(mapper_kw)

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/orm/decl_base.py:1047, in _ClassScanMapperConfig.map(self, mapper_kw)
   1042 else:
   1043     mapper_cls = mapper
   1045 return self.set_cls_attribute(
   1046     "__mapper__",
-> 1047     mapper_cls(self.cls, self.local_table, **self.mapper_args),
   1048 )

File <string>:2, in __init__(self, class_, local_table, properties, primary_key, non_primary, inherits, inherit_condition, inherit_foreign_keys, always_refresh, version_id_col, version_id_generator, polymorphic_on, _polymorphic_map, polymorphic_identity, concrete, with_polymorphic, polymorphic_load, allow_partial_pks, batch, column_prefix, include_properties, exclude_properties, passive_updates, passive_deletes, confirm_deleted_rows, eager_defaults, legacy_is_orphan, _compiled_cache_size)

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/util/deprecations.py:375, in deprecated_params.<locals>.decorate.<locals>.warned(fn, *args, **kwargs)
    368     if m in kwargs:
    369         _warn_with_version(
    370             messages[m],
    371             versions[m],
    372             version_warnings[m],
    373             stacklevel=3,
    374         )
--> 375 return fn(*args, **kwargs)

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py:693, in Mapper.__init__(self, class_, local_table, properties, primary_key, non_primary, inherits, inherit_condition, inherit_foreign_keys, always_refresh, version_id_col, version_id_generator, polymorphic_on, _polymorphic_map, polymorphic_identity, concrete, with_polymorphic, polymorphic_load, allow_partial_pks, batch, column_prefix, include_properties, exclude_properties, passive_updates, passive_deletes, confirm_deleted_rows, eager_defaults, legacy_is_orphan, _compiled_cache_size)
    691 self._configure_properties()
    692 self._configure_polymorphic_setter()
--> 693 self._configure_pks()
    694 self.registry._flag_new_mapper(self)
    695 self._log("constructed")

File /juno-tmp/virtualenvs/python3.11/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py:1391, in Mapper._configure_pks(self)
   1389 if self._primary_key_argument:
   1390     for k in self._primary_key_argument:
-> 1391         if k.table not in self._pks_by_table:
   1392             self._pks_by_table[k.table] = util.OrderedSet()
   1393         self._pks_by_table[k.table].add(k)

AttributeError: 'str' object has no attribute 'table'

Answer 1

来自 SQLAlchemy 2.0.16 的变更日志此处：

改进了

DeferredReflection.prepare()

以接受传递给

**kw

的任意

MetaData.reflect()

参数，允许诸如视图反射之类的用例

因此在 SQLA 2.0.16 及更高版本中，类似的视图

CREATE VIEW team AS
SELECT 'AB' AS prov, 'Calgary' AS city, 'Flames' AS team_name
UNION ALL
SELECT 'AB' AS prov, 'Edmonton' AS city, 'Oilers' AS team_name

可以与延迟反射一起使用，如下所示：

from sqlalchemy import create_engine, select
from sqlalchemy.ext.declarative import DeferredReflection
from sqlalchemy.orm import DeclarativeBase, Session


class Base(DeclarativeBase):
    pass


class Reflected(DeferredReflection):
    __abstract__ = True


class Team(Reflected, Base):
    __tablename__ = "team"  # name of the view
    __mapper_args__ = {"primary_key": ["prov", "city"]}


engine = create_engine("postgresql://scott:[email protected]/test")

Reflected.prepare(engine, views=True)

with Session(engine) as sess:
    home_team = sess.scalars(select(Team).where(Team.city == "Calgary")).one()
    print(home_team.team_name)  # Flames

对于 SQLAlchemy 的早期版本，您很可能必须省略延迟反射并简单地在 ORM 类中重新创建视图结构，例如，类似这样的

class Team(Base):
    __tablename__ = "team"  # name of the view
    prov = Column(String(5), primary_key=True)
    city = Column(String(25), primary_key=True)
    team_name = Column(String(25)