<h3 dir="auto">Describe the bug</h3>
<p dir="auto">I've registered to the <code class="notranslate">before_update</code> hook with:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="class MyTable(Base):
  id = Column(&quot;id&quot;, UUID_DB(as_uuid=True), primary_key=True, default=uuid4)

from sqlalchemy import MetaData, event
event.listen(MyTable, &quot;before_update&quot;, modified_at_hook)

def modified_at_hook(mapper, connection, target):
    ..."><pre class="notranslate"><code class="notranslate">class MyTable(Base):
  id = Column("id", UUID_DB(as_uuid=True), primary_key=True, default=uuid4)

from sqlalchemy import MetaData, event
event.listen(MyTable, "before_update", modified_at_hook)

def modified_at_hook(mapper, connection, target):
    ...
</code></pre></div>
<p dir="auto">then executing an update statement:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="stmt = (
    sqlalchemy.update(MyTable)
    .where(table.id == obj_id)
    .values(**dict_object)
    .execution_options(synchronize_session=&quot;fetch&quot;)
)
async with AsyncSession(db_engine.get_engine()) as async_session:
    result = await async_session.execute(stmt)
    await async_session.commit()"><pre class="notranslate"><code class="notranslate">stmt = (
    sqlalchemy.update(MyTable)
    .where(table.id == obj_id)
    .values(**dict_object)
    .execution_options(synchronize_session="fetch")
)
async with AsyncSession(db_engine.get_engine()) as async_session:
    result = await async_session.execute(stmt)
    await async_session.commit()
</code></pre></div>
<p dir="auto">but the hook is not being triggered.</p>
<p dir="auto">When I am trying to use the <code class="notranslate">merge</code> function of the async_session, only <code class="notranslate">before_insert</code> hook is triggered. and there is no <code class="notranslate">.update</code> method for async session.</p>
<p dir="auto">Further more, when using the <code class="notranslate">.add</code> method, and the entity has id, I get exception that the entity is already exist. how can I mark the entity for update without using <code class="notranslate">SELECT</code>  to retrieve it first?</p>
<h3 dir="auto">Optional link from <a href="https://docs.sqlalchemy.org" rel="nofollow">https://docs.sqlalchemy.org</a> which documents the behavior that is expected</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">SQLAlchemy Version in Use</h3>
<p dir="auto">1.4.40</p>
<h3 dir="auto">DBAPI (i.e. the database driver)</h3>
<p dir="auto">default</p>
<h3 dir="auto">Database Vendor and Major Version</h3>
<p dir="auto">PostgreSQL 12</p>
<h3 dir="auto">Python Version</h3>
<p dir="auto">3.11</p>
<h3 dir="auto">Operating system</h3>
<p dir="auto">losx</p>
<h3 dir="auto">To Reproduce</h3>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="aboe"><pre class="notranslate"><span class="pl-s1">aboe</span></pre></div>
<h3 dir="auto">Error</h3>
<p dir="auto">no error</p>
<h3 dir="auto">Additional context</h3>
<p dir="auto"><em>No response</em></p>
<p dir="auto"><strong>Migrated issue, originally created by Karol Narowski</strong></p>
<p dir="auto">Function is not called when set in listener. <strong>before_insert</strong> and <strong>after_insert</strong> is working well but <strong>before_update</strong> and <strong>after_update</strong> are not called.<br>
Ex.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="class User(Base):
    __tablename__ = 'users'

    uid = Column(String(36), primary_key=True, unique=True, default=generate_uuid)
    name = Column(String(15), nullable=False)
    description = Column(String(25), nullable=False)
    phone = Column(String(20), nullable=False)
    hash = Column(String(128), nullable=True, unique=True)
    created_on = Column(DateTime(), default=datetime.utcnow)
    updated_on = Column(DateTime(), default=datetime.utcnow, onupdate=datetime.utcnow)
@listens_for(User, 'before_insert')
def before_insert_function(mapper, connection, target):
    print(target.phone)
    target.hash = &quot;OtherInsert&quot;


@listens_for(User, 'before_update')
def before_update_function(mapper, connection, target):
    print(target.uid)
    print(target.phone)
    target.hash = &quot;OtherUpdate&quot;
"><pre class="notranslate"><code class="notranslate">class User(Base):
    __tablename__ = 'users'

    uid = Column(String(36), primary_key=True, unique=True, default=generate_uuid)
    name = Column(String(15), nullable=False)
    description = Column(String(25), nullable=False)
    phone = Column(String(20), nullable=False)
    hash = Column(String(128), nullable=True, unique=True)
    created_on = Column(DateTime(), default=datetime.utcnow)
    updated_on = Column(DateTime(), default=datetime.utcnow, onupdate=datetime.utcnow)
@listens_for(User, 'before_insert')
def before_insert_function(mapper, connection, target):
    print(target.phone)
    target.hash = "OtherInsert"


@listens_for(User, 'before_update')
def before_update_function(mapper, connection, target):
    print(target.uid)
    print(target.phone)
    target.hash = "OtherUpdate"

</code></pre></div>