我需要在postgresql中使用触发器(最好通过SQLALchemy ORM)。然而,我似乎不能得到与创建一个第一基地。作为一个基本测试,我尝试在每次插入行时将字符串“Hello world”放入名为test的列中。
我这样设置我的表:
from __future__ import annotations
from typing import List
import sqlalchemy
from sqlalchemy import create_engine, ForeignKey, Column, Integer, String, CheckConstraint, MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, DeclarativeBase, MappedAsDataclass, mapped_column, relationship
from sqlalchemy.schema import DDL
import datetime
dbUrl = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
engine = create_engine(dbUrl,
echo=True)
metadata = MetaData()
class Base(MappedAsDataclass, DeclarativeBase):
pass
class Company(Base):
__tablename__ = "company"
id: Mapped[int] = mapped_column(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(primary_key=True)
name: Mapped[str]
type: Mapped[str]
test: Mapped[str] = mapped_column(nullable=True)
__mapper_args__ = {
"polymorphic_identity": "employee",
"polymorphic_on": "type",
}
Employee.__table__
Base.metadata.create_all(engine)
class Manager(Employee):
__tablename__ = "manager"
id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
manager_name: Mapped[str]
CheckConstraint("manager_name == employee.name", name="check1")
company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
company: Mapped[Company] = relationship(back_populates="managers")
__mapper_args__ = {
"polymorphic_identity": "manager",
}
class Engineer(Employee):
__tablename__ = "engineer"
id: Mapped[int] = mapped_column(ForeignKey("employee.id"), primary_key=True)
engineer_name: Mapped[str]
company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
company: Mapped[Company] = relationship(back_populates="engineers")
__mapper_args__ = {
"polymorphic_identity": "engineer",
}
Engineer.__table__
Base.metadata.create_all(engine)
字符串
然后尝试创建一个触发器,像这样:
sqlalchemy.event.listen(metadata,
"before_create",
DDL("""CREATE TRIGGER autoCreateEmployee
AFTER INSERT
ON employee
FOR EACH ROW
INSERT INTO employee(test)
VALUES("Hello world"),
0)
"""))
型
创建了表,没有发出任何错误,但据我所知,没有添加触发器。我还尝试使用DBeaver直接在SQL中运行我的测试触发器语句:
CREATE TRIGGER autoCreateEmployee
AFTER INSERT
ON employee
FOR EACH ROW
INSERT INTO employee(test)
VALUES("Hello world")
型
这给了我这个错误:
SQL Error [42601]: ERROR: syntax error at or near "INSERT"
Position: 165
Error position: line: 5 pos: 164
型
我的触发器语句有什么问题?为什么我在通过SQLAlchemy运行时看不到错误?
编辑:我试图改编这个例子:
的数据
更新:根据评论,我又看了一下文档,发现我看到的一些例子甚至不是Postgres。现在,我可以运行命令来创建函数和触发器而不会出错,但据我所知,触发器存在(因为我不能在不删除的情况下重新创建它),但似乎与它应该在的表没有关联。
目前我只是运行原始SQL以避免ORM复杂性:
drop function hello_world;
CREATE FUNCTION hello_world() RETURNS trigger AS $hello_world$
BEGIN
INSERT INTO employee(test) VALUES ("HELLO WORLD!");
RETURN NEW;
END;
$hello_world$ LANGUAGE plpgsql;
drop trigger hello on employee;
CREATE TRIGGER hello
AFTER UPDATE ON employee
FOR EACH ROW
EXECUTE FUNCTION hello_world();
型
编辑2:删除了一个截图,因为它是错误的截图,无论如何都是基于误解。
当前状态:
我的函数是这样的:
CREATE FUNCTION hello_world()
RETURNS trigger
LANGUAGE PLPGSQL
AS $hello_world$
BEGIN
INSERT INTO employee(test) VALUES ('HELLO WORLD!');
END;
$hello_world$
型
我创建触发器如下:
CREATE TRIGGER hello
AFTER INSERT ON employee
FOR EACH ROW
EXECUTE FUNCTION hello_world();
型
但是,它试图在测试列中创建一个仅包含测试字符串的新行,而不是在我刚刚输入数据的行中。我的印象是FOR EACH ROW
意味着该函数将在所有插入的行上执行。很明显我还缺少一些东西。
如果我跑:
INSERT INTO employee (name,type) VALUES ('Merv', 'manager');
型
我明白
SQL Error [23502]: ERROR: null value in column "name" of relation "employee" violates not-null constraint
Detail: Failing row contains (31, null, null, HELLO WORLD!).
Where: SQL statement "INSERT INTO employee(test) VALUES ('HELLO WORLD!')"
PL/pgSQL function hello_world() line 3 at SQL statement
型
所以现在我需要弄清楚如何使用触发器函数来定位刚刚修改过的行。
2条答案
按热度按时间dgsult0t1#
这是触发器的工作代码。感谢大家的耐心等待。
字符串
kq0g1dla2#
可能是因为双引号。通常,双引号用于表或字段的名称&单引号用于字符串常量。
试试这个:
字符串
还有这个
型