FastAPI + SQLAlchemy + SSH + Doris 生产连接问题技术复盘

FastAPI + SQLAlchemy + SSH + Doris 生产连接报错1045

本文内容在公众号免费阅读；
获取方式：
关注公众号 [李指导数据前沿]，回复关键词 “FastApi隧道” 即可获取本文配套的完整源代码文件及高清晰度系统架构图，助你少踩坑，直接起飞！

背景：

将 FastAPI 项目拉到本地，在本地通过ssh_tunnel全局 SSH 跳板管理 连接有SSH的生产环境数据库进行调试；但是报错：
1045, “Access denied for user ‘数据库用户@SSH 服务器 IP’ (using password: YES)”；

但是单独创建一个测试类 doris_ssh_test.py 通过SSH Tunnel建立 127.0.0.1 连接 成功通过SSH跳转到生产数据库没有问题；

关键词：FastAPI、SQLAlchemy 2.x、Doris、SSH 隧道、URL.render_as_string、密码脱敏、1045

适用场景：

• FastAPI + SQLAlchemy 2.x
• 通过统一 SSH 跳板机访问 Doris（MySQL 协议）
• 使用.env/.env.prod管理配置

大致流程图

1. 业务需求 & 技术目标

1.1 业务要求

• 项目使用FastAPI提供接口。
• 数据层包括：
  • MySQL（多个库）
  • Doris（查询分析库）
• 生产环境（APP_ENV=prod）只能通过 SSH 跳板机访问 Doris，不能直连。
• 需要一个统一的 SSH 隧道通道，而不是“一个数据库建一个 SSH”。

目标行为：

当满足：

• APP_ENV=prod(Config.py)

• SSH_TUNNEL__ENABLED=true
  

两个图中的两处配置开启后此时所有数据库连接（包括 Doris、MySQL 等）都必须经由 SSH 隧道。

2. 相关模块总览

本次问题涉及四个关键模块：

1. 配置中心：app/core/config.py

   • 使用pydantic-settings管理多环境.env。
   • 暴露全局settings对象（settings.doris.url、settings.ssh_tunnel等）。

2. 统一 SSH 隧道：app/infra/ssh_tunnel.py

   • 封装SSHTunnelForwarder，为“某个远程 host:port”建立 / 复用本地端口。
   • 暴露get_tunneled_url(original_url: str) -> str：
     • 输入：原始 DB URL（例如 Doris 的mysql+pymysql://...@10.111.160.243:9030/product?...）
     • 输出：改写为通过本地端口的 URL（mysql+pymysql://...@127.0.0.1:随机端口/product?...）

3. Doris 接入：app/infra/doris.py

   • 用 SQLAlchemycreate_engine+sessionmaker构造 Doris Engine 和 Session。
   • 使用get_tunneled_url处理 URL，统一走 SSH。

4. 验证脚本：app/infra/doris_ssh_test.py

   • 独立脚本，通过 SSH 访问 Doris 并执行 SQL，用来验证配置和 SSH 是否正确。

建议在文档中画一张架构图（图 1）：
FastAPI →doris.py→ssh_tunnel.py→ SSH 跳板机 → Doris
方便大家快速理解链路。

3. 配置体系 &get_settings工作方式

3.1 Settings 结构

简化版app/core/config.py关键部分如下：

# app/core/config.py（关键字段）fromenumimportEnumfromtypingimportOptional,ListfrompydanticimportBaseModelfrompydantic_settingsimportBaseSettings,SettingsConfigDict""" 配置环境 """classEnvEnum(str,Enum):local="local"dev="dev"test="test"prod="prod"classMysqlSettings(BaseModel):""" MySQL 相关配置： - main_url: 主库连接串 - back_url: 备库 1 - back2_url: 备库 2 """main_url:str=""back_url:Optional[str]=Noneback2_url:Optional[str]=NoneclassDorisSettings(BaseModel):""" Doris 相关配置： """url:str=""classSshTunnelSettings(BaseModel):""" 全局 SSH 跳板机配置（统一控制 MySQL / Doris 等是否走 SSH）： - enabled: 是否启用 SSH 隧道 - ssh_host / ssh_port: 跳板机地址 - ssh_username / ssh_password: SSH 登录凭据 - private_key_path: 如需走私钥，可在 .env.* 中配置 """enabled:bool=Falsessh_host:str=""ssh_port:int=22ssh_username:str=""ssh_password:str=""classSettings(BaseSettings):""" 全局配置入口。 通过 pydantic-settings 从以下文件加载配置： - .env - .env.<env> （env 由 OS 环境变量 APP_ENV 决定，默认 prod） 嵌套配置使用 env_nested_delimiter="__"： - MYSQL__MAIN_URL -> settings.mysql.main_url - MYSQL__BACK_URL -> settings.mysql.back_url - MYSQL__BACK2_URL -> settings.mysql.back2_url - DORIS__URL -> settings.doris.url - REDIS__URL -> settings.redis.url - SSH_TUNNEL__ENABLED -> settings.ssh_tunnel.enabled 等等。 """model_config=SettingsConfigDict(extra="ignore",env_nested_delimiter="__",)app_name:str="项目名称"debug:bool=Trueapi_prefix:str="/api"# 只是标记当前环境，用于日志展示，真正选择 .env.<env> 的是 OS 的 APP_ENVenv:EnvEnum=EnvEnum.local mysql:MysqlSettings=MysqlSettings()doris:DorisSettings=DorisSettings()ssh_tunnel:SshTunnelSettings=SshTunnelSettings()defget_settings()->"Settings":""" 读取 OS 环境变量 APP_ENV 来决定加载哪一个 .env.<env> 文件。 如果 APP_ENV 非法，则回退为 local。 """env_from_os=os.getenv("APP_ENV","prod")try:env=EnvEnum(env_from_os)exceptValueError:env=EnvEnum.local base_env_file=".env"env_specific_file=f".env.{env.value}"returnSettings(_env_file=[base_env_file,env_specific_file],_env_file_encoding="utf-8",)settings:Settings=get_settings()defdump_critical_config()->None:""" 调试用：打印与环境 / Doris / SSH 隧道相关的关键配置。 """print("=== CONFIG DEBUG ===")print(f"os.getenv('APP_ENV') =