How to run queries on a Database server using Python

If you are looking for a way to run queries on a Database (DB) server using a Python script then I got you. It is useful for various tasks including but not limited to the Automation of Database Tasks, Data Extraction, and Data Analysis.

Photo by Sunder Muthukumaran on Unsplash

Connection

The first step would be to connect to your SQL server. In the following example, I will show how to connect to the Microsoft SQL server.

import pyodbc

# Connection details
server = "<IP_ADDR>,<PORT_NUM>"
database = "<DB_NAME>"
# Your SQL server credentials
username = ""
password = ""
connection_string = (f"DRIVER={{ODBC Driver 17 for SQL Server}};
SERVER={server};DATABASE={database};UID={username};PWD={password}")
# Establish connection
conn = pyodbc.connect(connection_string)
cursor = conn.cursor()

To connect to an SQL Server database using Python, the pyodbc library provides a straightforward approach. Start by importing pyodbc, which enables Open Database Connectivity (ODBC) database interactions. ODBC is a standard API used to access the DBMS. You'll need to define your connection details: the server's IP address and port, the target database's name, and your SQL Server credentials. These details are combined into a connection_string, where the DRIVER specifies the ODBC driver (in this case, ODBC Driver 17 for SQL Server). This connection string is then passed to pyodbc.connect() to establish a connection. After connecting, a cursor is created with conn.cursor(), which is used to execute SQL queries and interact with the database.

Execute the queries

Once the connection is established, you can move to the execution part.

query = "UPDATE Products SET Price = Price * 1.05 WHERE ProductID = {id};" #sample query
cursor.execute(query)
cursor.commit()

The line cursor.execute(query) runs the SQL query you provide, while cursor.commit() saves/commits any changes made to the database during that query. The commit command is necessary when you’re making changes to the database.

Real-life example

In this example, I am calculating the processing time for every SQL query in my data.

First, let’s generate the data:

# generate_queries.py

import random

queries = """
SELECT * FROM Products WHERE Price > 500;
SELECT * FROM Customers WHERE LastName LIKE 'S%';
SELECT OrderID, CustomerID FROM Orders WHERE OrderDate BETWEEN '2024-07-01' AND '2024-07-05';
SELECT ProductName, Stock FROM Products WHERE Stock < 100;
SELECT COUNT(*) FROM Orders WHERE TotalAmount > 100;
INSERT INTO Products (ProductID, ProductName, Price, Stock) VALUES (6, 'Mouse', 29.99, 150);
UPDATE Products SET Stock = Stock + 10 WHERE ProductID = 1;
DELETE FROM Customers WHERE CustomerID = 5;
SELECT o.OrderID, c.FirstName, c.LastName FROM Orders o JOIN Customers c ON o.CustomerID = c.CustomerID;
SELECT p.ProductName, od.Quantity FROM OrderDetails od JOIN Products p ON od.ProductID = p.ProductID WHERE od.OrderID = 1;
"""

operations = [
    "SELECT * FROM Products WHERE ProductID = {id};",
    "SELECT * FROM Customers WHERE CustomerID = {id};",
    "UPDATE Products SET Price = Price * 1.05 WHERE ProductID = {id};",
    "UPDATE Customers SET LastName = 'Updated' WHERE CustomerID = {id};",
    "DELETE FROM Orders WHERE OrderID = {id};",
    "INSERT INTO Products (ProductID, ProductName, Price, Stock) VALUES ({id}, 'Product{id}', {price}, {stock});",
    "INSERT INTO Customers (CustomerID, FirstName, LastName) VALUES ({id}, 'First{id}', 'Last{id}');",
    "SELECT o.OrderID, c.FirstName, c.LastName FROM Orders o JOIN Customers c ON o.CustomerID = c.CustomerID WHERE o.OrderID = {id};",
    "SELECT p.ProductName, od.Quantity FROM OrderDetails od JOIN Products p ON od.ProductID = p.ProductID WHERE od.OrderID = {id};",
    "SELECT COUNT(*) FROM Orders WHERE OrderID = {id};",
]

additional_queries = "\n".join(
    [
        random.choice(operations).format(
            id=random.randint(1, 100),
            price=random.uniform(10.0, 1000.0),
            stock=random.randint(1, 200),
        )
        for _ in range(990)
    ]
)

all_queries = queries + additional_queries

file_path = "sql_queries.txt"
with open(file_path, "w") as file:
    file.write(all_queries)

This generate_queries.py generates 1000 sample SQL queries and stores them in a .txt format.

Now we have the data, now let’s calculate the processing time.

import pyodbc
import pandas as pd
import time

# Connection details
server = "<IP_ADDR>,<PORT_NUM>"
database = "<DB_NAME>"
username = ""
password = ""
connection_string = f"DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password}"

# Establish connection
conn = pyodbc.connect(connection_string)
cursor = conn.cursor()

# Function to execute a SQL query and measure its execution time
def execute_query(query, cursor):
    start_time = time.time() 
    cursor.execute(query)
    cursor.commit()
    end_time = time.time() 
    return end_time - start_time

# Main function
def main():
  # Read the txt file into a DataFrame
  file_path = "./data/sql_queries.txt" 
  with open(file_path, "r") as file: 
      queries = file.readlines()
  df = pd.DataFrame(queries, columns=["query"]) # Create a DataFrame with a column named 'query'

  # Calculate the processing times
  processing_times = []
  for query in df["query"]:
      try:
          processing_time = execute_query(query.strip(), cursor)
      except Exception as e:
          print(f"Error executing query: {query.strip()}")
          print(f"Error message: {e}")
          processing_time = None
      processing_times.append(processing_time)

  # Append the processing times to the DataFrame
  df["processing_time"] = processing_times

  # Save the DataFrame to a new CSV file
  df.to_csv("queries_with_processing_times.csv", index=False)

main()

# Close the connection
cursor.close()
conn.close()

The execute_query function runs an SQL query, records the time before and after execution, and returns the time taken in milliseconds (ms). The main function reads SQL queries from a text file into a pandas DataFrame, iterates through each query to measure its execution time using execute_query, and stores these times in the DataFrame. Then the DataFrame, now containing both the queries and their execution times, is saved as a CSV file. Lastly, we close the connection.

I hope you found this helpful. Thank you.