在 Python 中使用 SQLAlchemy 连接到 Azure 数据库

Question

我正在尝试使用 Python 中的 SQLAlchemy 连接到 Azure 数据库。

我的代码如下：

engine_azure = \
create_engine('mssql+pyodbc://{Server admin login}:{password}@{Server name}.database.windows.net:1433/{AdventureWorksLT}', echo=True)

我收到以下消息：

C:\ProgramData\Anaconda3\lib\site-packages\sqlalchemy\connectors\pyodbc.py:92: SAWarning: No driver name specified; this is expected by PyODBC when using DSN-less connections
  "No driver name specified; "

然后我运行以下代码：

print(engine_azure.table_names())

我收到以下消息：

DBAPIError: (pyodbc.Error) ('01S00', '[01S00] [Microsoft][ODBC Driver Manager] Invalid connection string attribute (0) (SQLDriverConnect)')

Answer 1

您的连接字符串有两个问题：

1. 根据SQLAlchemy 文档：The delimeters must be URL escaped使用传递精确 pyodbc 字符串时。

2. 而且您也没有指定 sql 驱动程序名称。

您可以使用下面的代码，它在我身边工作正常：

import pyodbc
from sqlalchemy import create_engine
import urllib

params = urllib.parse.quote_plus \ # urllib.parse.quote_plus for python 3
(r'Driver={ODBC Driver 13 for SQL Server};Server=tcp:yourDBServerName.database.windows.net,1433;Database=dbname;Uid=username@dbserverName;Pwd=xxx;Encrypt=yes;TrustServerCertificate=no;Connection Timeout=30;')
conn_str = 'mssql+pyodbc:///?odbc_connect={}'.format(params)
engine_azure = create_engine(conn_str,echo=True)

print('connection is ok')
print(engine_azure.table_names())

测试结果：

对于连接字符串，您可以通过转到 azure 门户 -> 您的数据库 -> 连接字符串（在这种情况下选择 ODBC）来获取它：

Answer 2

这是我在Python3中使用的：

params = urllib.parse.quote_plus(
    'Driver=%s;' % driver +
    'Server=tcp:%s,1433;' % server +
    'Database=%s;' % database +
    'Uid=%s;' % username +
    'Pwd={%s};' % password +
    'Encrypt=yes;' +
    'TrustServerCertificate=no;' +
    'Connection Timeout=30;')

conn_str = 'mssql+pyodbc:///?odbc_connect=' + params
engine = create_engine(conn_str)

Answer 3

步骤 1：安装 Azure SQL 数据库驱动程序

使用官方文档安装新版本的 SQL DB 驱动程序： Linux、MacOS、Windows

对之前答案的主要更新：使用最新支持的数据库驱动程序版本ODBC Driver 17 for SQL Server，而不是过时的版本ODBC Driver 13 for SQL Server或没有明确定义版本的版本，例如SQL Server。

第2步：安装sqlalchemy包

只需在终端中打印：pip install SQLAlchemy

AzureDbConnection第 3 步：在类中包装特定的数据库逻辑

from dataclasses import dataclass
from typing import Dict, Any, Iterable
from pandas import DataFrame
from sqlalchemy import create_engine, inspect
import urllib

@dataclass(frozen=True)
class ConnectionSettings:
    """Connection Settings."""
    server: str
    database: str
    username: str
    password: str
    driver: str = '{ODBC Driver 18 for SQL Server}'
    timeout: int = 30


class AzureDbConnection:
    """
    Azure SQL database connection.
    """
    def __init__(self, conn_settings: ConnectionSettings, echo: bool = False) -> None:
        conn_params = urllib.parse.quote_plus(
            'Driver=%s;' % conn_settings.driver +
            'Server=tcp:%s.database.windows.net,1433;' % conn_settings.server +
            'Database=%s;' % conn_settings.database +
            'Uid=%s;' % conn_settings.username +
            'Pwd=%s;' % conn_settings.password +
            'Encrypt=yes;' +
            'TrustServerCertificate=no;' +
            'Connection Timeout=%s;' % conn_settings.timeout
        )
        conn_string = f'mssql+pyodbc:///?odbc_connect={conn_params}'

        self.db = create_engine(conn_string, echo=echo)

    def connect(self) -> None:
        """Estimate connection."""
        self.conn = self.db.connect()

    def get_tables(self) -> Iterable[str]:
        """Get list of tables."""
        inspector = inspect(self.db)
        return [t for t in inspector.get_table_names()]

    def dispose(self) -> None:
        """Dispose opened connections."""
        self.conn.close()
        self.db.dispose()

对之前答案的主要更新：不要忘记在不再需要数据库引擎时立即关闭连接并显式处置它。

享受！

使用Azure 门户上的 Azure DB 刀片设置连接设置和凭据：

conn_settings = ConnectionSettings(
    server='<db_server_name>', 
    database='<db_name>', 
    username='<user_name>', 
    password='***')

打开数据库连接：

db_conn = AzureDbConnection(conn_settings)
db_conn.connect()

测试连接（例如，获取可用表列表），执行其他操作，最后关闭它：

try:
    for t in db_conn.get_tables():
        print(t)
    # Do another DB-related stuff:
    # ...
finally:
    db_conn.dispose()