# AsyncPostgresStore

> **Class** in `langgraph.store.postgres`

📖 [View in docs](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore)

Asynchronous Postgres-backed store with optional vector search using pgvector.

!!! example "Examples"
    Basic setup and usage:
    ```python
    from langgraph.store.postgres import AsyncPostgresStore

    conn_string = "postgresql://user:pass@localhost:5432/dbname"

    async with AsyncPostgresStore.from_conn_string(conn_string) as store:
        await store.setup()  # Run migrations. Done once

        # Store and retrieve data
        await store.aput(("users", "123"), "prefs", {"theme": "dark"})
        item = await store.aget(("users", "123"), "prefs")
    ```

    Vector search using LangChain embeddings:
    ```python
    from langchain.embeddings import init_embeddings
    from langgraph.store.postgres import AsyncPostgresStore

    conn_string = "postgresql://user:pass@localhost:5432/dbname"

    async with AsyncPostgresStore.from_conn_string(
        conn_string,
        index={
            "dims": 1536,
            "embed": init_embeddings("openai:text-embedding-3-small"),
            "fields": ["text"]  # specify which fields to embed. Default is the whole serialized value
        }
    ) as store:
        await store.setup()  # Run migrations. Done once

        # Store documents
        await store.aput(("docs",), "doc1", {"text": "Python tutorial"})
        await store.aput(("docs",), "doc2", {"text": "TypeScript guide"})
        await store.aput(("docs",), "doc3", {"text": "Other guide"}, index=False)  # don't index

        # Search by similarity
        results = await store.asearch(("docs",), query="programming guides", limit=2)
    ```

    Using connection pooling for better performance:
    ```python
    from langgraph.store.postgres import AsyncPostgresStore, PoolConfig

    conn_string = "postgresql://user:pass@localhost:5432/dbname"

    async with AsyncPostgresStore.from_conn_string(
        conn_string,
        pool_config=PoolConfig(
            min_size=5,
            max_size=20
        )
    ) as store:
        await store.setup()  # Run migrations. Done once
        # Use store with connection pooling...
    ```

## Signature

```python
AsyncPostgresStore(
    self,
    conn: _ainternal.Conn,
    *,
    pipe: AsyncPipeline | None = None,
    deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
    index: PostgresIndexConfig | None = None,
    ttl: TTLConfig | None = None,
)
```

## Description

**Warning:**

Make sure to:
1. Call `setup()` before first use to create necessary tables and indexes
2. Have the pgvector extension available to use vector search
3. Use Python 3.10+ for async functionality

**Note:**

Semantic search is disabled by default. You can enable it by providing an `index` configuration
when creating the store. Without this configuration, all `index` arguments passed to
`put` or `aput` will have no effect.

**Note:**

If you provide a TTL configuration, you must explicitly call `start_ttl_sweeper()` to begin
the background task that removes expired items. Call `stop_ttl_sweeper()` to properly
clean up resources when you're done with the store.

## Extends

- `AsyncBatchedBaseStore`
- `BasePostgresStore[_ainternal.Conn]`

## Constructors

```python
__init__(
    self,
    conn: _ainternal.Conn,
    *,
    pipe: AsyncPipeline | None = None,
    deserializer: Callable[[bytes | orjson.Fragment], dict[str, Any]] | None = None,
    index: PostgresIndexConfig | None = None,
    ttl: TTLConfig | None = None,
) -> None
```

| Name | Type |
|------|------|
| `conn` | `_ainternal.Conn` |
| `pipe` | `AsyncPipeline \| None` |
| `deserializer` | `Callable[[bytes \| orjson.Fragment], dict[str, Any]] \| None` |
| `index` | `PostgresIndexConfig \| None` |
| `ttl` | `TTLConfig \| None` |


## Properties

- `supports_ttl`
- `conn`
- `pipe`
- `lock`
- `loop`
- `supports_pipeline`
- `index_config`
- `embeddings`
- `ttl_config`

## Methods

- [`abatch()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/abatch)
- [`from_conn_string()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/from_conn_string)
- [`setup()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/setup)
- [`sweep_ttl()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/sweep_ttl)
- [`start_ttl_sweeper()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/start_ttl_sweeper)
- [`stop_ttl_sweeper()`](https://reference.langchain.com/python/langgraph.store.postgres/aio/AsyncPostgresStore/stop_ttl_sweeper)

---

[View source on GitHub](https://github.com/langchain-ai/langgraph/blob/398d6cc59d4cf81ab23c09f037e9f521c3fedcd6/libs/checkpoint-postgres/langgraph/store/postgres/aio.py#L42)