[PR #130] Add Asynchronous Connection Pooling Support to PostgresChatMessageHistory #178

New Issue

2026-02-16T05:16:52-05:00

yindo commented

2026-02-16 05:16:52 -05:00

📋 Pull Request Information

Original PR: https://github.com/langchain-ai/langchain-postgres/pull/130
Author: @shamspias
Created: 10/16/2024
Status: 🔄 Open

Base: main ← Head: fix/async-connection-pooling

📝 Commits (9)

1065477 Add async connection pooling support to PostgresChatMessageHistory
084c619 Add async connection pooling support with corrected parameter passing
cbdc34d Add tests for async connection pooling in PostgresChatMessageHistory
c9a7b38 Update README with async connection pooling usage examples
ce04edc Merge branch 'main' into fix/async-connection-pooling
6e39bf2 Merge branch 'main' into fix/async-connection-pooling
6242ce6 Merge branch 'main' into fix/async-connection-pooling
2c8077a Merge branch 'main' into fix/async-connection-pooling
790ea8d Merge branch 'main' into fix/async-connection-pooling

📊 Changes

3 files changed (+193 additions, -44 deletions)

View changed files

📝 README.md (+65 -15)
📝 langchain_postgres/chat_message_histories.py (+54 -28)
📝 tests/unit_tests/test_chat_histories.py (+74 -1)

📄 Description

This PR adds support for asynchronous connection pooling in the PostgresChatMessageHistory class, addressing issues #122 and #129

Changes Made:

Modified PostgresChatMessageHistory:
- Added a conn_pool parameter to accept an AsyncConnectionPool instance.
- Adjusted the __init__ method to include conn_pool and reordered parameters for improved usability.
- Ensured session_id and table_name can be passed as keyword arguments.
Updated Asynchronous Methods:
- Modified aget_messages and aadd_messages to utilize the connection pool when provided.
- Maintained existing functionality for async_connection to ensure backward compatibility.
Added Unit Tests:
- Introduced test_async_chat_history_with_pool in tests/unit_tests/test_chat_histories.py to verify the new functionality.
Updated Documentation:
- Revised the README to include examples of using PostgresChatMessageHistory with an asynchronous connection pool.
- Adjusted usage examples to reflect the updated parameter order and new conn_pool parameter.

Example Usage:

import uuid
import asyncio

from langchain_core.messages import SystemMessage, AIMessage, HumanMessage
from langchain_postgres import PostgresChatMessageHistory
from psycopg_pool import AsyncConnectionPool

async def main():
    # Database connection string
    conn_info = "postgresql://user:password@host:port/dbname"  # Replace with your connection info

    # Initialize the connection pool
    pool = AsyncConnectionPool(conninfo=conn_info)

    try:
        # Create the table schema (only needs to be done once)
        async with pool.connection() as async_connection:
            table_name = "chat_history"
            await PostgresChatMessageHistory.adrop_table(async_connection, table_name)
            await PostgresChatMessageHistory.acreate_tables(async_connection, table_name)

        session_id = str(uuid.uuid4())

        # Initialize the chat history manager with the connection pool
        chat_history = PostgresChatMessageHistory(
            session_id=session_id,
            table_name=table_name,
            conn_pool=pool
        )

        # Add messages to the chat history asynchronously
        await chat_history.aadd_messages([
            SystemMessage(content="System message"),
            AIMessage(content="AI response"),
            HumanMessage(content="Human message"),
        ])

        # Retrieve messages from the chat history
        messages = await chat_history.aget_messages()
        print(messages)
    finally:
        # Close the connection pool
        await pool.close()

# Run the async main function
asyncio.run(main())

Testing:

Added a new test test_async_chat_history_with_pool in tests/unit_tests/test_chat_histories.py:

async def test_async_chat_history_with_pool() -> None:
    """Test the async chat history using a connection pool."""
    from psycopg_pool import AsyncConnectionPool
    from tests.utils import DSN

    # Initialize the connection pool
    pool = AsyncConnectionPool(conninfo=DSN)
    try:
        table_name = "chat_history"
        session_id = str(uuid.uuid4())

        # Create tables using a connection from the pool
        async with pool.connection() as async_connection:
            await PostgresChatMessageHistory.adrop_table(async_connection, table_name)
            await PostgresChatMessageHistory.acreate_tables(async_connection, table_name)

        # Create PostgresChatMessageHistory with conn_pool
        chat_history = PostgresChatMessageHistory(
            session_id=session_id,
            table_name=table_name,
            conn_pool=pool,
        )

        # Ensure the chat history is empty
        messages = await chat_history.aget_messages()
        assert messages == []

        # Add messages to the chat history
        await chat_history.aadd_messages(
            [
                SystemMessage(content="System message"),
                AIMessage(content="AI response"),
                HumanMessage(content="Human message"),
            ]
        )

        # Retrieve messages from the chat history
        messages = await chat_history.aget_messages()
        assert len(messages) == 3
        assert messages == [
            SystemMessage(content="System message"),
            AIMessage(content="AI response"),
            HumanMessage(content="Human message"),
        ]

        # Clear the chat history
        await chat_history.aclear()
        messages = await chat_history.aget_messages()
        assert messages == []
    finally:
        # Close the connection pool
        await pool.close()

Ensured all existing tests pass, maintaining backward compatibility.

Documentation:

README Updates:
- Adjusted parameter usage in examples to match the updated __init__ method.
- Added a new section demonstrating asynchronous usage with connection pooling.

Notes:

Backward Compatibility:
- Existing code using sync_connection or async_connection continues to work without modifications.
Benefits:
- Improves efficiency by reusing database connections through a connection pool.
- Enhances resource management in asynchronous applications.

Related Issues:

_{🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.}

## 📋 Pull Request Information **Original PR:** https://github.com/langchain-ai/langchain-postgres/pull/130 **Author:** [@shamspias](https://github.com/shamspias) **Created:** 10/16/2024 **Status:** 🔄 Open **Base:** `main` ← **Head:** `fix/async-connection-pooling` --- ### 📝 Commits (9) - [`1065477`](https://github.com/langchain-ai/langchain-postgres/commit/1065477d557bb0ef3820e3c769bccab4d2c57101) Add async connection pooling support to PostgresChatMessageHistory - [`084c619`](https://github.com/langchain-ai/langchain-postgres/commit/084c61953c419d01dba8173e38517225967bcc66) Add async connection pooling support with corrected parameter passing - [`cbdc34d`](https://github.com/langchain-ai/langchain-postgres/commit/cbdc34dc9ea459fa4d44a574d8564804e8af5b0c) Add tests for async connection pooling in PostgresChatMessageHistory - [`c9a7b38`](https://github.com/langchain-ai/langchain-postgres/commit/c9a7b387af6e731b1293d207f6ee7f334a51d4f8) Update README with async connection pooling usage examples - [`ce04edc`](https://github.com/langchain-ai/langchain-postgres/commit/ce04edc19440b7e90a9609967d23fd182aeade15) Merge branch 'main' into fix/async-connection-pooling - [`6e39bf2`](https://github.com/langchain-ai/langchain-postgres/commit/6e39bf20d5f1b8e58010bb386e603c4d022676b8) Merge branch 'main' into fix/async-connection-pooling - [`6242ce6`](https://github.com/langchain-ai/langchain-postgres/commit/6242ce614fd8e8664024e1483ffbdc1cbbec5ff7) Merge branch 'main' into fix/async-connection-pooling - [`2c8077a`](https://github.com/langchain-ai/langchain-postgres/commit/2c8077a31c76448f94384d26bdfd3142e596eea5) Merge branch 'main' into fix/async-connection-pooling - [`790ea8d`](https://github.com/langchain-ai/langchain-postgres/commit/790ea8d1d6560cee2143121b5beb2604c5672111) Merge branch 'main' into fix/async-connection-pooling ### 📊 Changes **3 files changed** (+193 additions, -44 deletions) <details> <summary>View changed files</summary> 📝 `README.md` (+65 -15) 📝 `langchain_postgres/chat_message_histories.py` (+54 -28) 📝 `tests/unit_tests/test_chat_histories.py` (+74 -1) </details> ### 📄 Description This PR adds support for asynchronous connection pooling in the `PostgresChatMessageHistory` class, addressing issues #122 and #129 **Changes Made:** - **Modified `PostgresChatMessageHistory`:** - Added a `conn_pool` parameter to accept an `AsyncConnectionPool` instance. - Adjusted the `__init__` method to include `conn_pool` and reordered parameters for improved usability. - Ensured `session_id` and `table_name` can be passed as keyword arguments. - **Updated Asynchronous Methods:** - Modified `aget_messages` and `aadd_messages` to utilize the connection pool when provided. - Maintained existing functionality for `async_connection` to ensure backward compatibility. - **Added Unit Tests:** - Introduced `test_async_chat_history_with_pool` in `tests/unit_tests/test_chat_histories.py` to verify the new functionality. - **Updated Documentation:** - Revised the README to include examples of using `PostgresChatMessageHistory` with an asynchronous connection pool. - Adjusted usage examples to reflect the updated parameter order and new `conn_pool` parameter. **Example Usage:** ```python import uuid import asyncio from langchain_core.messages import SystemMessage, AIMessage, HumanMessage from langchain_postgres import PostgresChatMessageHistory from psycopg_pool import AsyncConnectionPool async def main(): # Database connection string conn_info = "postgresql://user:password@host:port/dbname" # Replace with your connection info # Initialize the connection pool pool = AsyncConnectionPool(conninfo=conn_info) try: # Create the table schema (only needs to be done once) async with pool.connection() as async_connection: table_name = "chat_history" await PostgresChatMessageHistory.adrop_table(async_connection, table_name) await PostgresChatMessageHistory.acreate_tables(async_connection, table_name) session_id = str(uuid.uuid4()) # Initialize the chat history manager with the connection pool chat_history = PostgresChatMessageHistory( session_id=session_id, table_name=table_name, conn_pool=pool ) # Add messages to the chat history asynchronously await chat_history.aadd_messages([ SystemMessage(content="System message"), AIMessage(content="AI response"), HumanMessage(content="Human message"), ]) # Retrieve messages from the chat history messages = await chat_history.aget_messages() print(messages) finally: # Close the connection pool await pool.close() # Run the async main function asyncio.run(main()) ``` **Testing:** - Added a new test `test_async_chat_history_with_pool` in `tests/unit_tests/test_chat_histories.py`: ```python async def test_async_chat_history_with_pool() -> None: """Test the async chat history using a connection pool.""" from psycopg_pool import AsyncConnectionPool from tests.utils import DSN # Initialize the connection pool pool = AsyncConnectionPool(conninfo=DSN) try: table_name = "chat_history" session_id = str(uuid.uuid4()) # Create tables using a connection from the pool async with pool.connection() as async_connection: await PostgresChatMessageHistory.adrop_table(async_connection, table_name) await PostgresChatMessageHistory.acreate_tables(async_connection, table_name) # Create PostgresChatMessageHistory with conn_pool chat_history = PostgresChatMessageHistory( session_id=session_id, table_name=table_name, conn_pool=pool, ) # Ensure the chat history is empty messages = await chat_history.aget_messages() assert messages == [] # Add messages to the chat history await chat_history.aadd_messages( [ SystemMessage(content="System message"), AIMessage(content="AI response"), HumanMessage(content="Human message"), ] ) # Retrieve messages from the chat history messages = await chat_history.aget_messages() assert len(messages) == 3 assert messages == [ SystemMessage(content="System message"), AIMessage(content="AI response"), HumanMessage(content="Human message"), ] # Clear the chat history await chat_history.aclear() messages = await chat_history.aget_messages() assert messages == [] finally: # Close the connection pool await pool.close() ``` - Ensured all existing tests pass, maintaining backward compatibility. **Documentation:** - **README Updates:** - Adjusted parameter usage in examples to match the updated `__init__` method. - Added a new section demonstrating asynchronous usage with connection pooling. **Notes:** - **Backward Compatibility:** - Existing code using `sync_connection` or `async_connection` continues to work without modifications. - **Benefits:** - Improves efficiency by reusing database connections through a connection pool. - Enhances resource management in asynchronous applications. **Related Issues:** - #122 - #129 --- <sub>🔄 This issue represents a GitHub Pull Request. It cannot be merged through Gitea due to API limitations.</sub>

yindo added the pull-request label 2026-02-16 05:16:52 -05:00

yindo referenced this issue

2026-02-16 05:16:58 -05:00

[PR #178] [MERGED] docs(migration): Add migration script #207

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: langchain-ai/langchain-postgres#178