Using Python and pyodbc to Operate MS SQL Server: A Step‑by‑Step Guide

The author emphasizes that SQL is fundamental for data work and argues that mastering only basic queries is insufficient, prompting a deeper exploration of programmatic SQL manipulation using Python.

From the basics

A Sql class is introduced to encapsulate the connection and query handling logic. The constructor accepts a database name and an optional server address, establishing a pyodbc connection with a standard connection string and initializing a query log.

<code>import pyodbc
from datetime import datetime
class Sql:
    def __init__(self, database, server="XXVIR00012,55000"):
        self.cnxn = pyodbc.connect(
            "Driver={SQL Server Native Client 11.0};"
            "Server=" + server + ";"
            "Database=" + database + ";"
            "Trusted_Connection=yes;"
        )
        self.query = "-- {}\n\n-- Made in Python".format(datetime.now().strftime("%d/%m/%Y"))
</code>

Instantiating the class is straightforward:

<code>sql = Sql('database123')</code>

The class contains several utility methods:

push_dataframe : creates a table based on a pandas DataFrame schema, inserts rows in configurable batch sizes, and logs the generated SQL.

manual : executes arbitrary SQL statements, optionally returning results as a DataFrame.

union : builds a UNION query from a list of tables and stores the result in a new table.

drop : removes one or multiple tables safely.

Example of the push_dataframe implementation:

<code>def push_dataframe(self, data, table="raw_data", batchsize=500):
    cursor = self.cnxn.cursor()
    cursor.fast_executemany = True
    query = "CREATE TABLE [" + table + "] (\n"
    for i in range(len(list(data))):
        query += "\t[{}] varchar(255)".format(list(data)[i])
        if i != len(list(data)) - 1:
            query += ",\n"
        else:
            query += "\n);"
    cursor.execute(query)
    self.cnxn.commit()
    self.query += "\n\n-- create table\n" + query
    insert_query = "INSERT INTO [{}] ({}) VALUES (" + ", ?" * (len(list(data)) - 1) + ")"
    insert_query = insert_query.format(table, '[' + '], ['.join(list(data)) + ']')
    for i in range(0, len(data), batchsize):
        batch = data[i:i+batchsize].values.tolist()
        cursor.executemany(insert_query, batch)
        self.cnxn.commit()
</code>

The manual helper simplifies execution and optional result retrieval:

<code>def manual(self, query, response=False):
    cursor = self.cnxn.cursor()
    if response:
        return read_sql(query, self.cnxn)
    try:
        cursor.execute(query)
    except pyodbc.ProgrammingError as error:
        print("Warning:\n{}".format(error))
    self.cnxn.commit()
    return "Query complete."
</code>

Using union to combine tables:

<code>def union(self, table_list, name="union", join="UNION"):
    query = "SELECT * INTO [" + name + "] FROM (\n"
    query += f'\n{join}\n'.join([f'SELECT [{x}].* FROM [{x}]' for x in table_list])
    query += ") x"
    self.manual(query, fast=True)
</code>

And a drop method to clean up tables:

<code>def drop(self, tables):
    if isinstance(tables, str):
        tables = [tables]
    for table in tables:
        query = "IF OBJECT_ID ('[" + table + "]', 'U') IS NOT NULL DROP TABLE [" + table + "]"
        self.manual(query)
</code>

Finally, a workflow is shown that imports CSV files from a directory, pushes each DataFrame to SQL, unions them into a single table, drops the intermediate tables, extracts distinct categories, and creates separate tables per category using the manual method.

The author concludes that combining these simple functions enables powerful, automated data pipelines, and hopes the guide helps readers integrate Python with their SQL workflows.