Condividi tramite


Introduzione rapida: Distribuzioni ripetibili con il driver mssql-python per Python

In questo tutorial, si userà uv per gestire le dipendenze del progetto e gli ambienti per lo script Python che si connette a un database che hai creato e riempito con dati di esempio. Usare il mssql-python driver per Python per connettersi al database ed eseguire operazioni di base, ad esempio la lettura e la scrittura di dati.

Il mssql-python driver non richiede dipendenze esterne nei computer Windows. Il driver installa tutti gli elementi necessari con un'unica pip installazione, consentendo di usare la versione più recente del driver per i nuovi script senza interrompere altri script che non è necessario aggiornare e testare.

Documentazione mssql-python | Codice sorgente mssql-python | Pacchetto (PyPi) | uv

Prerequisiti


Creare un database SQL

Questa guida introduttiva richiede lo schema AdventureWorks2025 Lightweight su Microsoft SQL Server, database SQL in Fabric o database SQL di Azure.

Creare il progetto ed eseguire il codice

Creare un nuovo progetto

  1. Apri un prompt dei comandi nella directory di sviluppo. Se non è disponibile, creare una nuova directory denominata python, scriptse così via. Evitare cartelle in OneDrive, la sincronizzazione può interferire con la gestione dell'ambiente virtuale.

  2. Creare un nuovo progetto con uv.

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

Aggiungere dipendenze

Nella stessa directory installare i pacchetti mssql-python, python-dotenv e rich.

uv add mssql-python python-dotenv rich

Avviare Visual Studio Code

Nella stessa directory eseguire il comando seguente.

code .

Aggiornare pyproject.toml

  1. Pyproject.toml contiene i metadati per il progetto. Aprire il file nell'editor preferito.

  2. Esaminare il contenuto del file. Dovrebbe essere simile a questo esempio. Si noti la versione di Python e la dipendenza per mssql-python, che usa >= per definire una versione minima. Se si preferisce una versione esatta, cambiare >= prima del numero di versione in ==. Le versioni risolte di ogni pacchetto vengono quindi memorizzate nel uv.lock. Il file di blocco garantisce che gli sviluppatori che lavorano sul progetto usino versioni coerenti del pacchetto. Garantisce inoltre che lo stesso set di versioni del pacchetto venga usato durante la distribuzione del pacchetto agli utenti finali. Non è consigliabile modificare il uv.lock file.

    [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. Aggiornare la descrizione in modo che sia più descrittiva.

    description = "Connects to a SQL database using mssql-python"
    
  4. Salva e chiudi il file.

Aggiornare main.py

  1. Aprire il file denominato main.py. Dovrebbe essere simile a questo esempio.

    def main():
        print("Hello from mssql-python-repeatable-qs!")
    
    if __name__ == "__main__":
        main()
    
  2. Nella parte superiore del file aggiungere le importazioni seguenti sopra la riga con def main().

    Suggerimento

    Se Visual Studio Code non riesce a risolvere i pacchetti, è necessario aggiornare l'interprete per usare l'ambiente virtuale.

    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. Tra le importazioni e la riga con def main(), aggiungere il codice seguente.

    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. Tra le importazioni e def get_results(sleep_time: int = 0) -> None:, aggiungere questo codice.

    _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. Trova questo codice.

    def main():
        print("Hello from test!")
    
  6. Sostituirlo con questo codice.

    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. Salvare e chiudere main.py.

Salvare la stringa di connessione

  1. Aprire il .gitignore file e aggiungere un'esclusione per .env i file. Il file dovrebbe essere simile a questo esempio. Assicurati di salvarlo e chiuderlo quando hai finito.

    # Python-generated files
    __pycache__/
    *.py[oc]
    build/
    dist/
    wheels/
    *.egg-info
    
    # Virtual environments
    .venv
    
    # Connection strings and secrets
    .env
    
  2. Nella directory corrente creare un nuovo file denominato .env.

  3. All'interno del .env file aggiungere una voce per la stringa di connessione denominata SQL_CONNECTION_STRING. Sostituire l'esempio qui con il valore effettivo della stringa di connessione.

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

    Suggerimento

    La stringa di connessione usata qui dipende in gran parte dal tipo di database SQL a cui ci si connette. Se ci si connette a un database SQL di Azure o a un database SQL in Fabric, usare la stringa di connessione ODBC dalla scheda Stringhe di connessione. Potrebbe essere necessario modificare il tipo di autenticazione a seconda dello scenario. Per altre informazioni sulle stringhe di connessione e sulla relativa sintassi, vedere informazioni di riferimento sulla sintassi delle stringhe di connessione.

Usare uv run per eseguire lo script

Suggerimento

Per usare l'autenticazione di Microsoft Entra in macOS, è necessario eseguire l'accesso tramite l'estensione Azure Repos in Visual Studio Code o utilizzando az login tramite l'interfaccia a riga di comando di Azure.

  1. Nella finestra del terminale precedente o in una nuova finestra del terminale aperta nella stessa directory eseguire il comando seguente.

     uv run main.py
    
  2. Ora eseguiamolo di nuovo, ma più lentamente per poter vedere entrambi gli aggiornamenti di stato.

     uv run main.py --sleep-time 5
    

    Ecco l'output previsto al termine dello script.

                             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. Per distribuire lo script in un altro computer, copiare tutti i file ad eccezione della .venv cartella nell'altro computer. L'ambiente virtuale viene ricreato con la prima esecuzione.

Passo successivo

Per altri esempi, visitare il mssql-python repository GitHub del driver per contribuire a idee o segnalare problemi.