定义后更改 SQLAlchemy 主键

Posted 2023-04-12

【中文标题】定义后更改 SQLAlchemy 主键

【英文标题】：Change SQLAlchemy Primary Key after it has been defined

【发布时间】：2018-07-30 04:10:25

【问题描述】：

问题：简单地说，我正在尝试在 SQLAlchemy ORM 表的主键已经定义之后重新定义它。

示例：

class Base:

    @declared_attr
    def __tablename__(cls):
        return f"cls.__name__"


    @declared_attr
    def id(cls):
        return Column(Integer, cls.seq, unique=True, 
                      autoincrement=True, primary_key=True)

Base = declarative_base(cls=Base)

class A_Table(Base):
    newPrimaryKeyColumnsDerivedFromAnotherFunction = []
    # Please Note: as the variable name tries to say,
    # these columns are auto-generated and not known until after all
    # ORM classes (models) are defined

# OTHER CLASSES


def changePriKeyFunc(model):
    pass # DO STUFF

# Then do
Base.metadata.create_all(bind=arbitraryEngine)
# After everything has been altered and tied into a little bow

*请注意，这是对我要解决的真正问题的简化。

可能的解决方案：您的第一个想法可能是这样做：

def possibleSolution(model):
    for pricol in model.__table__.primary_key:
        pricol.primary_key = False

    model.__table__.primary_key = PrimaryKeyConstraint(
        *model.newPrimaryKeyColumnsDerivedFromAnotherFunction,

        # TODO: ADD all the columns that are in the model that are also a primary key
        # *[col for col in model.__table__.c if col.primary_key]
        )

但是，这不起作用，因为在尝试添加、刷新和提交时，会抛出错误：

InvalidRequestError: Instance <B_Table at 0x104aa1d68> cannot be refreshed - 
it's not persistent and does not contain a full primary key.

即使这样：

In [2]: B_Table.__table__.primary_key
Out[2]: PrimaryKeyConstraint(Column('a_TableId', Integer(),
                                    ForeignKey('A_Table.id'), table=<B_Table>, 
                                    primary_key=True, nullable=False))

还有这个：

In [3]: B_Table.__table__
Out[3]: Table('B_Table', MetaData(bind=None), 
              Column('id', Integer(), table=<B_Table>, nullable=False,
                     default=Sequence('test_1', start=1, increment=1, 
                                      metadata=MetaData(bind=None))), 
              Column('a_TableId', Integer(), 
                     ForeignKey('A_Table.id'), table=<B_Table>, 
                     primary_key=True, nullable=False), 
              schema=None)

最后：

In [5]: b.a_TableId
Out[5]: 1

另请注意，数据库实际上反映了更改的（和真实的）主键，所以我知道 ORM/SQLAlchemy 发生了一些事情。

问题：综上所述，在模型已经定义好之后，如何更改模型的主键？

编辑：完整代码见下文（相同类型的错误，仅在 SQLite 中）

from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declared_attr, declarative_base
from sqlalchemy.schema import PrimaryKeyConstraint
from sqlalchemy import Sequence, create_engine


class Base:

    @declared_attr
    def __tablename__(cls):
        return f"cls.__name__"

    @declared_attr
    def seq(cls):
        return Sequence("test_1", start=1, increment=1)

    @declared_attr
    def id(cls):
        return Column(Integer, cls.seq, unique=True, autoincrement=True, primary_key=True)


Base = declarative_base(cls=Base)


def relate(model, x):
    """Model is the original class, x is what class needs to be as
    an attribute for model"""

    attributeName = x.__tablename__
    idAttributeName = "Id".format(attributeName)
    setattr(model, idAttributeName,
            Column(ForeignKey(x.id)))
    setattr(model, attributeName,
            relationship(x,
                         foreign_keys=getattr(model, idAttributeName),
                         primaryjoin=getattr(
                             model, idAttributeName) == x.id,
                         remote_side=x.id
                         )
            )
    return model.__table__.c[idAttributeName]


def possibleSolution(model):
    if len(model.defined):
        newPriCols = []
        for x in model.defined:
            newPriCols.append(relate(model, x))

        for priCol in model.__table__.primary_key:
            priCol.primary_key = False
            priCol.nullable = True

        model.__table__.primary_key = PrimaryKeyConstraint(
            *newPriCols
            # TODO: ADD all the columns that are in the model that are also a primary key
            # *[col for col in model.__table__.c if col.primary_key]
            )


class A_Table(Base):
    pass


class B_Table(Base):
    defined = [A_Table]


possibleSolution(B_Table)

engine = create_engine('sqlite://')
Base.metadata.create_all(bind=engine)

Session = sessionmaker(bind=engine)
session = Session()

a = A_Table()
b = B_Table(A_TableId=a.id)

print(B_Table.__table__.primary_key)

session.add(a)
session.commit()
session.add(b)
session.commit()

【问题讨论】：

我的第一个想法是保留自动增量主键，并为您的“其他”键值设置另一列索引。这可能吗？

@SuperShoot 您可以假设 id 列没有被删除。至于有另一个专栏，我不确定你的意思。

我的意思是在你的表中添加另一列，命名为other_id，在其上放置一个索引以便快速查询，并将该列的值设置为可以仅在已创建对象后才生成，以免您在 PK 存在时弄乱它。如果您无法更改架构，那么这不是一个选项，但这就​​是我要问的原因。

对不起，我刚刚意识到您不想更改存储在表中的行的主键值，您想在实际创建表之前更改表的主键定义数据库基于与 ORM 类本身的创建相关的一些变量，对吗？我的错。

请提供一个明确的问题。为什么要重新定义模型，而不是简单地以正确的形式创建它们？这个问题对您来说似乎毫无意义，但请注意您的用例似乎相当罕见。同时提供minimal reproducible example。给出的示例代码无法重现您遇到的错误。

【参考方案1】：

原来你说的PK重新分配导致的错误是：

InvalidRequestError: Instance <B_Table at 0x104aa1d68> cannot be refreshed - 
it's not persistent and does not contain a full primary key.

我没有让你运行 MCVE，而是首先收到一个非常有用的警告：

SAWarning：列“B_Table.A_TableId”被标记为 表“B_Table”的主键，但没有 Python 端或服务器端 指示默认生成器，也不指示'autoincrement = True' 或 'nullable=True'，并且没有传递显式值。首要的关键 列通常不能存储 NULL。

以及脚本失败时非常详细的异常信息：

sqlalchemy.orm.exc.FlushError: 实例有 一个 NULL 身份密钥。如果这是一个自动生成的值，请检查 数据库表允许生成新的主键值，并且 映射的 Column 对象被配置为期望这些生成 价值观。还要确保此 flush() 不会发生在 不适当的时间，例如在 load() 事件中。

因此，假设该示例准确地描述了您的问题，那么答案就很简单了。主键不能为空。

A_Table 继承 Base:

class A_Table(Base):
    pass

Base 给A_Table 一个autoincrement PK 通过declared_attr id()：

@declared_attr
def id(cls):
    return Column(Integer, cls.seq, unique=True, autoincrement=True, primary_key=True)

同样，B_Table 是在 Base 之外定义的，但 PK 在 possibleSolution() 中被覆盖，因此它变成了 ForeignKey 到 A_Table：

PrimaryKeyConstraint(Column('A_TableId', Integer(), ForeignKey('A_Table.id'), table=<B_Table>, primary_key=True, nullable=False))

然后，我们实例化一个不带任何kwargs的A_Table实例，并在构造b时立即将实例a的id属性分配给字段A_TableId：

a = A_Table()
b = B_Table(A_TableId=a.id)

此时我们可以停下来检查一下每个的属性值：

print(a.id, b.A_TableId)
# None None

a.id 是 None，因为它是一个 autoincrement，需要由数据库而不是 ORM 填充。所以 SQLAlchemy 直到实例刷新到数据库后才知道它的值。

那么如果我们在将实例a 添加到session 之后包含flush() 操作会发生什么：

a = A_Table()
session.add(a)
session.flush()
b = B_Table(A_TableId=a.id)
print(a.id, b.A_TableId)
# 1 1

因此，通过首先发出 flush，我们得到了 a.id 的值，这意味着我们还有 b.A_TableId 的值。

session.add(b)
session.commit()
# no error

【讨论】：

这适用于 SQLite仅。我在使用本地托管的 PostgreSQL 数据库时复制了错误 InvalidRequestError: Instance <B_Table at 0x1073fe518> cannot be refreshed - it's not persistent and does not contain a full primary key. 。当我 only 将引擎连接字符串更改为内存中的 SQLite 数据库时，错误消失了。顺便说一句，感谢您花时间尝试解决此问题。