Compartir a través de


Paso 3: Prueba de concepto de la conexión a SQL con pyodbc

Esta prueba de concepto de ejemplo usa pyodbc para conectarse a una base de datos SQL. En este ejemplo se supone que usa la base de datos de ejemplo AdventureWorksLT.

Nota:

Este ejemplo solo debe considerarse una prueba de concepto. El código de ejemplo se simplifica para mayor claridad y no representa necesariamente los procedimientos recomendados por Microsoft.

Requisitos previos

Conexión y consulta de datos

Conéctese a una base de datos con sus credenciales.

  1. Cree un nuevo archivo llamado app.py.

  2. Agregue una docstring de módulo.

    """
    Connects to a SQL database using pyodbc
    """
    
  3. Importe el paquete pyodbc.

    import pyodbc
    
  4. Cree variables para las credenciales de conexión.

    SERVER = '<server-address>'
    DATABASE = '<database-name>'
    USERNAME = '<username>'
    PASSWORD = '<password>'
    
  5. Cree una variable de cadena de conexión mediante la interpolación de cadenas.

    connectionString = f'DRIVER={{ODBC Driver 18 for SQL Server}};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD}'
    
  6. Use la función para conectarse a una base de datos SQL.

    conn = pyodbc.connect(connectionString) 
    

Ejecutar una consulta

Use una cadena de consulta SQL para ejecutar una consulta y analizar los resultados.

  1. Cree una variable para la cadena de consulta 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. Use cursor.execute para recuperar un conjunto de resultados de una consulta en la base de datos.

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

    Nota:

    Esta función acepta cualquier consulta y devuelve un conjunto de resultados que se puede iterar mediante el uso de cursor.fetchone().

  3. Use cursor.fetchall con un loop foreach para obtener todos los registros de la base de datos. A continuación, imprima los registros.

    records = cursor.fetchall()
    for r in records:
        print(f"{r.CustomerID}\t{r.OrderCount}\t{r.CompanyName}")
    
  4. Guarde el archivo app.py.

  5. Abra un terminal y pruebe la aplicación.

    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
    

Inserte una fila como transacción

En este ejemplo, se ejecuta una sentencia INSERT de forma segura y se pasan parámetros. El paso de parámetros como valores protege la aplicación del ataque por inyección de código SQL.

  1. Importe randrange desde la biblioteca random.

    from random import randrange
    
  2. Genere un número de producto aleatorio.

    productNumber = randrange(1000)
    

    Sugerencia

    La generación de un número de producto aleatorio aquí garantiza que puede ejecutar este ejemplo varias veces.

  3. Cree una cadena de instrucción SQL.

    SQL_STATEMENT = """
    INSERT SalesLT.Product (
    Name, 
    ProductNumber, 
    StandardCost, 
    ListPrice, 
    SellStartDate
    ) OUTPUT INSERTED.ProductID 
    VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
    """
    
  4. Ejecute la instrucción mediante cursor.execute.

    cursor.execute(
        SQL_STATEMENT,
        f'Example Product {productNumber}', 
        f'EXAMPLE-{productNumber}', 
        100,
        200
    )
    
  5. Capture la primera columna del único resultado mediante cursor.fetchval, imprima el identificador único del resultado y, a continuación, confirme la operación como una transacción mediante connection.commit.

    resultId = cursor.fetchval()
    print(f"Inserted Product ID : {resultId}")
    conn.commit()
    

    Sugerencia

    Opcionalmente, puede usar connection.rollback para revertir la transacción.

  6. Cierre el cursor y la conexión mediante cursor.close y connection.close.

    cursor.close()
    conn.close()
    
  7. Guarde el archivo app.py y vuelva a probar la aplicación.

    python app.py
    
    Inserted Product ID : 1001
    

Pasos siguientes