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. 相关模块总览本次问题涉及四个关键模块：配置中心：app/core/config.py使用pydantic-settings管理多环境.env。暴露全局settings对象（settings.doris.url、settings.ssh_tunnel等）。统一 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?...）Doris 接入：app/infra/doris.py用 SQLAlchemycreate_engine+sessionmaker构造 Doris Engine 和 Session。使用get_tunneled_url处理 URL，统一走 SSH。验证脚本：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') =