如何基于Google Cloud连接器的SQLAlchemy引擎创建vecs客户端

方案1：子类化vecs.Client（推荐） #

通过继承官方vecs.Client类并重写构造方法，直接接收现成的SQLAlchemy引擎作为参数，无需修改原库代码，完全符合项目规范。

代码实现 #

from sqlalchemy import text, MetaData
from sqlalchemy.orm import sessionmaker
from vecs import Client as BaseVecsClient
# 导入你已创建好的SQLAlchemy引擎
from books_vector_db.utils.db import engine


class GCPVecsClient(BaseVecsClient):
    def __init__(self, engine):
        # 复用原Client的核心初始化逻辑，但直接使用传入的引擎
        self.engine = engine
        self.meta = MetaData(schema="vecs")
        self.Session = sessionmaker(bind=self.engine)

        # 执行vecs必需的初始化操作（创建schema、vector扩展）
        with self.Session() as sess:
            with sess.begin():
                sess.execute(text("create schema if not exists vecs;"))
                sess.execute(text("create extension if not exists vector;"))
                self.vector_version = sess.execute(
                    text("select installed_version from pg_available_extensions where name = 'vector' limit 1;")
                ).scalar_one()


def get_vecs_client() -> GCPVecsClient:
    return GCPVecsClient(engine)

优势 #

• 完全继承原vecs.Client的所有方法（如create_collection、upsert、query等），功能不受影响
• 合法扩展原类，无侵入式修改，符合项目代码规范
• 直接复用你已通过GCP Cloud SQL Connector创建的安全连接

方案2：封装连接逻辑为自定义连接字符串（备选） #

如果必须使用原生vecs.Client，可以通过临时替换SQLAlchemy的create_engine函数注入GCP连接器逻辑，让vecs内部创建的引擎复用安全连接。此方案为弱侵入式操作，仅作备选：

代码实现 #

from vecs import Client
from google.cloud.sql.connector import Connector
import sqlalchemy

# 全局初始化GCP连接器
connector = Connector()

def getconn():
    return connector.connect(
        CLOUD_SQL_INSTANCE,
        "pg8000",
        user=DATABASE_USER,
        password=DATABASE_PASSWORD,
        db=DATABASE_NAME,
    )

# 临时替换SQLAlchemy的create_engine注入creator
original_create_engine = sqlalchemy.create_engine

def patched_create_engine(url, **kwargs):
    # 仅对vecs使用的postgresql连接注入GCP连接器逻辑
    if "postgresql" in str(url):
        kwargs["creator"] = getconn
    return original_create_engine(url, **kwargs)

def get_vecs_client() -> Client:
    # 替换后创建vecs客户端
    sqlalchemy.create_engine = patched_create_engine
    client = Client(connection_string="postgresql+pg8000://")
    # 恢复原create_engine避免影响其他逻辑
    sqlalchemy.create_engine = original_create_engine
    return client

注意事项 #

• 需确保创建客户端后及时恢复原create_engine函数，避免影响项目中其他数据库操作
• 仅在无法使用子类化方案时考虑使用

内容的提问来源于stack exchange，提问作者KZiovas