使用SQLAlchemy在Azure SQL创建数据库失败：无法打开请求的数据库

使用以下SQLAlchemy代码尝试在Azure SQL中创建数据库：

from sqlalchemy import Engine
from sqlalchemy_utils import create_database, database_exists    

engine: Engine = create_sync_db_engine()
    
if not database_exists(engine.url):
    create_database(engine.url)

执行create_database时失败，报错如下：

>       return self.loaded_dbapi.connect(*cargs, **cparams)
E       sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42000', '[42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Cannot open database "test" requested by the login. The login failed. (4060) (SQLDriverConnect)')

已知情况：

• 相同代码在PostgreSQL后端可正常运行；
• 手动连接master数据库并执行create database db_name;可成功创建数据库；
• 手动创建数据库后，SQLAlchemy代码可正常操作该数据库，说明连接字符串无问题。

疑问：这是SQLAlchemy针对Azure SQL后端的bug吗？除了使用exec(text(f'CREATE DATABASE test'))之外，还有其他解决方法吗？

这不是SQLAlchemy或sqlalchemy_utils的bug，而是SQL Server（包括Azure SQL）的特性限制：创建数据库必须连接到master系统数据库，而你当前的engine连接的是尚未存在的目标数据库，因此触发“无法打开请求的数据库”错误——PostgreSQL无此限制，所以代码能正常运行。

除了直接执行原生CREATE DATABASE语句外，还有两种更优雅的解决思路：

1. 临时切换到master库创建数据库 #

修改连接字符串，临时创建一个连接到master库的engine，用它执行数据库创建操作：

from sqlalchemy import create_engine, Engine
from sqlalchemy_utils import create_database, database_exists
from urllib.parse import urlparse, urlunparse

# 原目标数据库engine
target_engine: Engine = create_sync_db_engine()

if not database_exists(target_engine.url):
    # 解析原URL，替换数据库名为master
    parsed_url = urlparse(target_engine.url.render_as_string(hide_password=False))
    master_url = urlunparse(parsed_url._replace(path="/master"))
    # 创建连接master的engine
    master_engine = create_engine(master_url)
    # 通过master_engine创建目标数据库
    create_database(target_engine.url, engine=master_engine)

这种方法利用sqlalchemy_utils的create_database支持指定engine参数的特性，让创建操作通过连接master的engine执行，无需硬写原生SQL。

2. 自定义适配SQL Server的创建逻辑 #

如果不想依赖sqlalchemy_utils，可以封装一个兼容SQL Server的创建函数：

from sqlalchemy import Engine, text
from urllib.parse import urlparse, urlunparse
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists

def create_sqlserver_database(engine: Engine):
    db_name = engine.url.database
    # 连接master库执行创建
    parsed_url = urlparse(engine.url.render_as_string(hide_password=False))
    master_url = urlunparse(parsed_url._replace(path="/master"))
    with create_engine(master_url).connect() as conn:
        conn.execute(text(f"CREATE DATABASE [{db_name}]"))
        conn.commit()

# 使用示例
target_engine = create_sync_db_engine()
if not database_exists(target_engine.url):
    create_sqlserver_database(target_engine)

核心逻辑都是绕开SQL Server必须连接master才能创建数据库的限制，通过临时切换连接的数据库完成创建操作。

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