基于Meltano SDK开发SQL Singer Tap的参数错误排障求助

问题描述 #

测试自研Tap时运行 poetry run tap-<my tap> --config=ENV 出现以下异常：

TypeError: Invalid argument(s) 'json_serializer','json_deserializer' sent to create_engine(), using configuration Progress_pyodbc/QueuePool/Engine.  Please check that the keyword arguments are appropriate for this combination of components.

背景信息 #

• 开发环境：Windows WSL2 Ubuntu
• 已通过自定义SQLAlchemy方言progress_sa成功连接Progress Openedge，连接字符串：

  pyodbc_connstr = urllib.parse.quote_plus(r"""
      DRIVER={DataDirect 7.1 OpenEdge Wire Protocol};
      HOST=xxxx;
      PORT=xxxx;
      DB=xxxxx;
      UID=sysprogress;
      PWD=;
      DEFAULTSCHEMA=PUB;
  """)                                        
  
  sa_str = "progress+pyodbc:///?odbc_connect={}".format(pyodbc_connstr)
  

• 基于Meltano官方Cookie Cutter搭建，Stream Type=3 - SQL，Auth Type=6 - Custom or N/A
• 已通过 poetry add <本地包路径> 安装progress_sa到项目
• 已完成SDK中所有开发者TODO项，配置好测试用.env文件

解决方案 #

问题根源 #

Meltano SDK的SQL Connector默认会向sqlalchemy.create_engine()传递json_serializer和json_deserializer参数，但Progress OpenEdge的自定义SQLAlchemy方言（progress_sa）对应的引擎不支持这两个参数——它们主要为原生支持JSON类型的数据库设计，不适用于OpenEdge。

修复步骤 #

在你的Tap类（继承自SQLTap）中重写create_engine方法，移除这两个不兼容的参数：

1. 打开Tap主文件（通常是tap_<your_tap_name>/tap.py）
2. 找到继承自SQLTap的类，添加以下重写逻辑：

   from sqlalchemy import create_engine
   from sqlalchemy.engine import URL, Engine
   from singer_sdk.connectors.sql import SQLConnector
   from singer_sdk import SQLTap
   
   class YourTapConnector(SQLConnector):
       def create_engine(self) -> Engine:
           """创建SQLAlchemy引擎，移除不支持的JSON序列化参数"""
           if self.sqlalchemy_url:
               url = self.sqlalchemy_url
           else:
               url = URL.create(
                   self.driver_name,
                   username=self.config.get("user"),
                   password=self.config.get("password"),
                   host=self.config.get("host"),
                   port=self.config.get("port"),
                   database=self.config.get("database"),
               )
   
           # 复制SDK默认配置但移除json_serializer和json_deserializer
           return create_engine(
               url,
               echo=self.config.get("sqlalchemy_echo") or False,
               pool_size=self.config.get("sqlalchemy_pool_size") or 5,
               max_overflow=self.config.get("sqlalchemy_max_overflow") or 10,
           )
   
   class TapYourTapName(SQLTap):
       connector_class = YourTapConnector
       # 保留原有类配置（如name、config_jsonschema等）
   

3. 保存文件后重新运行测试命令：poetry run tap-<my tap> --config=ENV

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