langgraph-checkpoint-postgres


Namelanggraph-checkpoint-postgres JSON
Version 2.0.22 PyPI version JSON
download
home_pageNone
SummaryLibrary with a Postgres implementation of LangGraph checkpoint saver.
upload_time2025-07-10 22:45:24
maintainerNone
docs_urlNone
authorNone
requires_python>=3.9
licenseNone
keywords
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # LangGraph Checkpoint Postgres

Implementation of LangGraph CheckpointSaver that uses Postgres.

## Dependencies

By default `langgraph-checkpoint-postgres` installs `psycopg` (Psycopg 3) without any extras. However, you can choose a specific installation that best suits your needs [here](https://www.psycopg.org/psycopg3/docs/basic/install.html) (for example, `psycopg[binary]`).

## Usage

> [!IMPORTANT]
> When using Postgres checkpointers for the first time, make sure to call `.setup()` method on them to create required tables. See example below.

> [!IMPORTANT]
> When manually creating Postgres connections and passing them to `PostgresSaver` or `AsyncPostgresSaver`, make sure to include `autocommit=True` and `row_factory=dict_row` (`from psycopg.rows import dict_row`). See a full example in this [how-to guide](https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/).
>
> **Why these parameters are required:**
> - `autocommit=True`: Required for the `.setup()` method to properly commit the checkpoint tables to the database. Without this, table creation may not be persisted.
> - `row_factory=dict_row`: Required because the PostgresSaver implementation accesses database rows using dictionary-style syntax (e.g., `row["column_name"]`). The default `tuple_row` factory returns tuples that only support index-based access (e.g., `row[0]`), which will cause `TypeError` exceptions when the checkpointer tries to access columns by name.
>
> **Example of incorrect usage:**
> ```python
> # ❌ This will fail with TypeError during checkpointer operations
> with psycopg.connect(DB_URI) as conn:  # Missing autocommit=True and row_factory=dict_row
>     checkpointer = PostgresSaver(conn)
>     checkpointer.setup()  # May not persist tables properly
>     # Any operation that reads from database will fail with:
>     # TypeError: tuple indices must be integers or slices, not str
> ```

```python
from langgraph.checkpoint.postgres import PostgresSaver

write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}

DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # call .setup() the first time you're using the checkpointer
    checkpointer.setup()
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    checkpointer.put(write_config, checkpoint, {}, {})

    # load checkpoint
    checkpointer.get(read_config)

    # list checkpoints
    list(checkpointer.list(read_config))
```

### Async

```python
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpoint = {
        "v": 4,
        "ts": "2024-07-31T20:14:19.804150+00:00",
        "id": "1ef4f797-8335-6428-8001-8a1503f9b875",
        "channel_values": {
            "my_key": "meow",
            "node": "node"
        },
        "channel_versions": {
            "__start__": 2,
            "my_key": 3,
            "start:node": 3,
            "node": 3
        },
        "versions_seen": {
            "__input__": {},
            "__start__": {
            "__start__": 1
            },
            "node": {
            "start:node": 2
            }
        },
    }

    # store checkpoint
    await checkpointer.aput(write_config, checkpoint, {}, {})

    # load checkpoint
    await checkpointer.aget(read_config)

    # list checkpoints
    [c async for c in checkpointer.alist(read_config)]
```

            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "langgraph-checkpoint-postgres",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.9",
    "maintainer_email": null,
    "keywords": null,
    "author": null,
    "author_email": null,
    "download_url": "https://files.pythonhosted.org/packages/56/10/bfab9d031c0eeff9785e28ebcb79107b0a4c57ad3e0f21935679935f77ee/langgraph_checkpoint_postgres-2.0.22.tar.gz",
    "platform": null,
    "description": "# LangGraph Checkpoint Postgres\n\nImplementation of LangGraph CheckpointSaver that uses Postgres.\n\n## Dependencies\n\nBy default `langgraph-checkpoint-postgres` installs `psycopg` (Psycopg 3) without any extras. However, you can choose a specific installation that best suits your needs [here](https://www.psycopg.org/psycopg3/docs/basic/install.html) (for example, `psycopg[binary]`).\n\n## Usage\n\n> [!IMPORTANT]\n> When using Postgres checkpointers for the first time, make sure to call `.setup()` method on them to create required tables. See example below.\n\n> [!IMPORTANT]\n> When manually creating Postgres connections and passing them to `PostgresSaver` or `AsyncPostgresSaver`, make sure to include `autocommit=True` and `row_factory=dict_row` (`from psycopg.rows import dict_row`). See a full example in this [how-to guide](https://langchain-ai.github.io/langgraph/how-tos/persistence_postgres/).\n>\n> **Why these parameters are required:**\n> - `autocommit=True`: Required for the `.setup()` method to properly commit the checkpoint tables to the database. Without this, table creation may not be persisted.\n> - `row_factory=dict_row`: Required because the PostgresSaver implementation accesses database rows using dictionary-style syntax (e.g., `row[\"column_name\"]`). The default `tuple_row` factory returns tuples that only support index-based access (e.g., `row[0]`), which will cause `TypeError` exceptions when the checkpointer tries to access columns by name.\n>\n> **Example of incorrect usage:**\n> ```python\n> # \u274c This will fail with TypeError during checkpointer operations\n> with psycopg.connect(DB_URI) as conn:  # Missing autocommit=True and row_factory=dict_row\n>     checkpointer = PostgresSaver(conn)\n>     checkpointer.setup()  # May not persist tables properly\n>     # Any operation that reads from database will fail with:\n>     # TypeError: tuple indices must be integers or slices, not str\n> ```\n\n```python\nfrom langgraph.checkpoint.postgres import PostgresSaver\n\nwrite_config = {\"configurable\": {\"thread_id\": \"1\", \"checkpoint_ns\": \"\"}}\nread_config = {\"configurable\": {\"thread_id\": \"1\"}}\n\nDB_URI = \"postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable\"\nwith PostgresSaver.from_conn_string(DB_URI) as checkpointer:\n    # call .setup() the first time you're using the checkpointer\n    checkpointer.setup()\n    checkpoint = {\n        \"v\": 4,\n        \"ts\": \"2024-07-31T20:14:19.804150+00:00\",\n        \"id\": \"1ef4f797-8335-6428-8001-8a1503f9b875\",\n        \"channel_values\": {\n            \"my_key\": \"meow\",\n            \"node\": \"node\"\n        },\n        \"channel_versions\": {\n            \"__start__\": 2,\n            \"my_key\": 3,\n            \"start:node\": 3,\n            \"node\": 3\n        },\n        \"versions_seen\": {\n            \"__input__\": {},\n            \"__start__\": {\n            \"__start__\": 1\n            },\n            \"node\": {\n            \"start:node\": 2\n            }\n        },\n    }\n\n    # store checkpoint\n    checkpointer.put(write_config, checkpoint, {}, {})\n\n    # load checkpoint\n    checkpointer.get(read_config)\n\n    # list checkpoints\n    list(checkpointer.list(read_config))\n```\n\n### Async\n\n```python\nfrom langgraph.checkpoint.postgres.aio import AsyncPostgresSaver\n\nasync with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:\n    checkpoint = {\n        \"v\": 4,\n        \"ts\": \"2024-07-31T20:14:19.804150+00:00\",\n        \"id\": \"1ef4f797-8335-6428-8001-8a1503f9b875\",\n        \"channel_values\": {\n            \"my_key\": \"meow\",\n            \"node\": \"node\"\n        },\n        \"channel_versions\": {\n            \"__start__\": 2,\n            \"my_key\": 3,\n            \"start:node\": 3,\n            \"node\": 3\n        },\n        \"versions_seen\": {\n            \"__input__\": {},\n            \"__start__\": {\n            \"__start__\": 1\n            },\n            \"node\": {\n            \"start:node\": 2\n            }\n        },\n    }\n\n    # store checkpoint\n    await checkpointer.aput(write_config, checkpoint, {}, {})\n\n    # load checkpoint\n    await checkpointer.aget(read_config)\n\n    # list checkpoints\n    [c async for c in checkpointer.alist(read_config)]\n```\n",
    "bugtrack_url": null,
    "license": null,
    "summary": "Library with a Postgres implementation of LangGraph checkpoint saver.",
    "version": "2.0.22",
    "project_urls": {
        "Repository": "https://www.github.com/langchain-ai/langgraph"
    },
    "split_keywords": [],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "2b269594505e5698f40e9a76d66c06f412c48a119c3ca41493c432d66f7e1e44",
                "md5": "210d1d5a3c6d426c3805a625430e39ad",
                "sha256": "81623697050ea755abd3cab936e60ae0203c0c492675b16d4d4608da8b586bd5"
            },
            "downloads": -1,
            "filename": "langgraph_checkpoint_postgres-2.0.22-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "210d1d5a3c6d426c3805a625430e39ad",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.9",
            "size": 40339,
            "upload_time": "2025-07-10T22:45:23",
            "upload_time_iso_8601": "2025-07-10T22:45:23.737780Z",
            "url": "https://files.pythonhosted.org/packages/2b/26/9594505e5698f40e9a76d66c06f412c48a119c3ca41493c432d66f7e1e44/langgraph_checkpoint_postgres-2.0.22-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "5610bfab9d031c0eeff9785e28ebcb79107b0a4c57ad3e0f21935679935f77ee",
                "md5": "23cad536193668d144c63056b35034b6",
                "sha256": "4b58346f9d7d44994fc8141310bbd3429fe0e17a18c4a606bf3d7ff673325391"
            },
            "downloads": -1,
            "filename": "langgraph_checkpoint_postgres-2.0.22.tar.gz",
            "has_sig": false,
            "md5_digest": "23cad536193668d144c63056b35034b6",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.9",
            "size": 118024,
            "upload_time": "2025-07-10T22:45:24",
            "upload_time_iso_8601": "2025-07-10T22:45:24.941269Z",
            "url": "https://files.pythonhosted.org/packages/56/10/bfab9d031c0eeff9785e28ebcb79107b0a4c57ad3e0f21935679935f77ee/langgraph_checkpoint_postgres-2.0.22.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-07-10 22:45:24",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "langchain-ai",
    "github_project": "langgraph",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "lcname": "langgraph-checkpoint-postgres"
}
        
Elapsed time: 0.43339s