我有一些触发器,可以在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插入数据),并且我不确定这些问题中的哪一个(如果有的话)实际上是问题所在.但我希望我想要做的事情很清楚,也希望有人能帮我.

推荐答案

这对我很管用:

import datetime
from typing import List

import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import (
    mapped_column,
    DeclarativeBase,
    Mapped,
    MappedAsDataclass,
)


class Base(MappedAsDataclass, DeclarativeBase):
    pass


class Company(Base):
    __tablename__ = 't76903096c'
    id: Mapped[int] = mapped_column(init=False, primary_key=True)
    name: Mapped[str]
    managers: Mapped[List['Manager']] = orm.relationship(
        back_populates='company',
        init=False,
    )


class Employee(Base):
    __tablename__ = 't76903096e'
    id: Mapped[int] = mapped_column(init=False, primary_key=True)
    name: Mapped[str]
    type: Mapped[str] = mapped_column(init=False)
    social: Mapped[str]
    test: Mapped[str] = mapped_column(
        init=False,
        nullable=True,
        server_default=sa.FetchedValue(),
        server_onupdate=sa.FetchedValue(),
    )
    date_added: Mapped[datetime.datetime] = mapped_column(
        init=False,
        nullable=True,
        server_default=sa.FetchedValue(),
        server_onupdate=sa.FetchedValue(),
    )

    __mapper_args__ = {
        'polymorphic_identity': 'employee',
        'polymorphic_on': 'type',
        'eager_defaults': True,
    }


class Manager(Employee):
    __tablename__ = 't76903096m'
    id: Mapped[int] = mapped_column(
        sa.ForeignKey('t76903096e.id'), init=False, primary_key=True
    )
    name: Mapped[str]
    sa.CheckConstraint('name == t76903096e.name', name='check1')
    branch: Mapped[str]

    company_id: Mapped[int] = mapped_column(sa.ForeignKey('t76903096c.id'), init=False)
    company: Mapped[Company] = orm.relationship(
        back_populates='managers'
    )

    __mapper_args__ = {
        'polymorphic_identity': 'manager',
        'eager_defaults': True,
    }


engine = sa.create_engine('postgresql+psycopg2:///test', echo=True)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
Session = orm.sessionmaker(engine)

with engine.begin() as conn:
    # The function and trigger:
    conn.execute(
        sa.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(
        sa.text(
            """
    CREATE TRIGGER new_person
        BEFORE INSERT ON t76903096e
        FOR EACH ROW
        EXECUTE FUNCTION add_date();
    """
        )
    )


company = Company(name='ACME Corp')

manager = Manager(
    name='testname',
    social=23468101112,
    branch='testbranch',
    company=company,
)

with Session.begin() as s:
    s.add(manager)

with Session() as s:
    manager = s.scalar(sa.select(Manager).fetch(count=1))
    print(f'{manager.date_added=}')

本质上,在需要自动生成的列上设置init=False,否则将存在循环依赖(Company.managers).为了便于使用,Manager需要company,而不是company_id.将social设为字符串,因为它对于标准的postgres int来说太大了,而且在任何情况下都不是一个数字.

Python相关问答推荐

如何将 map 数组组合到pyspark中每列的单个 map 中

Pandas使用过滤器映射多列

如何在Python中增量更新DF

有没有方法可以修复删除了换码字符的无效的SON记录?

按照行主要蛇扫描顺序对点列表进行排序

Python -Polars库中的滚动索引?

Pandas 第二小值有条件

Select 用a和i标签包裹的复选框?

点到面的Y距离

Telethon加入私有频道

Pandas—合并数据帧,在公共列上保留非空值,在另一列上保留平均值

numpy卷积与有效

如何创建一个缓冲区周围的一行与manim?

在pandas中使用group_by,但有条件

python中字符串的条件替换

如何根据一列的值有条件地 Select 前N个组,然后按两列分组?

在单个对象中解析多个Python数据帧

SQLAlchemy bindparam在mssql上失败(但在mysql上工作)

OpenGL仅渲染第二个三角形,第一个三角形不可见

Odoo16:模板中使用的docs变量在哪里定义?