手順 3:pymssql を使用した SQL への接続を概念実証する

[アーティクル]
11/06/2024

この概念実証のサンプルでは、SQL データベースへの接続に使用 pymssql します。このサンプルでは、AdventureWorksLT サンプルデータベースを使用していることを前提としています。

Note

この例は概念実証としてのみ検討してください。わかりやすさのためにサンプルコードは簡略化されており、Microsoft が推奨するベストプラクティスを表しているとは限りません。

前提条件

Python 3
- Pythonをまだお持ちでない場合は、Python ランタイムと Python パッケージインデックス (PyPI) パッケージマネージャーを python.org からインストールします。
- 自身の環境以外を使用しますか？ GitHub Codespacesを使用して devcontainer として開きます。
  - .
PyPI からの pymssql パッケージ。
SQL データベースと資格情報。

データに接続してクエリを実行する

資格情報を使用してデータベースに接続します。

app.py という名前で新しいファイルを作成します。

モジュール docstringを追加します。

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

pymssql パッケージをインポートします。
```
import pymssql
```

関数を pymssql.connect 使用して SQL データベースに接続します。

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

クエリを実行する

SQL クエリ文字列を使用してクエリを実行し、結果を解析します。

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

cursor.executeを使用してデータベースに対するクエリから結果セットを取得します。
```
cursor = conn.cursor()
cursor.execute(SQL_QUERY)
```
Note

この関数は基本的に任意のクエリを受け取り、cursor.fetchone()を使用して反復処理できる結果セットを返します。
cursor.fetchallを foreach ループとともに使用して、データベースからすべてのレコードを取得します。次に、レコードを出力します。
```
records = cursor.fetchall()
for r in records:
    print(f"{r['CustomerID']}\t{r['OrderCount']}\t{r['CompanyName']}")
```
app.py ファイルを保存します。

ターミナルを開き、アプリケーションをテストします。

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インジェクションから保護されます。

randrangeを random ライブラリからインポートします。
```
from random import randrange
```
ランダムな製品番号を生成します。
```
productNumber = randrange(1000)
```
ヒント

ここでランダムな製品番号を生成することで、このサンプルを複数回実行できるようになります。

SQL ステートメント文字列を作成します。

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

cursor.execute を使用してステートメントを実行します。

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

cursor.fetchone を使用して単一の結果をフェッチし、結果の一意識別子を出力した後、connection.commit を使用して操作をトランザクションとしてコミットします。
```
result = cursor.fetchone()
print(f"Inserted Product ID : {result['ProductID']}")
conn.commit()
```
ヒント

必要に応じて、connection.rollbackを使用してトランザクションをロールバックできます。
cursor.close と connection.closeを使用してカーソルと接続を閉じます。
```
cursor.close()
conn.close()
```
app.py ファイルを保存し、アプリケーションをう一度テストする
```
python app.py
```
```
Inserted Product ID : 1001
```

次のステップ

Python デベロッパーセンター。

次の方法で共有