Skip to content

fix(postgres): update table names(v3) and add checkpoint_blobs deletion #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
35C4n0r merged 1 commit into from
May 24, 2026
Merged

fix(postgres): update table names(v3) and add checkpoint_blobs deletion #8

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions src/langgraph_ephemeral_checkpointer/_strategies/postgres.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,8 +32,9 @@ class PostgresStrategy(Strategy):
WHERE checkpoint_id > %s
GROUP BY thread_id
"""
BATCH_DELETE_WRITES = "DELETE FROM checkpoint_writes WHERE thread_id = ANY(%s)"
BATCH_DELETE_BLOBS = "DELETE FROM checkpoint_blobs WHERE thread_id = ANY(%s)"
BATCH_DELETE_CHECKPOINTS = "DELETE FROM checkpoints WHERE thread_id = ANY(%s)"
BATCH_DELETE_WRITES = "DELETE FROM writes WHERE thread_id = ANY(%s)"

def __init__(self, checkpointer) -> None:
self._checkpointer = checkpointer
Expand All @@ -60,17 +61,19 @@ def batch_delete(self, thread_ids: list[str], checkpointer) -> None:
if not thread_ids:
return
with self._checkpointer._cursor() as cur:
cur.execute(self.BATCH_DELETE_CHECKPOINTS, (thread_ids,))
cur.execute(self.BATCH_DELETE_WRITES, (thread_ids,))
cur.execute(self.BATCH_DELETE_BLOBS, (thread_ids,))
cur.execute(self.BATCH_DELETE_CHECKPOINTS, (thread_ids,))


class AsyncPostgresStrategy(Strategy):
"""Optimised strategy for AsyncPostgresSaver."""

COLLECT_ALL = PostgresStrategy.COLLECT_ALL
COLLECT_SINCE = PostgresStrategy.COLLECT_SINCE
BATCH_DELETE_CHECKPOINTS = PostgresStrategy.BATCH_DELETE_CHECKPOINTS
BATCH_DELETE_WRITES = PostgresStrategy.BATCH_DELETE_WRITES
BATCH_DELETE_BLOBS = PostgresStrategy.BATCH_DELETE_BLOBS
BATCH_DELETE_CHECKPOINTS = PostgresStrategy.BATCH_DELETE_CHECKPOINTS

def __init__(self, checkpointer) -> None:
self._checkpointer = checkpointer
Expand All @@ -95,5 +98,6 @@ async def abatch_delete(self, thread_ids: list[str], checkpointer) -> None:
if not thread_ids:
return
async with self._checkpointer._cursor() as cur:
await cur.execute(self.BATCH_DELETE_CHECKPOINTS, (thread_ids,))
await cur.execute(self.BATCH_DELETE_WRITES, (thread_ids,))
await cur.execute(self.BATCH_DELETE_BLOBS, (thread_ids,))
await cur.execute(self.BATCH_DELETE_CHECKPOINTS, (thread_ids,))