如何在SQLAlchemy中正确指定数据库URL？密码含@致连接报错

问题描述 #

我用SQLAlchemy将数据库表映射到Python类时，创建引擎遇到URL解析错误。代码如下：

# 创建声明式基类
Base = declarative_base()

class State(Base):
    """关联hbtn_0e_6_usa数据库中states表的类"""
    __tablename__ = 'states'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
    name = Column(String(128), nullable=False)

# 创建连接引擎
Engine = create_engine(f"mysql+mysqldb://{user}:{passwd}@{host}:{port}/{db}")
Base.metadata.create_all(Engine)

报错信息：

Traceback (most recent call last):
...
MySQLdb.OperationalError: (2005, "Unknown MySQL server host '{part of my password}@{host}' (-2)")

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
...
sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (2005, "Unknown MySQL server host '{part of my password}@{host}' (-2)")

注：我的密码包含@字符，推测是该原因导致问题。

尝试用sqlalchemy.engine.URL.create()方法时，又出现新错误：

Engine = sqlalchemy.engine.URL.create(
drivername="mysql+mysqldb",
username="root",
password="la@1993#",
host="localhost",
port=3306,
database="hbtn_0e_6_usa"
)

报错：

AttributeError: 'URL' object has no attribute '_run_ddl_visitor' 

错误原因 #

1. 初始报错：密码中的@字符会被URL解析器误判为「用户名密码」与「主机地址」的分隔符，导致主机地址解析错误，把{部分密码}@{host}当成了完整主机名。
2. 后续报错：URL.create()返回的是URL对象，并非SQLAlchemy的Engine对象，直接传给Base.metadata.create_all()会报错，因为该方法需要接收Engine实例。

正确解决方法 #

方法一：对密码进行URL编码 #

使用urllib.parse.quote_plus()对包含特殊字符的密码编码，再拼接URL：

from urllib.parse import quote_plus
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class State(Base):
    __tablename__ = 'states'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
    name = Column(String(128), nullable=False)

# 对密码进行URL编码，将@转译为%40
encoded_passwd = quote_plus(passwd)
# 创建合法引擎
engine = create_engine(f"mysql+mysqldb://{user}:{encoded_passwd}@{host}:{port}/{db}")
Base.metadata.create_all(engine)

方法二：正确使用URL.create()创建引擎 #

先通过URL.create()生成合法URL对象，再传入create_engine()生成Engine实例，而非直接将URL对象当作引擎：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.engine import URL

Base = declarative_base()

class State(Base):
    __tablename__ = 'states'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=False)
    name = Column(String(128), nullable=False)

# 创建URL对象
db_url = URL.create(
    drivername="mysql+mysqldb",
    username="root",
    password="la@1993#",
    host="localhost",
    port=3306,
    database="hbtn_0e_6_usa"
)
# 基于URL对象创建引擎
engine = create_engine(db_url)
Base.metadata.create_all(engine)

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