步驟 3：使用 pymssql 連線到 SQL 的概念證明

這個概念證明範例會使用 pymssql 來聯機到 SQL 資料庫。 此範例假設您使用 AdventureWorksLT 範例資料庫。

注意

這個範例只應被視為一個概念證明。 為了清楚起見，已將範例程式碼簡化，而其不一定代表 Microsoft 建議的最佳做法。

必要條件

• Python 3
  • 如果您還沒有 Python，請從 python.org 安裝 Python 執行階段和 Python 套件索引 (PyPI) 套件管理員。
  • 偏好不使用您自己的環境？ 使用 GitHub Codespaces 以 devcontainer 開啟。
    • .
• 來自 PyPI 的 pymssql 套件。
• SQL 資料庫和認證。

連線和查詢資料

使用您的認證連線至資料庫。

1. 建立名為 app.py 的新檔案。

2. 新增模組 docstring。

   """
   Connects to a SQL database using pymssql
   """
   

3. 匯入 pymssql 套件。

   import pymssql
   

4. 使用函式 pymssql.connect 連線到 SQL 資料庫。

   conn = pymssql.connect(
       server='<server-address>',
       user='<username>',
       password='<password>',
       database='<database-name>',
       as_dict=True
   )  
   

執行查詢

使用 SQL 查詢字串來執行查詢並剖析結果。

1. 建立 SQL 查詢字串的變數。

   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;
   """
   

2. 使用 cursor.execute 從資料庫查詢擷取結果集。

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

   注意

   這個函式基本上會接受任何查詢並傳回結果集，您可以使用 cursor.fetchone() 反覆查詢結果集。

3. 使用 cursor.fetchall 搭配 foreach 迴圈，以從資料庫取得所有記錄。 然後列印記錄。

   records = cursor.fetchall()
   for r in records:
       print(f"{r['CustomerID']}\t{r['OrderCount']}\t{r['CompanyName']}")
   

4. 儲存 app.py 檔案。

5. 開啟終端機，並測試應用程式。

   python app.py
   

   29485   1       Professional Sales and Service
   29531   1       Remarkable Bike Store
   29546   1       Bulk Discount Store
   29568   1       Coalition Bike Company
   29584   1       Futuristic Bikes
   

插入一個資料列作為交易

在此範例中，您會安全地執行 INSERT 陳述式，並傳遞參數。 將參數作為值傳遞，可協助您的應用程式防禦 SQL 插入式攻擊。

1. 從 random 程式庫匯入 randrange。

   from random import randrange
   

2. 產生隨機的產品數字。

   productNumber = randrange(1000)
   

   提示

   在這裡產生隨機的產品數字可確保您能多次執行此範例。

3. 建立 SQL 陳述式字串。

   SQL_STATEMENT = """
   INSERT SalesLT.Product (
   Name, 
   ProductNumber, 
   StandardCost, 
   ListPrice, 
   SellStartDate
   ) OUTPUT INSERTED.ProductID 
   VALUES (%s, %s, %s, %s, CURRENT_TIMESTAMP)
   """
   

4. 使用 cursor.execute 執行陳述式。

   cursor.execute(
       SQL_STATEMENT,
       (
           f'Example Product {productNumber}', 
           f'EXAMPLE-{productNumber}', 
           100,
           200
       )
   )
   

5. 使用 cursor.fetchone 擷取單一結果，列印結果的唯一識別碼，然後使用 connection.commit 將作業提交為異動。

   result = cursor.fetchone()
   print(f"Inserted Product ID : {result['ProductID']}")
   conn.commit()
   

   提示

   您可以選擇性地使用 connection.rollback 來回復異動。

6. 使用 cursor.close 和 connection.close 關閉資料指標和關係。

   cursor.close()
   conn.close()
   

7. 儲存 app.py 檔案，然後再次測試應用程式

   python app.py
   

   Inserted Product ID : 1001
   

下一步

Python 開發人員中心。