python - 定义后更改 SQLAlchemy 主键
问题描述
问题:简单地说,我试图在已经定义 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()
解决方案
最初,您说的 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
通过PK :autoincrement
_declared_attr
id()
@declared_attr
def id(cls):
return Column(Integer, cls.seq, unique=True, autoincrement=True, primary_key=True)
同样,B_Table
定义为 off,Base
但 PK 被覆盖possibleSolution()
,使其变为 a ForeignKey
to A_Table
:
PrimaryKeyConstraint(Column('A_TableId', Integer(), ForeignKey('A_Table.id'), table=<B_Table>, primary_key=True, nullable=False))
然后,我们实例化一个A_Table
没有任何 kwargs的实例,并在构造时立即将id
实例的属性分配a
给 field :A_TableId
b
a = A_Table()
b = B_Table(A_TableId=a.id)
此时我们可以停下来检查每个的属性值:
print(a.id, b.A_TableId)
# None None
a.id
是None
因为它autoincrement
需要由数据库而不是 ORM 填充。所以 SQLAlchemy 直到实例刷新到数据库之后才知道它的值。
那么如果我们flush()
在添加实例之后包含一个操作会a
发生什么session
:
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
推荐阅读
- python - 如何在某个位置将列从一个数据帧插入/移植到另一个数据帧?
- typescript - 打字稿 - d.ts 文件中的类型未找到
- c# - 对派生类中的重写方法使用基类 Doxygen 注释
- python - Python/ Beautiful Soup 数据显示问题
- node.js - FaunaDB 应用程序返回 401 但凭据很好
- c# - 如何重定向到不同的视图?Asp.net MVC 未加载重新调用的视图
- python - OpenMesh 随机访问一个面属性
- r - 如何为绘图选择观察值?
- reactjs - 我在循环中循环 axios,但我没有得到与请求相关的响应,响应变得混乱
- sql - DBMS 中更新异常的确切含义是什么?