由于某种原因,以下使用sql alchemy创建数据库的代码:
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)')
尽管如此,
create database db_name;
创建数据库;engine
),因此连接字符串没有问题。我的结论是,这是 AzureSQL 后端的 SQLAlchemy 中的一个错误。是这样吗?除了使用
exec(text(f'CREATE DATABASE test'))
之外还有其他解决方法吗?
尝试使用create_engine而不是create_sync_db_engine。例如:
import urllib
from sqlalchemy import create_engine,text
import pandas as pd
server = 'servername.database.windows.net'
database = 'databasename'
username = 'username'
password = 'password!'
driver = '{ODBC Driver 17 for SQL Server}'
conn = f"""Driver={driver};Server=tcp:{server},1433;Database={database};
Uid={username};Pwd={password};Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;"""
params = urllib.parse.quote_plus(conn)
conn_str = 'mssql+pyodbc:///?autocommit=true&odbc_connect={}'.format(params)
engine = create_engine(conn_str, echo=True)
query = 'SELECT @@version'
with engine.connect() as connection:
df = pd.read_sql_query(sql=text(query), con=connection)
print(df.to_markdown())