FULL PYTHON + MS SQL SCRIPT (BEST PRACTICE)

 

FULL PYTHON + MS SQL SCRIPT (BEST PRACTICE)

import pyodbc

# -----------------------------------------

# SQL CONNECTION (EDIT THIS PART)

# -----------------------------------------

def get_connection():

    try:

        conn = pyodbc.connect(

            "DRIVER={ODBC Driver 17 for SQL Server};"

            "SERVER=YOUR_SERVER_NAME;"      # e.g. DESKTOP-123\SQLEXPRESS

            "DATABASE=YOUR_DATABASE_NAME;"

            "UID=YOUR_USERNAME;"

            "PWD=YOUR_PASSWORD;",

            autocommit=False                # Manual commit = safer

        )

        return conn

    except Exception as ex:

        print("Database connection failed:", ex)

        return None

# -----------------------------------------

# INSERT RECORD

# -----------------------------------------

def insert_employee(emp_id, name, salary):

    conn = get_connection()

    if not conn:

        return

    try:

        cursor = conn.cursor()

        query = """

            INSERT INTO Employees (EmpID, Name, Salary)

            VALUES (?, ?, ?)

        """

        cursor.execute(query, (emp_id, name, salary))

        conn.commit()

        print("Record inserted successfully!")

    except Exception as ex:

        conn.rollback()

        print("Insert failed:", ex)

    finally:

        conn.close()

# -----------------------------------------

# UPDATE RECORD

# -----------------------------------------

def update_employee(emp_id, new_salary):

    conn = get_connection()

    if not conn:

        return

    try:

        cursor = conn.cursor()

        query = """

            UPDATE Employees

            SET Salary = ?

            WHERE EmpID = ?

        """

        cursor.execute(query, (new_salary, emp_id))

        conn.commit()

        print("Record updated successfully!")

    except Exception as ex:

        conn.rollback()

        print("Update failed:", ex)

    finally:

        conn.close()

# -----------------------------------------

# DELETE RECORD

# -----------------------------------------

def delete_employee(emp_id):

    conn = get_connection()

    if not conn:

        return

    try:

        cursor = conn.cursor()

        query = "DELETE FROM Employees WHERE EmpID = ?"

        cursor.execute(query, (emp_id,))

        conn.commit()

        print("Record deleted successfully!")

    except Exception as ex:

        conn.rollback()

        print("Delete failed:", ex)

    finally:

        conn.close()

# -----------------------------------------

# SELECT ALL RECORDS

# -----------------------------------------

def get_all_employees():

    conn = get_connection()

    if not conn:

        return

    try:

        cursor = conn.cursor()

        cursor.execute("SELECT EmpID, Name, Salary FROM Employees")

        print("\n--- Employee List ---")

        for row in cursor.fetchall():

            print(row.EmpID, row.Name, row.Salary)

    except Exception as ex:

        print("Select failed:", ex)

    finally:

        conn.close()

# -----------------------------------------

# CALL STORED PROCEDURE

# -----------------------------------------

def call_stored_procedure():

    """

    Example Stored Procedure:

    CREATE PROCEDURE GetEmployees

    AS

    BEGIN

        SELECT * FROM Employees

    END

    """

    conn = get_connection()

    if not conn:

        return

    try:

        cursor = conn.cursor()

        cursor.execute("{CALL GetEmployees}")   # or EXEC GetEmployees

        rows = cursor.fetchall()

        print("\n--- Stored Procedure Result ---")

        for row in rows:

            print(row.EmpID, row.Name, row.Salary)

    except Exception as ex:

        print("Stored procedure failed:", ex)

    finally:

        conn.close()

# -----------------------------------------

# MAIN TESTING

# -----------------------------------------

if __name__ == "__main__":

    insert_employee(1, "Vijay", 50000)

    update_employee(1, 60000)

    get_all_employees()

    delete_employee(1)

    get_all_employees()

    call_stored_procedure()

🚀 How to Use

1. Update:

SERVER=
DATABASE=
UID=
PWD=

2. Create sample table in SQL:

CREATE TABLE Employees (
    EmpID INT PRIMARY KEY,
    Name VARCHAR(100),
    Salary DECIMAL(18,2)
);

3. Run the Python file:

python mssql_python_demo.py