Repository: langchain-ai/langgraph
Issue category: Bug Report

Issue:
langgraph-checkpoint-postgres: Calls to postgres async checkpointer setup() fail on new postgres db
### Checked other resources

- [X] I added a very descriptive title to this issue.
- [X] I searched the [LangGraph](https://langchain-ai.github.io/langgraph/)/LangChain documentation with the integrated search.
- [X] I used the GitHub search to find a similar question and didn't find it.
- [X] I am sure that this is a bug in LangGraph/LangChain rather than my code.
- [X] I am sure this is better as an issue [rather than a GitHub discussion](https://github.com/langchain-ai/langgraph/discussions/new/choose), since this is a LangGraph bug and not a design question.

### Example Code

```python
# set POSTGRES_URI=...
class CheckpointerManager:
    """
    A manager class to handle checkpointer initialization and lifecycle.
    """
    def __init__(self, conninfo: str, max_pool_size: int = 20):
        self.conninfo = conninfo
        self.max_pool_size = max_pool_size
        self.pool = None
        self.checkpointer = None

    async def setup(self):
        """
        Initialize the connection pool and checkpointer.
        """
        self.pool = AsyncConnectionPool(conninfo=self.conninfo,
                                        max_size=self.max_pool_size,
                                        open=False,timeout=5)
        await self.pool.open(wait=True, timeout=5)
        self.checkpointer = AsyncPostgresSaver(conn=self.pool)
        try:
            await self.checkpointer.setup()
        except Exception as e:
            print(f"Error setting up checkpointer: {e}")
            await self.close()
            raise e
        return self

    async def close(self):
        """
        Close the connection pool and cleanup resources.
        """
        if self.pool:
            await self.pool.close()

    def get_checkpointer(self):
        """
        Get the initialized checkpointer.
        """
        if not self.checkpointer:
            raise RuntimeError("Checkpointer has not been initialized. Call setup() first.")
        return self.checkpointer
checkpointer_manager = CheckpointerManager(os.getenv(POSTGRES_URI))
checkpointer_manager = asyncio.run(checkpointer_manager.setup())
```
```


### Error Message and Stack Trace (if applicable)

```shell
setting up checkpointer: current transaction is aborted, commands ignored until end of transaction block
Traceback (most recent call last):
  File "/home/tai/project/main.py", line 91, in <module>
    main()
  File "/home/tai/project/main.py", line 49, in main
    checkpointer_manager = asyncio.run(checkpointer_manager.setup())
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/runners.py", line 194, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.12/asyncio/base_events.py", line 687, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/home/tai/project/checkpointer/__init__.py", line 31, in setup
    raise e
  File "/home/tai/project/checkpointer/__init__.py", line 27, in setup
    await self.checkpointer.setup()
  File "/home/tai/project/.venv/lib/python3.12/site-packages/langgraph/checkpoint/postgres/aio.py", line 98, in setup
    await cur.execute(migration)
  File "/home/tai/project/.venv/lib/python3.12/site-packages/psycopg/cursor_async.py", line 97, in execute
    raise ex.with_traceback(None)
psycopg.errors.InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block
```


### Description

When running the async postgres setup() function, the first execution for SELECT fails, but the cursor is reused in the except block, so postgres ignores it. The CREATE IF NOT EXISTS statement should be hoisted up and out of the main block such that it runs in a separate transaction context, or a least before the SELECT statement runs. For now, I've got it fixed by running this manually:

```
class CheckpointerManager:
    """
    A manager class to handle checkpointer initialization and lifecycle.
    """
    initialization = """CREATE TABLE IF NOT EXISTS checkpoint_migrations (
    v INTEGER PRIMARY KEY
);"""

    def __init__(self, conninfo: str, max_pool_size: int = 20):
        self.conninfo = conninfo
        self.max_pool_size = max_pool_size
        self.pool = None
        self.checkpointer = None


    async def setup(self):
        """
        Initialize the connection pool and checkpointer.
        """
        self.pool = AsyncConnectionPool(conninfo=self.conninfo,
                                        max_size=self.max_pool_size,
                                        open=False,timeout=5)
        await self.pool.open(wait=True, timeout=5)
        self.checkpointer = AsyncPostgresSaver(conn=self.pool)
        async with self.pool.connection() as conn:
            await conn.execute(self.initialization)
        try:
            await self.checkpointer.setup()
        except Exception as e:
            print(f"Error setting up checkpointer: {e}")
            await self.close()
            raise e
        return self

    async def close(self):
        """
        Close the connection pool and cleanup resources.
        """
        if self.pool:
            await self.pool.close()

    def get_checkpointer(self):
        """
        Get the initialized checkpointer.
        """
        if not self.checkpointer:
            raise RuntimeError("Checkpointer has not been initialized. Call setup() first.")
        return self.checkpointer
checkpointer_manager = CheckpointerManager(os.getenv(POSTGRES_URI))
checkpointer_manager = asyncio.run(checkpointer_manager.setup())
```

(note the `async with self.pool.connection() as conn: ; await conn.execute(self.initialization)` before the `.setup()` call. This fixes it.

### System Info

pip freeze | grep langgraph
langgraph==0.2.53
langgraph-checkpoint==2.0.6
langgraph-checkpoint-postgres==2.0.4
langgraph-checkpoint-sqlite==2.0.1
langgraph-sdk==0.1.36



Hints:

Patch:
diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
index 440cb452e..4c0f5295c 100644
--- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
+++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
@@ -5,7 +5,6 @@
 
 from langchain_core.runnables import RunnableConfig
 from psycopg import AsyncConnection, AsyncCursor, AsyncPipeline, Capabilities
-from psycopg.errors import UndefinedTable
 from psycopg.rows import DictRow, dict_row
 from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool
@@ -81,17 +80,15 @@ async def setup(self) -> None:
         the first time checkpointer is used.
         """
         async with self._cursor() as cur:
-            try:
-                results = await cur.execute(
-                    "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
-                )
-                row = await results.fetchone()
-                if row is None:
-                    version = -1
-                else:
-                    version = row["v"]
-            except UndefinedTable:
+            await cur.execute(self.MIGRATIONS[0])
+            results = await cur.execute(
+                "SELECT v FROM checkpoint_migrations ORDER BY v DESC LIMIT 1"
+            )
+            row = await results.fetchone()
+            if row is None:
                 version = -1
+            else:
+                version = row["v"]
             for v, migration in zip(
                 range(version + 1, len(self.MIGRATIONS)),
                 self.MIGRATIONS[version + 1 :],


Test patch: