The Postgres store removes expired items only via the background TTL
sweeper, so between sweeps a read can still return a logically expired
row. Adds an opt-in flag so reads can filter expired rows at query time,
closing the window without depending on sweep cadence.
## Changes
- Add `omit_expired: bool` to `TTLConfig` (default `False`). When unset
or `False`, behavior is unchanged
- When enabled, inject `(expires_at IS NULL OR expires_at > NOW())` into
the read query builders on `BasePostgresStore`, so
`get`/`search`/`list_namespaces` do not surface an expired row:
- **GET** (`_get_batch_GET_ops_queries`): predicate on both the final
SELECT *and* the refresh `UPDATE` — the update is driven from an
unfiltered key list, so gating only the SELECT would hide an expired row
yet still refresh it back to life.
- **SEARCH** (`_prepare_batch_search_queries`): inside the inner scans
(before `LIMIT`/`OFFSET`), which also gates the refresh `UPDATE` fed by
`search_results` and keeps pagination correct
- **list_namespaces** (`_get_batch_list_namespaces_queries`): predicate
appended to the existing scan conditions.
## Testing
- Four behaviors covered sync + async, across the
`default`/`pipe`/`pool` fixtures: expired-unswept row omitted from all
read paths (with a raw SQL check proving the row is still physically
present), default/explicit-`False` still returns it, `refresh_ttl=True`
doesn't resurrect an expired row while still extending live ones, and
search pagination stays correct when an expired row falls inside the
page window.
- Full `checkpoint-postgres` suite green on PG16 (210 passed); `ruff
format`/`check` clean on both packages.
This PR updates the dependencies in all Python packages using `uv lock
--upgrade`.
This is an automated PR created by the UV Lock Upgrade workflow.
To make tests pass:
* linting fixes
* whitespace fixes in snapshots
---------
Co-authored-by: sydney-runkle <54324534+sydney-runkle@users.noreply.github.com>
Co-authored-by: Sydney Runkle <sydneymarierunkle@gmail.com>
The configuration expects the key "fields", not "text_fields": I had
failed to update across all implementations in the original PR
Thank you to Vincent Min for the fix!
---------
Co-authored-by: Vincent Min <93780551+VMinB12@users.noreply.github.com>
- Initializing the store with an 'embedding config' -> this contains the
'dims' (used to create the table) and the encoder object (rn langchain
embeddings object, though that is ......)
- Call setup() -> creates the vector table.
Each document has 1 or more vectors associated with it for each json
path in the embedding config.
Would welcome critique and requests!
Leaving the params as the defaults for pgvector but open to feedback if
you think it's important to be able to more transparently configure that
in setup()
```python
from typing import TypedDict, List, Dict, Any, Optional
from langchain_openai import OpenAIEmbeddings
from langgraph.graph import StateGraph
from langgraph.store.postgres import PostgresStore
emb_config = {
"dims": 1536, # OpenAI embedding dimensions
"embed": OpenAIEmbeddings(model="text-embedding-3-small"),
"distance_type": "cosine",
}
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
store.setup()
# Define the state type for our graph
class State(TypedDict):
query: str
results: Optional[List[Dict[str, Any]]]
def put_stuff(state: State) -> State:
docs = [
("doc1", {"text": "red apple in kitchen"}),
("doc2", {"text": "blue car in garage"}),
("doc3", {"text": "green apple on table"}),
]
for key, value in docs:
store.put(("docs",), key, value)
def search_stuff(state: State) -> State:
"""Search for documents using vector similarity."""
results = store.search(("docs",), query=state["query"])
return {"results": results}
builder = StateGraph(State)
builder.add_node(put_stuff)
builder.add_node(search_stuff)
builder.add_edge("__start__", "put_stuff")
builder.add_edge("put_stuff", "search_stuff")
# Compile
with PostgresStore.from_conn_string(
"postgres://postgres:postgres@localhost:5441",
embedding=emb_config,
) as store:
chain = builder.compile(store=store)
result = chain.invoke({"query": "sour apple"})
# Print results
for doc in result["results"]:
print(doc.key)
print(doc.value)
print(doc.response_metadata)
```
