Pyodbc - reusable cursor function

Python Help
1

Hello, forgive me I’m new to programming and Python.

I am trying to write a function that implements pyodbc to execute any stored procedures in SQL Server that I pass as an argument. Also, these stored procedures can have different numbers of parameters. The following code works fine but I was wondering if there is a way of achieving the same aim more succintly.

Thanks in advance.

import pyodbc
import os

class SqlDb:

    cnx = os.environ["BFDBCNX"]

    def Exec(self, proc, param1 = None, param2 = None, param3 = None, param4 = None, param5 = None):

        try:
            connection = pyodbc.connect(self.cnx)
            cursor = connection.cursor()

            if (param1 is None):
                cursor.execute("{CALL " + proc + "}")
                dataset = cursor.fetchall()
                
                return dataset

            elif(param2 is None):
                cursor.execute("{CALL " + proc + "(?)}", param1)
                connection.commit()

            elif(param3 is None):
                cursor.execute("{CALL " + proc + "(?, ?)}", param1, param2)
                connection.commit()

            elif(param4 is None):
                cursor.execute("{CALL " + proc + "(?, ?, ?)}", param1, param2, param3)
                connection.commit()

            elif(param5 is None):
                cursor.execute("{CALL " + proc + "(?, ?, ?, ?)}", param1, param2, param3, param4)
                connection.commit()                                    

            else:
                cursor.execute("{CALL " + proc + "(?, ?, ?, ?, ?)}", param1, param2, param3, param4, param5)
                connection.commit()              

        except pyodbc.Error as ex:
            print("An error occurred in SQL Server:", ex)
            connection.rollback()

        finally:
            if 'connection' in locals():
                connection.close()
2

Hello,

If you just want to write a function, is there a reason that you’re wrapping it in a class? Is this a requirement?

I notice that this line is written five different times:

connection.commit()

If you slightly modify the script a bit, by separating the subsequent elif and last else conditional statements under one else, then instead of including this statement five different times in your script, you can include it only once (it will execute after all of them have been tested).

class SqlDb:
    cnx = os.environ["BFDBCNX"]

    def Exec(self, proc, param1=None, param2=None, param3=None, param4=None, param5=None):

        try:
            connection = pyodbc.connect(self.cnx)
            cursor = connection.cursor()

            if (param1 is None):
                cursor.execute("{CALL " + proc + "}")
                dataset = cursor.fetchall()

                return dataset

            else:

                if (param2 is None):
                    cursor.execute("{CALL " + proc + "(?)}", param1)

                elif (param3 is None):
                    cursor.execute("{CALL " + proc + "(?, ?)}", param1, param2)

                elif (param4 is None):
                    cursor.execute("{CALL " + proc + "(?, ?, ?)}", param1, param2, param3)

                elif (param5 is None):
                    cursor.execute("{CALL " + proc + "(?, ?, ?, ?)}", param1, param2, param3, param4)

                else:
                    cursor.execute("{CALL " + proc + "(?, ?, ?, ?, ?)}", param1, param2, param3, param4, param5)

                connection.commit()  # Only need to write once

        except pyodbc.Error as ex:
            print("An error occurred in SQL Server:", ex)
            connection.rollback()

        finally:
            if 'connection' in locals():
                connection.close()

Note that as written, only one elif conditional statement is executed even if all are true. If you want to check if every parameter is true, then you might want to think about employing if conditional statements for every parameter.

3

A function can be defined to accept multiple arguments and collect them into a tuple:

>>> def my_func(*args):
...     print(f'args is {args}')
...
>>> my_func()
args is ()
>>> my_func(1)
args is (1,)
>>> my_func(1, 2)
args is (1, 2)

You can then make the placeholders using the number of parameters:

if params:
    cursor.execute("{CALL " + proc + "(" + ", ".join("?" * len(params)) + ")}")
else:
    cursor.execute("{CALL " + proc + "}")

Also, it’s probably better to make the connection outside the try block:

connection = pyodbc.connect(self.cnx)

try:
    ...
except pyodbc.Error as ex:
    print("An error occurred in SQL Server:", ex)
    connection.rollback()
finally:
    connection.close()

Putting these together gives:

import pyodbc
import os

class SqlDb:
    cnx = os.environ["BFDBCNX"]

    def Exec(self, proc, *params):
        connection = pyodbc.connect(self.cnx)

        try:
            cursor = connection.cursor()

            if params:
                cursor.execute("{CALL " + proc + "(" + ", ".join("?" * len(params)) + ")}", *params)
            else:
                cursor.execute("{CALL " + proc + "}")

            dataset = cursor.fetchall()
            connection.commit()
            return dataset
        except pyodbc.Error as ex:
            print("An error occurred in SQL Server:", ex)
            connection.rollback()
        finally:
            connection.close()
4

Hi Paul,

This is just one section of a much larger program, there are other functions in this class which I haven’t shown plus other classes in fact. I wanted to use classes to keep my Python scripts smaller and modularised, but I am new to this so maybe I shouldn’t be doing this.

Thanks for pointing out that I’m reusing the commit five times. Thanks for pointing that out and I can see now that I didn’t need to do this.

5

Thanks so much for this Matthew. This is exactly what I had in mind but didn’t know how to do.