Here’s the situation I see all the time: a Python service writes data just fine, but updating existing rows turns into a messy mix of raw SQL, ad‑hoc string formatting, and confusing transaction behavior. When that happens, bugs pile up—rows don’t change, changes disappear after the request finishes, and nobody knows whether the ORM or the database is at fault. I’ve been on both sides of that incident review.
In this post I’ll walk you through how I update existing rows in SQLAlchemy with clear, repeatable patterns. I’ll show the Core-style update statement, the ORM-style update on objects, bulk updates with filters, and safe transaction handling. I’ll also cover common mistakes I see in teams (and how to avoid them), and when you should avoid updates altogether in favor of inserts or soft-delete strategies. I’ll use PostgreSQL in the examples, but the patterns apply to most relational backends. By the end you’ll have runnable examples, plus the mental model I rely on when I build update-heavy APIs.
The mental model I use: updates are just SQL with guardrails
I treat SQLAlchemy as a tool that helps me express SQL safely, not as a mystery layer. That means I start with the SQL I want, then pick the most appropriate SQLAlchemy pattern to express it. If you keep that mental model, you won’t get stuck wondering why an update “didn’t work.”
In SQL terms, an update is always:
- a target table
- a set of column assignments
- a WHERE clause
- a transaction
If any of those are missing, the update either fails or behaves dangerously. SQLAlchemy gives you guardrails: it will help you build the statement correctly, and it can warn you if you attempt a full-table update without a filter. But you still need to choose the right approach for your workload.
There are two main APIs:
- Core: you build a SQL statement (update, select, insert) and execute it directly.
- ORM: you load objects, mutate them, and commit.
Both are valid. I decide based on scale and clarity:
- If I’m updating a small number of rows and I need ORM features (events, relationship handling, validation logic), I use ORM.
- If I’m updating many rows with a clear filter, I use Core.
In the next section, I’ll build a small schema and a dataset so we can explore both patterns in a runnable way.
Setup: a runnable PostgreSQL example
I’ll keep the schema simple: a books table with an id, price, genre, and name. I’ll show the Core table definition because it maps cleanly to updates. If you already have a database, you can skip the table creation and just point create_engine() at it.
from decimal import Decimal
from sqlalchemy import (
create_engine,
MetaData,
Table,
Column,
Integer,
Numeric,
String,
insert,
)
Replace with your actual connection string
engine = create_engine(
"postgresql+psycopg://username:password@localhost:5432/library"
)
meta = MetaData()
books = Table(
"books",
meta,
Column("bookid", Integer, primarykey=True),
Column("book_price", Numeric(10, 2), nullable=False),
Column("genre", String(50), nullable=False),
Column("book_name", String(200), nullable=False),
)
meta.create_all(engine)
seed_data = [
{"bookid": 1, "bookprice": Decimal("12.20"), "genre": "fiction", "book_name": "Old Age"},
{"bookid": 2, "bookprice": Decimal("13.20"), "genre": "non-fiction", "book_name": "Saturn Rings"},
{"bookid": 3, "bookprice": Decimal("121.60"), "genre": "fiction", "book_name": "Supernova"},
{"bookid": 4, "bookprice": Decimal("100.00"), "genre": "non-fiction", "book_name": "History of the World"},
{"bookid": 5, "bookprice": Decimal("1112.20"), "genre": "fiction", "book_name": "Sun City"},
]
with engine.begin() as conn:
conn.execute(insert(books), seed_data)
A couple of things I’m deliberate about here:
- I use
Decimalfor prices, because floating point math will surprise you. - I use
engine.begin()for a transaction that commits automatically if no errors occur. - I specify a numeric precision so the database enforces a consistent shape.
Now let’s update rows using SQLAlchemy Core, which is the most direct mapping to SQL.
Core-style updates: direct SQL with safe construction
When I need to update a set of rows, I use update() from SQLAlchemy Core. It keeps the SQL explicit while still protecting me from unsafe string concatenation.
Here’s a basic pattern that changes all fiction to sci-fi:
from sqlalchemy import update, select
Build the UPDATE statement
stmt = (
update(books)
.where(books.c.genre == "fiction")
.values(genre="sci-fi")
)
with engine.begin() as conn:
result = conn.execute(stmt)
# result.rowcount tells you how many rows were affected
print("updated rows:", result.rowcount)
Verify
with engine.connect() as conn:
rows = conn.execute(select(books)).fetchall()
for row in rows:
print(row)
I like this pattern because it mirrors SQL almost exactly. The where() clause is a must: it’s your safety harness. In SQLAlchemy 2.0, if you execute a bulk update without a filter, you can enable warnings that catch it during development.
Updating multiple columns at once
You can set multiple columns in one .values() call:
stmt = (
update(books)
.where(books.c.book_id == 3)
.values(bookprice=Decimal("129.99"), bookname="Supernova (Revised)")
)
with engine.begin() as conn:
conn.execute(stmt)
For single-row updates, you can still use Core. It’s often faster and clearer than loading ORM objects when you don’t need them.
Returning updated rows (PostgreSQL)
PostgreSQL supports RETURNING, which is extremely handy for APIs that need to show what changed. SQLAlchemy Core supports it directly:
stmt = (
update(books)
.where(books.c.book_id == 2)
.values(book_price=Decimal("14.50"))
.returning(books.c.bookid, books.c.bookprice, books.c.book_name)
)
with engine.begin() as conn:
updated = conn.execute(stmt).fetchone()
print(updated)
I use this all the time in APIs. You avoid a second select query, and you can return the authoritative value after database-side logic runs (triggers, defaults, constraints).
Conditional updates with expressions
You can also set values based on current values. SQLAlchemy lets you express that safely:
from sqlalchemy import literal
stmt = (
update(books)
.where(books.c.genre == "sci-fi")
.values(bookprice=books.c.bookprice + literal(5))
)
with engine.begin() as conn:
conn.execute(stmt)
That generates SQL like SET bookprice = bookprice + 5. I use this for counters, price adjustments, and idempotent fixes. It’s both fast and consistent.
Core updates with parameter binding (safe, reusable)
A real service rarely updates just one hard-coded value. You usually update based on input, so parameter binding matters. SQLAlchemy binds parameters for you; you do not need to build strings.
from sqlalchemy import bindparam
stmt = (
update(books)
.where(books.c.book_id == bindparam("bid"))
.values(bookprice=bindparam("newprice"))
)
with engine.begin() as conn:
conn.execute(stmt, {"bid": 1, "new_price": Decimal("10.50")})
conn.execute(stmt, {"bid": 2, "new_price": Decimal("15.25")})
This keeps SQL injection risks out of your updates, and it lets you re-use prepared statements efficiently.
Core updates with a subquery filter
Sometimes you want to update rows based on a related query. Here’s a pattern that changes genre for books above a threshold price:
from sqlalchemy import select
subq = select(books.c.bookid).where(books.c.bookprice > Decimal("100.00"))
stmt = (
update(books)
.where(books.c.bookid.in(subq))
.values(genre="collector")
)
with engine.begin() as conn:
conn.execute(stmt)
You can express complex rules without leaving Core. The generated SQL is still clear and database-friendly.
ORM-style updates: object changes with commit semantics
If you already use SQLAlchemy’s ORM, you can update rows by loading objects, mutating them, and committing. This is great when you want ORM events, validation logic, or relationship management to run.
Here’s a minimal ORM mapping for the same table:
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy import Column, Integer, Numeric, String
Base = declarative_base()
class Book(Base):
tablename = "books"
bookid = Column(Integer, primarykey=True)
book_price = Column(Numeric(10, 2), nullable=False)
genre = Column(String(50), nullable=False)
book_name = Column(String(200), nullable=False)
Use the same engine as before
with Session(engine) as session:
book = session.get(Book, 1)
if book is None:
raise ValueError("Book not found")
book.genre = "sci-fi"
book.book_price = Decimal("12.99")
session.commit()
This is the pattern I use for small updates where a single record matters and I want ORM features like type coercion or validation. The key is that session.commit() is the step that persists changes. If you forget it, nothing happens.
ORM bulk updates (with care)
SQLAlchemy lets you do bulk ORM updates without loading objects. This is closer to Core and is fast, but it bypasses some ORM events.
with Session(engine) as session:
count = (
session.query(Book)
.filter(Book.genre == "fiction")
.update({Book.genre: "sci-fi"}, synchronize_session=False)
)
session.commit()
print("updated rows:", count)
I usually set synchronizesession=False to avoid overhead when I don’t have those objects in memory. If you already loaded the objects, use synchronizesession="fetch" to keep them in sync.
ORM updates with validation logic
One advantage of ORM updates is the ability to centralize validation. I often add simple guards directly in the model or service layer:
def updateprice(session: Session, bookid: int, new_price: Decimal) -> Book:
if new_price <= 0:
raise ValueError("Price must be positive")
book = session.get(Book, book_id)
if book is None:
raise ValueError("Book not found")
book.bookprice = newprice
session.commit()
session.refresh(book)
return book
This kind of function is boring but effective. It prevents invalid updates and makes tests straightforward.
Partial updates with ORM merge
Sometimes you receive partial data and want to update only what’s present. session.merge() can help, but it can also surprise you by overwriting fields with None. I only use merge when I fully trust the incoming object.
updated = Book(bookid=3, bookname="Supernova (Collector Edition)")
with Session(engine) as session:
session.merge(updated)
session.commit()
If you don’t control the data, prefer explicit assignments so you can decide what changes.
Traditional vs modern update patterns
I see a lot of teams mixing raw SQL strings with ORM updates, which leads to inconsistent behavior. Here’s the way I think about it in 2026, with a pragmatic comparison:
Traditional Pattern
Why I Prefer It
—
—
Raw SQL string + cursor.execute
Clear intent, safety, reuse of mappings
Raw SQL with string formatting
Less risk of SQL injection, easier to compose
Two queries: UPDATE then SELECT
Fewer round trips, consistent output
Batch SQL with manual pagination
Predictable transaction size
Custom SQL in scripts
Transaction handling and error visibilityI’m not anti-SQL. I just want my updates to be safe, consistent, and easy to read. SQLAlchemy gives me that without making me give up control.
Common mistakes I fix in code reviews
Here are issues I see repeatedly. I’ll show the error and then the fix I use.
1) Forgetting to commit
Problem:
with Session(engine) as session:
book = session.get(Book, 2)
book.genre = "sci-fi"
# No commit — changes vanish
Fix:
with Session(engine) as session:
book = session.get(Book, 2)
book.genre = "sci-fi"
session.commit()
If you’re building a service, I recommend an explicit transaction boundary per request so this isn’t missed.
2) Updating the entire table by accident
Problem:
stmt = update(books).values(genre="sci-fi") # No WHERE clause
with engine.begin() as conn:
conn.execute(stmt)
Fix:
stmt = update(books).where(books.c.genre == "fiction").values(genre="sci-fi")
I’ve seen this wipe out production data. I recommend turning on SQLAlchemy’s warnings for full-table updates in dev and test environments.
3) Using strings for numeric operations
Problem:
stmt = update(books).where(books.c.bookid == 1).values(bookprice="12.99")
Fix:
stmt = update(books).where(books.c.bookid == 1).values(bookprice=Decimal("12.99"))
This avoids subtle rounding issues. For money, always use Decimal and fixed precision columns.
4) Assuming rowcount is always accurate
Some backends don’t report an exact rowcount for bulk updates. PostgreSQL usually does, but it can be affected by triggers or deferred constraints. I treat rowcount as a hint and verify only when it matters.
5) Updating rows with stale ORM objects
If you update rows in SQLAlchemy Core and still have ORM objects in memory, you can end up with stale data. The fix is simple: expire or refresh them.
with Session(engine) as session:
book = session.get(Book, 1)
# Core update in a separate connection
with engine.begin() as conn:
conn.execute(update(books).where(books.c.book_id == 1).values(genre="sci-fi"))
session.refresh(book) # pull latest data
If you mix Core and ORM heavily, add a refresh step or keep them in separate layers.
6) Forgetting to handle “not found” cases
It’s easy to update a row that doesn’t exist and still return success. I treat updates as a contract: if the row doesn’t exist, raise a clear error. With Core, check rowcount. With ORM, check if the object is None.
stmt = update(books).where(books.c.book_id == 999).values(genre="sci-fi")
with engine.begin() as conn:
result = conn.execute(stmt)
if result.rowcount == 0:
raise ValueError("Book not found")
7) Silent failures due to transaction scope
If you open a connection, run an update, and never commit, your changes may be rolled back when the connection closes. This is common when people mix engine.connect() and engine.begin() without realizing the difference. Use engine.begin() for automatic commit, or call commit() explicitly.
Performance and reliability: what matters in practice
I care about three things in update-heavy systems: transaction size, index usage, and lock time. Here’s how I think about each.
Transaction size
Large updates can lock rows for longer than you expect, especially under concurrent load. If you’re updating thousands of rows, I recommend batching. A typical batch size might be 500–5,000 rows depending on row size and index complexity. In my experience, keeping a batch under a few seconds of execution time is a good baseline.
Here’s a batching pattern using Core and a deterministic filter:
from sqlalchemy import select
BATCH_SIZE = 500
with engine.begin() as conn:
while True:
ids = conn.execute(
select(books.c.book_id)
.where(books.c.genre == "fiction")
.limit(BATCH_SIZE)
).scalars().all()
if not ids:
break
conn.execute(
update(books)
.where(books.c.bookid.in(ids))
.values(genre="sci-fi")
)
This pattern keeps each update small and reduces lock contention.
Index usage
If your WHERE clause isn’t indexed, updates can scan the entire table. For large datasets, that can push query times into seconds or more. If you update by book_id, you’re fine. If you update by genre, add an index if it’s frequently used as a filter. I recommend verifying with EXPLAIN when performance matters.
Lock time and concurrency
Updates lock rows. In a busy system, you want those locks to be as short as possible. This is another reason I use RETURNING instead of doing a separate select: fewer round trips, fewer locks. In a moderate workload, you can expect typical update operations to complete in the 10–50ms range, but a missing index can easily push that into hundreds of milliseconds or more.
Using server-side defaults and triggers
If you rely on triggers or computed columns, always use RETURNING or a follow-up select to fetch the final values. Otherwise, your application might show stale data to the user.
A note on isolation levels
Most services run on the default isolation level (often READ COMMITTED). That’s usually fine. If you require stronger guarantees, you might use REPEATABLE READ or SERIALIZABLE, but they can increase lock contention and reduce throughput. I only go there when the business logic demands it.
When I update, and when I avoid it
Not every change should be an update. Here’s how I decide:
I update when:
- It’s a true correction (price fix, name edit, status update).
- The row represents a mutable entity (user profile, inventory count).
- The change is small and reversible.
I avoid updates when:
- I need a full audit trail. I prefer append-only with a new record and a pointer to the latest version.
- The data is event-based. I’d rather insert an event and derive state.
- Regulatory or compliance rules require immutability.
A practical compromise is soft updates: add updatedat, updatedby, and maybe previous_value if you need auditability without a full event store.
Safe update workflows with modern tooling (2026)
I’ll keep this short, but it matters. In 2026, I rarely push update logic without a workflow that catches errors early. Here’s what I do:
- Static checks: I run linters and type checkers so I don’t accidentally pass a string where a decimal is expected. SQLAlchemy’s typing in 2.x is good enough that mypy can catch obvious mistakes.
- Migration reviews: I avoid surprise schema changes that shift how updates behave. A column constraint added without a data backfill is a classic outage cause.
- Staging tests: I run realistic update flows against a staging database to catch transaction issues and lock contention.
- Observability: I log slow updates with context, and I monitor for deadlocks or repeated rollbacks.
This workflow sounds heavy, but it pays off the first time you avoid a production incident.
A more complete, real-world update example
Let’s build a realistic update function for an API endpoint. The requirements are typical:
- Update
booknameandbookpriceif they are provided. - Validate that price is positive.
- Return the updated row.
- Fail clearly if the book doesn’t exist.
Here’s a Core-based approach using RETURNING:
from typing import Optional
from sqlalchemy import update
class NotFoundError(Exception):
pass
class ValidationError(Exception):
pass
def updatebookcore(
conn,
book_id: int,
book_name: Optional[str] = None,
book_price: Optional[Decimal] = None,
):
values = {}
if book_name is not None:
values["bookname"] = bookname
if book_price is not None:
if book_price <= 0:
raise ValidationError("Price must be positive")
values["bookprice"] = bookprice
if not values:
raise ValidationError("No updates provided")
stmt = (
update(books)
.where(books.c.bookid == bookid)
.values(values)
.returning(books.c.bookid, books.c.bookname, books.c.book_price, books.c.genre)
)
result = conn.execute(stmt).fetchone()
if result is None:
raise NotFoundError("Book not found")
return result
This function is small, testable, and precise. It also avoids an extra SELECT query. For high-throughput APIs, that matters.
Now here’s the ORM equivalent if you need ORM features:
from typing import Optional
from sqlalchemy.orm import Session
def updatebookorm(
session: Session,
book_id: int,
book_name: Optional[str] = None,
book_price: Optional[Decimal] = None,
) -> Book:
book = session.get(Book, book_id)
if book is None:
raise NotFoundError("Book not found")
if book_name is not None:
book.bookname = bookname
if book_price is not None:
if book_price <= 0:
raise ValidationError("Price must be positive")
book.bookprice = bookprice
session.commit()
session.refresh(book)
return book
I choose the Core version for raw speed and explicitness, and the ORM version when I need object relationships, events, or a consistent service layer.
Edge cases you should anticipate
Updates are simple until they aren’t. These are the edge cases that show up in real systems.
1) Lost updates in concurrent requests
If two requests update the same row, the last write wins. That might be fine, or it might be a bug. If you need to prevent lost updates, use optimistic concurrency control with a version column.
Here’s a simple pattern:
from sqlalchemy import Column, Integer
class Book(Base):
tablename = "books"
bookid = Column(Integer, primarykey=True)
version = Column(Integer, nullable=False, default=1)
# other columns...
Then update using a version check:
stmt = (
update(books)
.where(books.c.bookid == bookid)
.where(books.c.version == current_version)
.values(book_price=Decimal("12.99"), version=books.c.version + 1)
)
with engine.begin() as conn:
result = conn.execute(stmt)
if result.rowcount == 0:
raise ValueError("Conflict: record was updated by someone else")
This protects you from overwriting changes silently.
2) Updating rows with nullable columns
If your schema allows nulls, decide explicitly whether None means “clear the value” or “leave it unchanged.” I usually treat None as “no change” unless the API explicitly supports clearing fields.
3) Updates that trigger constraints
If you update columns that participate in unique constraints, you can get integrity errors. Don’t catch those broadly and ignore them. Surface them as clear validation errors.
4) Bulk updates with in-memory objects
If you bulk update via Core or query.update(), your in-memory ORM objects will be stale. Either refresh them or keep that bulk update in a separate layer with no objects loaded.
Alternative approaches: when updates aren’t the best tool
Sometimes you can avoid complex updates by choosing a different model.
Append-only with “current” pointer
Instead of updating a row in place, insert a new version and update a pointer to the current version. This gives you history without a full event store.
Soft updates with audit columns
Add updatedat, updatedby, and maybe a JSONB change_log or metadata column. You still do updates, but you retain a lightweight audit trail.
Event sourcing (only if you need it)
If you need strict immutability and full history, store every change as an event. This is powerful, but it adds complexity. I only use it when compliance or business needs demand it.
Practical scenarios: which pattern I pick
Here are quick decision rules I use in real projects:
- Admin UI editing a single record → ORM update with validation and commit.
- Price adjustments across thousands of items → Core update with a filter and batching.
- API update that must return new data → Core update with
RETURNING. - Bulk change and you don’t need events → Core update or
query.update()withsynchronize_session=False. - Complex business rules with related tables → ORM update (or explicit service layer) so you can load related objects.
These aren’t strict rules, but they keep me consistent.
A clear transaction strategy that avoids surprise rollbacks
A lot of update bugs are actually transaction bugs. Here’s the strategy I use:
- For Core, use
engine.begin()to ensure commit on success. - For ORM, use a
Sessionper request, commit on success, rollback on error. - Never mix
engine.connect()with updates unless you are explicitly committing.
A minimal pattern for ORM in a web service looks like this:
from contextlib import contextmanager
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(bind=engine)
@contextmanager
def session_scope():
session = SessionLocal()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()
Usage
with session_scope() as session:
book = session.get(Book, 1)
book.genre = "sci-fi"
This avoids dangling sessions and the “why didn’t it commit” problem.
Monitoring updates in production
If your app updates rows frequently, you need a feedback loop. I do these things by default:
- Log slow updates with execution time and the filter (not the full statement).
- Track the count of updated rows for key operations.
- Alert on deadlocks or rollback spikes.
- Periodically sample updated rows to verify data integrity.
You don’t need a massive observability stack to do this. Even basic logging with context helps you debug issues faster.
Choosing between Core and ORM: a deeper take
Sometimes teams argue about Core vs ORM as if it’s an ideological choice. I see it as a spectrum:
- ORM-first projects still benefit from Core for bulk updates.
- Core-first projects still benefit from ORM for complex object graphs.
- Mixing them is fine if you understand the implications for caching and session state.
The key is consistency and explicit transaction management. If your team knows which pattern to use in which scenario, updates become predictable.
A quick checklist I use before shipping update code
Before I ship update code, I run through this checklist:
- Does every update have an explicit WHERE clause?
- Is there a clear transaction boundary?
- Are numeric updates using
Decimalor other precise types? - Do I handle the “not found” case?
- Do I need to refresh ORM objects after a Core update?
- Is there an index that supports the WHERE clause?
- Do I need
RETURNINGto avoid a second SELECT?
This takes 30 seconds and saves hours of debugging.
Putting it all together
Updating rows in SQLAlchemy isn’t hard, but it’s easy to get wrong if you treat it as magic. The patterns are straightforward:
- Use Core when you want explicit SQL control and bulk efficiency.
- Use ORM when you want object semantics, validation, and relationship logic.
- Always use transactions and commit on success.
- Prefer
RETURNINGfor updates that need to return data. - Watch for common mistakes like missing WHERE clauses or forgotten commits.
If you stick to these patterns, you’ll avoid most of the update bugs that cause production headaches. You’ll also be able to scale your update-heavy workflows without resorting to raw SQL or fragile hacks.
Updates are just SQL with guardrails. Use the guardrails, and the rest falls into place.


