通过

快速入门：使用适用于 Python 的 mssql-python 驱动程序进行可重复部署

在本快速入门中，你将使用 uv 来管理 Python 脚本的项目依赖项和环境。该脚本连接到你创建并加载了示例数据的数据库。 使用适用于 Python 的 mssql-python 驱动程序连接到数据库并执行基本作，例如读取和写入数据。

驱动程序 mssql-python 不需要 Windows 计算机上的任何外部依赖项。 驱动程序通过单个 pip 安装来安装它所需的一切，使你能够将最新版本的驱动程序用于新脚本，同时不会影响那些没有时间升级和测试的其他脚本。

mssql-python 文档 | mssql-python 源代码 | 包 （PyPi） | uv

先决条件

创建 SQL 数据库

本快速入门要求在 Microsoft SQL Server、Fabric 或 Azure SQL 数据库中的 SQL 数据库上使用 AdventureWorks2022 轻型 架构。

创建项目并运行代码

创建新项目

  1. 在开发目录中打开命令提示符。 如果没有目录，请创建名为“pythonscripts等”的新目录。避免 OneDrive 上的文件夹，同步可能会干扰管理虚拟环境。

  2. 使用 . 创建一个新uv。

    uv init mssql-python-repeatable-qs
cd mssql-python-repeatable-qs

添加依赖项

在同一目录中，安装mssql-python、python-dotenv和rich包。

uv add mssql-python python-dotenv rich

启动 Visual Studio Code

在同一目录中运行以下命令。

code .

更新 pyproject.toml

  1. pyproject.toml 包含项目的元数据。 在喜欢的编辑器中打开该文件。

  2. 查看文件的内容。 它应类似于此示例。 请注意，Python 版本及其依赖项通过mssql-python 使用>=来定义最低版本。 如果更喜欢确切的版本，请将 >= 版本号之前的版本号更改为 ==。 然后，每个包的解析版本存储在 uv.lock 中。 锁定文件确保参与项目的开发人员使用一致的包版本。 它还可确保在将包分发给最终用户时使用完全相同的包版本集。 不应编辑 uv.lock 该文件。

    [project]
name = "mssql-python-repeatable-qs"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "mssql-python>=0.10.0",
    "python-dotenv>=1.1.1",
    "rich>=14.1.0",
]

  3. 更新说明以更具描述性。

    description = "Connects to a SQL database using mssql-python"

  4. 保存并关闭该文件。

更新 main.py

  1. 打开名为main.py的文件。 它应类似于此示例。

    def main():
    print("Hello from mssql-python-repeatable-qs!")

if __name__ == "__main__":
    main()

  2. 在文件的顶部，在包含def main()的行上方添加以下导入。

    小窍门

    如果 Visual Studio Code 在解决包时遇到问题，则需要 更新解释器以使用虚拟环境。

    from os import getenv
from dotenv import load_dotenv
from mssql_python import connect, Connection, Cursor
from rich.console import Console
from rich.progress import Progress, SpinnerColumn, TextColumn
from rich.table import Table
from argparse import ArgumentParser
from time import sleep

  3. 在导入和行 def main()之间添加以下代码。

    def get_results(sleep_time: int = 0) -> None:
 with Progress(
     SpinnerColumn(),
     TextColumn("[progress.description]{task.description}"),
     transient=True,
 ) as progress:
     task = progress.add_task(
         description="Connecting to SQL...")

     cursor = query_sql()

     # Simulate a slow connection for demo purposes
     sleep(sleep_time)

     progress.update(task, description="Formatting results...")

     table = Table(title="Orders by Customer")
     # https://rich.readthedocs.io/en/stable/appendix/colors.html
     table.add_column("Customer ID", style="bright_blue", justify="center")
     table.add_column("Company Name", style="bright_white", justify="left")
     table.add_column("Order Count", style="bold green", justify="right")

     records = cursor.fetchall()
     for r in records:
         table.add_row(f"{r.CustomerID}",
                       f"{r.CompanyName}", f"{r.OrderCount}")

     if cursor:
         cursor.close()

     # Simulate a slow connection for demo purposes
     sleep(sleep_time)

     progress.stop()

     Console().print(table)

  4. 在导入和 def get_results(sleep_time: int = 0) -> None:之间，添加此代码。

    _connection = None

def get_connection() -> Connection:
   global _connection
   if not _connection:
       load_dotenv()
       _connection = connect(getenv("SQL_CONNECTION_STRING"))  # type: ignore
   return _connection

def query_sql() -> Cursor:

   SQL_QUERY = """
     SELECT TOP 5
     c.CustomerID,
     c.CompanyName,
     COUNT(soh.SalesOrderID) AS OrderCount
     FROM
     SalesLT.Customer AS c
     LEFT OUTER JOIN SalesLT.SalesOrderHeader AS soh
     ON c.CustomerID = soh.CustomerID
     GROUP BY
     c.CustomerID,
     c.CompanyName
     ORDER BY
     OrderCount DESC;
   """

   conn = get_connection()
   cursor = conn.cursor()
   cursor.execute(SQL_QUERY)
   return cursor

  5. 查找此代码。

    def main():
    print("Hello from test!")

  6. 将其替换为下面的代码。

    def main() -> None:
   parser = ArgumentParser()
   parser.add_argument("--sleep-time", type=int, default=0,
                       help="Time to sleep in seconds to simulate slow connection")
   args = parser.parse_args()

   if args.sleep_time > 0:
       get_results(args.sleep_time)
   else:
       get_results()

   if _connection:
       _connection.close()

  7. 保存并关闭 main.py。

保存连接字符串

  1. 打开.gitignore文件，并为.env文件添加一个排除项。 文件应类似于此示例。 完成后，请务必保存并关闭它。

    # Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

# Connection strings and secrets
.env

  2. 在当前目录中，创建一个名为 .env 的新文件。

  3. 在 .env 文件中，为连接字符串 SQL_CONNECTION_STRING 添加一个条目。 将此处的示例替换为实际连接字符串值。

    SQL_CONNECTION_STRING="Server=<server_name>;Database={<database_name>};Encrypt=yes;TrustServerCertificate=no;Authentication=ActiveDirectoryInteractive"

    小窍门

    此处使用的连接字符串在很大程度上取决于要连接到的 SQL 数据库的类型。 如果要连接到 Fabric 中的 Azure SQL 数据库 或 SQL 数据库，请使用连接字符串选项卡中的 ODBC 连接字符串。可能需要根据方案调整身份验证类型。 有关连接字符串及其语法的详细信息，请参阅 连接字符串语法参考。

使用 uv run 命令执行脚本

小窍门

若要在 macOS 中使用 Microsoft Entra 身份验证，需要通过 Visual Studio Code 中的 Azure Repos 扩展或通过 az login运行来登录。

  1. 在之前所在的终端窗口中，或打开同一目录的新终端窗口，运行以下命令。

     uv run main.py

  2. 现在，让我们再次运行它，不过速度放慢一些，以便可以看到这两个状态更新。

     uv run main.py --sleep-time 5

    下面是脚本完成时的预期输出。

                             Orders by Customer
┏━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━┓
┃ Customer ID ┃ Company Name                   ┃ Order Count ┃
┡━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━┩
│    29485    │ Professional Sales and Service │           1 │
│    29531    │ Remarkable Bike Store          │           1 │
│    29546    │ Bulk Discount Store            │           1 │
│    29568    │ Coalition Bike Company         │           1 │
│    29584    │ Futuristic Bikes               │           1 │
└─────────────┴────────────────────────────────┴─────────────┘

  3. 若要将脚本部署到另一台计算机，请将除文件夹以外的 .venv 所有文件复制到另一台计算机。 第一次运行时重新创建虚拟环境。

后续步骤

访问mssql-python驱动程序 GitHub 存储库以查看更多示例，提出想法或报告问题。

GitHub 上的 mssql-python 驱动程序