async def test_custom_engines(db: None, node_embeddings: List[TextNode]) -> None:
    """Test that PGVectorStore works correctly with custom engines."""
    from sqlalchemy import create_engine
    from sqlalchemy.ext.asyncio import create_async_engine

    # Create custom engines
    engine = create_engine(
        f"postgresql+psycopg2://{PARAMS['user']}:{PARAMS['password']}@{PARAMS['host']}:{PARAMS['port']}/{TEST_DB}",
        echo=False,
    )

    async_engine = create_async_engine(
        f"postgresql+asyncpg://{PARAMS['user']}:{PARAMS['password']}@{PARAMS['host']}:{PARAMS['port']}/{TEST_DB}",
    )

    # Create PGVectorStore with custom engines
    pg = PGVectorStore(
        embed_dim=TEST_EMBED_DIM,
        engine=engine,
        async_engine=async_engine,
    )

    # Test sync add
    pg.add(node_embeddings[:2])

    # Test async add
    await pg.async_add(node_embeddings[2:])

    # Query to verify nodes were added correctly
    q = VectorStoreQuery(query_embedding=_get_sample_vector(0.5), similarity_top_k=10)

    # Test sync query
    res = pg.query(q)
    assert len(res.nodes) == 4
    assert set(res.ids) == {"aaa", "bbb", "ccc", "ddd"}

    # Test async query
    res = await pg.aquery(q)
    assert len(res.nodes) == 4
    assert set(res.ids) == {"aaa", "bbb", "ccc", "ddd"}

    # Clean up
    await pg.aclear()
    await pg.close()
    await async_engine.dispose()
    engine.dispose()
Test that PGVectorStore works correctly with custom engines.
llama-index-integrations/vector_stores/llama-index-vector-stores-postgres/tests/test_postgres.py