From 50f5de8f39cb40ac20cf56000915722da3e3c7e2 Mon Sep 17 00:00:00 2001 From: Sangkyoon Nam Date: Mon, 5 Oct 2026 00:58:40 +0900 Subject: [PATCH 1/2] fix: list each app once in SQL persister list_app_ids, newest save first Co-Authored-By: Claude Opus 5.5 --- burr/core/persistence.py | 5 ++- burr/integrations/persisters/b_aiosqlite.py | 5 ++- burr/integrations/persisters/b_asyncpg.py | 5 ++- burr/integrations/persisters/b_psycopg2.py | 5 ++- tests/core/test_persistence.py | 19 +++++++++ .../persisters/test_b_aiosqlite.py | 18 ++++++++ .../persisters/test_postgresql.py | 42 +++++++++++++++++++ 7 files changed, 91 insertions(+), 8 deletions(-) diff --git a/burr/core/persistence.py b/burr/core/persistence.py index c32bf8e96..8b852c444 100644 --- a/burr/core/persistence.py +++ b/burr/core/persistence.py @@ -453,9 +453,10 @@ def list_app_ids(self, partition_key: Optional[str], **kwargs) -> list[str]: cursor = self.connection.cursor() try: cursor.execute( - f"SELECT DISTINCT app_id FROM {self.table_name} " + f"SELECT app_id FROM {self.table_name} " f"WHERE partition_key = ? " - f"ORDER BY created_at DESC", + f"GROUP BY app_id " + f"ORDER BY MAX(created_at) DESC", (partition_key,), ) except sqlite3.OperationalError as e: diff --git a/burr/integrations/persisters/b_aiosqlite.py b/burr/integrations/persisters/b_aiosqlite.py index 8439b965b..797f784a9 100644 --- a/burr/integrations/persisters/b_aiosqlite.py +++ b/burr/integrations/persisters/b_aiosqlite.py @@ -196,9 +196,10 @@ async def list_app_ids(self, partition_key: Optional[str] = None, **kwargs) -> l cursor = await self.connection.cursor() await cursor.execute( - f"SELECT DISTINCT app_id FROM {self.table_name} " + f"SELECT app_id FROM {self.table_name} " f"WHERE partition_key = ? " - f"ORDER BY created_at DESC", + f"GROUP BY app_id " + f"ORDER BY MAX(created_at) DESC", (partition_key,), ) app_ids = [row[0] for row in await cursor.fetchall()] diff --git a/burr/integrations/persisters/b_asyncpg.py b/burr/integrations/persisters/b_asyncpg.py index c694350d5..c2fbe6a49 100644 --- a/burr/integrations/persisters/b_asyncpg.py +++ b/burr/integrations/persisters/b_asyncpg.py @@ -289,9 +289,10 @@ async def list_app_ids(self, partition_key: str, **kwargs) -> list[str]: conn, acquired = await self._get_connection() try: query = ( - f"SELECT DISTINCT app_id, created_at FROM {self.table_name} " + f"SELECT app_id FROM {self.table_name} " "WHERE partition_key = $1 " - "ORDER BY created_at DESC" + "GROUP BY app_id " + "ORDER BY MAX(created_at) DESC" ) fetched_data = await conn.fetch(query, partition_key) app_ids = [row[0] for row in fetched_data] diff --git a/burr/integrations/persisters/b_psycopg2.py b/burr/integrations/persisters/b_psycopg2.py index 26425f805..afef55903 100644 --- a/burr/integrations/persisters/b_psycopg2.py +++ b/burr/integrations/persisters/b_psycopg2.py @@ -154,9 +154,10 @@ def list_app_ids(self, partition_key: str, **kwargs) -> list[str]: """Lists the app_ids for a given partition_key.""" cursor = self.connection.cursor() cursor.execute( - f"SELECT DISTINCT app_id, created_at FROM {self.table_name} " + f"SELECT app_id FROM {self.table_name} " "WHERE partition_key = %s " - "ORDER BY created_at DESC", + "GROUP BY app_id " + "ORDER BY MAX(created_at) DESC", (partition_key,), ) app_ids = [row[0] for row in cursor.fetchall()] diff --git a/tests/core/test_persistence.py b/tests/core/test_persistence.py index 18c65a0d1..190eaada2 100644 --- a/tests/core/test_persistence.py +++ b/tests/core/test_persistence.py @@ -74,6 +74,25 @@ def test_persistence_lists_app_ids(persistence): assert set(app_ids) == set(["app_id1", "app_id2"]) +def test_sqlite_persister_list_app_ids_is_unique_and_newest_first(): + """Each app appears once, ordered by its most recent save.""" + persister = SQLLitePersister(db_path=":memory:", table_name="test_table") + persister.initialize() + saves = [ + ("a-app", 0, "2026-01-01 00:00:01"), + ("z-app", 0, "2026-01-01 00:00:02"), + ("a-app", 1, "2026-01-01 00:00:03"), + ] + for app_id, sequence_id, created_at in saves: + persister.save("pk", app_id, sequence_id, "position", State({}), "completed") + persister.connection.execute( + "UPDATE test_table SET created_at = ? WHERE app_id = ? AND sequence_id = ?", + (created_at, app_id, sequence_id), + ) + persister.connection.commit() + assert persister.list_app_ids("pk") == ["a-app", "z-app"] + + def test_persistence_is_initialized_false(initializing_persistence): assert not initializing_persistence.is_initialized() diff --git a/tests/integrations/persisters/test_b_aiosqlite.py b/tests/integrations/persisters/test_b_aiosqlite.py index adb97532c..fd2c679ea 100644 --- a/tests/integrations/persisters/test_b_aiosqlite.py +++ b/tests/integrations/persisters/test_b_aiosqlite.py @@ -73,6 +73,24 @@ async def test_async_persistence_lists_app_ids(async_persistence): assert set(app_ids) == set(["app_id1", "app_id2"]) +async def test_async_persistence_list_app_ids_is_unique_and_newest_first(async_persistence): + """Each app appears once, ordered by its most recent save.""" + await async_persistence.initialize() + saves = [ + ("a-app", 0, "2026-01-01 00:00:01"), + ("z-app", 0, "2026-01-01 00:00:02"), + ("a-app", 1, "2026-01-01 00:00:03"), + ] + for app_id, sequence_id, created_at in saves: + await async_persistence.save("pk", app_id, sequence_id, "position", State({}), "completed") + await async_persistence.connection.execute( + "UPDATE test_table SET created_at = ? WHERE app_id = ? AND sequence_id = ?", + (created_at, app_id, sequence_id), + ) + await async_persistence.connection.commit() + assert await async_persistence.list_app_ids("pk") == ["a-app", "z-app"] + + @pytest.mark.parametrize( "method_name,kwargs", [ diff --git a/tests/integrations/persisters/test_postgresql.py b/tests/integrations/persisters/test_postgresql.py index 0e8cd700a..f1e9ee463 100644 --- a/tests/integrations/persisters/test_postgresql.py +++ b/tests/integrations/persisters/test_postgresql.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. +import datetime import os import pickle @@ -57,6 +58,26 @@ def test_list_app_ids(postgresql_persister): assert "app_id2" in app_ids +def test_list_app_ids_is_unique_and_newest_first(postgresql_persister): + saves = [ + ("a-app", 0, "2026-01-01 00:00:01"), + ("z-app", 0, "2026-01-01 00:00:02"), + ("a-app", 1, "2026-01-01 00:00:03"), + ] + cursor = postgresql_persister.connection.cursor() + for app_id, sequence_id, created_at in saves: + postgresql_persister.save( + "list-order", app_id, sequence_id, "pos", state.State({}), "completed" + ) + cursor.execute( + "UPDATE testtable SET created_at = %s " + "WHERE partition_key = %s AND app_id = %s AND sequence_id = %s", + (created_at, "list-order", app_id, sequence_id), + ) + postgresql_persister.connection.commit() + assert postgresql_persister.list_app_ids("list-order") == ["a-app", "z-app"] + + def test_load_nonexistent_key(postgresql_persister): state_data = postgresql_persister.load("pk", "nonexistent_key") assert state_data is None @@ -158,6 +179,27 @@ async def test_async_list_app_ids(asyncpostgresql_persister): assert "app_id2" in app_ids +async def test_async_list_app_ids_is_unique_and_newest_first(asyncpostgresql_persister): + saves = [ + ("a-app", 0, "2026-01-01 00:00:01"), + ("z-app", 0, "2026-01-01 00:00:02"), + ("a-app", 1, "2026-01-01 00:00:03"), + ] + for app_id, sequence_id, created_at in saves: + await asyncpostgresql_persister.save( + "list-order", app_id, sequence_id, "pos", state.State({}), "completed" + ) + await asyncpostgresql_persister.connection.execute( + "UPDATE testtable_async SET created_at = $1 " + "WHERE partition_key = $2 AND app_id = $3 AND sequence_id = $4", + datetime.datetime.fromisoformat(created_at), + "list-order", + app_id, + sequence_id, + ) + assert await asyncpostgresql_persister.list_app_ids("list-order") == ["a-app", "z-app"] + + async def test_async_load_nonexistent_key(asyncpostgresql_persister): state_data = await asyncpostgresql_persister.load("pk", "nonexistent_key") assert state_data is None From 7031e00e68438b9532d04d83c19db7b1568f552e Mon Sep 17 00:00:00 2001 From: Sangkyoon Nam Date: Mon, 5 Oct 2026 10:39:19 +0900 Subject: [PATCH 2/2] docs: note list_app_ids order for SQL persisters Co-Authored-By: Claude Opus 5.5 --- burr/core/persistence.py | 1 + burr/integrations/persisters/b_aiosqlite.py | 1 + burr/integrations/persisters/b_asyncpg.py | 2 +- burr/integrations/persisters/b_psycopg2.py | 2 +- 4 files changed, 4 insertions(+), 2 deletions(-) diff --git a/burr/core/persistence.py b/burr/core/persistence.py index 8b852c444..c5ca02379 100644 --- a/burr/core/persistence.py +++ b/burr/core/persistence.py @@ -446,6 +446,7 @@ def is_initialized(self) -> bool: return self._initialized def list_app_ids(self, partition_key: Optional[str], **kwargs) -> list[str]: + """Lists the app_ids for a given partition_key, each once, most recently saved first.""" partition_key = ( partition_key if partition_key is not None else SQLitePersister.PARTITION_KEY_DEFAULT ) diff --git a/burr/integrations/persisters/b_aiosqlite.py b/burr/integrations/persisters/b_aiosqlite.py index 797f784a9..e6f56d479 100644 --- a/burr/integrations/persisters/b_aiosqlite.py +++ b/burr/integrations/persisters/b_aiosqlite.py @@ -188,6 +188,7 @@ async def is_initialized(self) -> bool: return self._initialized async def list_app_ids(self, partition_key: Optional[str] = None, **kwargs) -> list[str]: + """Lists the app_ids for a given partition_key, each once, most recently saved first.""" partition_key = ( partition_key if partition_key is not None diff --git a/burr/integrations/persisters/b_asyncpg.py b/burr/integrations/persisters/b_asyncpg.py index c2fbe6a49..39428d7a7 100644 --- a/burr/integrations/persisters/b_asyncpg.py +++ b/burr/integrations/persisters/b_asyncpg.py @@ -285,7 +285,7 @@ async def is_initialized(self) -> bool: await self._release_connection(conn, acquired) async def list_app_ids(self, partition_key: str, **kwargs) -> list[str]: - """Lists the app_ids for a given partition_key.""" + """Lists the app_ids for a given partition_key, each once, most recently saved first.""" conn, acquired = await self._get_connection() try: query = ( diff --git a/burr/integrations/persisters/b_psycopg2.py b/burr/integrations/persisters/b_psycopg2.py index afef55903..c4415404c 100644 --- a/burr/integrations/persisters/b_psycopg2.py +++ b/burr/integrations/persisters/b_psycopg2.py @@ -151,7 +151,7 @@ def is_initialized(self) -> bool: return self._initialized def list_app_ids(self, partition_key: str, **kwargs) -> list[str]: - """Lists the app_ids for a given partition_key.""" + """Lists the app_ids for a given partition_key, each once, most recently saved first.""" cursor = self.connection.cursor() cursor.execute( f"SELECT app_id FROM {self.table_name} "