我有一些触发器,可以在Postgres数据库上插入时为某些字段生成数据.然而,ORM(SQLAlChemy)似乎并不知道这一点,因为它仍然希望我为这些字段提供值.我以为创建FetchedValue()
个列就足以通知ORM,但似乎并非如此.
我举的最小例子是:
# The table
class People(Base):
__tablename__ = 'people'
id: Mapped[int] = mapped_column(init=False, primary_key=True)
name: Mapped[str]
date_added: Mapped[datetime.datetime] = mapped_column(nullable=True, server_default=FetchedValue(), server_onupdate=FetchedValue())
People.__table__
Base.metadata.create_all(engine)
# The function and trigger:
conn.execute(sqlalchemy.sql.text(
"""
CREATE OR REPLACE FUNCTION add_date()
RETURNS trigger
LANGUAGE PLPGSQL
AS $add_date$
DECLARE
thisdate date;
BEGIN
SELECT CURRENT_DATE INTO thisdate;
NEW.date_added := thisdate;
RETURN NEW;
END;
$add_date$
"""))
conn.execute(sqlalchemy.sql.text(
"""
CREATE TRIGGER new_person
BEFORE INSERT ON people
FOR EACH ROW
EXECUTE FUNCTION add_date();
"""
))
# A test insert without the ORM works as expected
conn.execute(sqlalchemy.sql.text("INSERT INTO people (name) VALUES ('Rando');"))
conn.commit()
但是,当我try 使用ORM创建一个实例时,它抱怨缺少date_added
(由触发器提供).
person = People(
name='NewGuy')
with Session(engine) as session:
session.add(person)
session.commit()
给了我:
Traceback (most recent call last):
File "main.py", line 156, in <module>
person = People(
TypeError: __init__() missing 1 required positional argument: 'date_added'
我看着this part of the documentation,但显然我遗漏了一些东西.
然后是I found this thread,所以我try 这样创建行:
person = People(
name='NewGuy',
date_added=text("DEFAULT"))
这确实起作用了,但在我稍微复杂一点的用例中,text("DEFAULT")
不起作用:
class Company(Base):
__tablename__ = "company"
id: Mapped[int] = mapped_column(init=False, primary_key=True)
name: Mapped[str]
managers: Mapped[List[Manager]] = relationship(back_populates="company")
Company.__table__
Base.metadata.create_all(engine)
class Employee(Base):
__tablename__ = "employee"
id: Mapped[int] = mapped_column(init=False, primary_key=True)
name: Mapped[str]
type: Mapped[str] = mapped_column(init=False)
social: Mapped[int]
test: Mapped[str] = mapped_column(init=False, nullable=True, server_default=FetchedValue(), server_onupdate=FetchedValue())
date_added: Mapped[datetime.datetime] = mapped_column(init=False, nullable=True, server_default=FetchedValue(), server_onupdate=FetchedValue())
__mapper_args__ = {
"polymorphic_identity": "employee",
"polymorphic_on": "type", "eager_defaults": True
}
class Manager(Employee):
__tablename__ = "manager"
id: Mapped[int] = mapped_column(ForeignKey("employee.id"), init=False, primary_key=True)
name: Mapped[str]
CheckConstraint("name == employee.name", name="check1")
branch: Mapped[str]
company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
company: Mapped[Company] = relationship(back_populates="managers", init=False)
__mapper_args__ = {
"polymorphic_identity": "manager", "eager_defaults": True
}
Manager.__table__
Base.metadata.create_all(engine)
manager = Manager(
name='testname',
social=23468101112,
branch='testbranch',
company=text("DEFAULT"),
test=text("DEFAULT"),
type=text("DEFAULT"),
date_added=text("DEFAULT"),
company_id=1)
但仍然不能处理此错误:
File "/home/runner/SQLAlchemy-tests/venv/lib/python3.10/site-packages/sqlalchemy/orm/attributes.py", line 2152, in emit_backref_from_scalar_set_event
instance_state(child),
AttributeError: 'TextClause' object has no attribute '_sa_instance_state'
因此,我的猜测是,在技术层面上,这是两个独立的问题,但重点是我仍然没有做我想做的事情(当一些必需的字段由服务器端触发器填充时,使用SQLALChemy ORM插入数据),并且我不确定这些问题中的哪一个(如果有的话)实际上是问题所在.但我希望我想要做的事情很清楚,也希望有人能帮我.