How to tell SQLAlchemy that required arguments will be supplied server-side by the database?

Question

I have some triggers that generate data for certain fields on insert on a Postgres database. However, the ORM (SQLAlchemy) doesn't seem to know this, as it still wants me to provide values for those fields. I thought that creating the columns with FetchedValue() would be sufficient to inform the ORM, but it seems not.

My minimal example:

# 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()

But when I try to create an instance with the ORM, it complains about the lack of date_added (supplied by the trigger).

person = People(
    name='NewGuy')

with Session(engine) as session:
  session.add(person)
  session.commit()

Gives me:

Traceback (most recent call last):
  File "main.py", line 156, in <module>
    person = People(
TypeError: __init__() missing 1 required positional argument: 'date_added'

I was looking at this part of the documentation, but obviously I was missing something.

Then I found this thread, so I tried creating the row like this:

person = People(
    name='NewGuy',
    date_added=text("DEFAULT"))

That did actually work, but in my slightly more complicated use case text("DEFAULT") did not work:

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)

but still didn't work with this error:

  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'

So my guess is that these are separate issues on a technical level, but the point is that I still haven't done what I want (insert data using the SQLALchemy ORM when some required fields are filled by server-side triggers) and I'm not sure which of these issues (if any) is actually the issue. But I hope it's clear what I'm trying to do and that someone can help me out.

Answer 1

This works for me:

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=}')

Essentially, set init=False on columns that need to be autogenerated and where there would otherwise be a circular dependency (Company.managers). Require company rather than company_id for the Manager for ease of use. Make social a string because it is too large for a standard Postgres int and in any case is semantically not a number.