Python – Management of a SQLServer database

By | 15/11/2023

In this post, we will see how Python can seamlessly integrate with SQL Server to perform various tasks, from connecting to the database and executing SQL queries to performing advanced data analysis and visualization.
In the past , we have already seen how to connect to SQL Server with python but, it was only an introduction

First of all, we have to install the library ‘pyodbc’, used to connect to the SQL Server database, with the command:

pip install pyodbc


Then, we run a sql script to create a database called ‘TestPython’ with a table called ‘TabUser’:

USE [TestPython]
GO
/****** Object:  Table [dbo].[TabUser]    Script Date: 29/10/2023 23:36:17 ******/
SET ANSI_NULLS ON
GO
SET QUOTED_IDENTIFIER ON
GO
CREATE TABLE [dbo].[TabUser](
	[Id] [int] IDENTITY(1,1) NOT NULL,
	[Username] [nvarchar](50) NOT NULL,
	[Password] [nvarchar](50) NOT NULL,
 CONSTRAINT [PK_TabUser] PRIMARY KEY CLUSTERED 
(
	[Id] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, IGNORE_DUP_KEY = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON, OPTIMIZE_FOR_SEQUENTIAL_KEY = OFF) ON [PRIMARY]
) ON [PRIMARY]
GO
SET IDENTITY_INSERT [dbo].[TabUser] ON 
GO
INSERT [dbo].[TabUser] ([Id], [Username], [Password]) VALUES (5, N'User1', N'pass123')
GO
INSERT [dbo].[TabUser] ([Id], [Username], [Password]) VALUES (6, N'User2', N'pass567')
GO
INSERT [dbo].[TabUser] ([Id], [Username], [Password]) VALUES (7, N'User3', N'pass987')
GO
INSERT [dbo].[TabUser] ([Id], [Username], [Password]) VALUES (9, N'User4', N'pass34123')
GO
SET IDENTITY_INSERT [dbo].[TabUser] OFF
GO


Before we start writing code, we create a config file so defined:
[CONFIG.INI]

[sqlserver]
server = 127.0.0.1
user = usertest
password = pas123
database = TestPython
driver = {ODBC Driver 17 for SQL Server}


Finally, we define a class called user and a class called repository:

[USER.PY]

class User:
    def __init__(self, id, username, password):
        self.Id = id
        self.UserName = username
        self.Password = password


[REPOSITORY.PY]

import configparser
import pyodbc
from user import User

class Repository:
    connectionstring = None
    cursor = None
    connection = None

    def _getsqlconnection(self):
        # definition of the object used to read the config file
        configfile = configparser.ConfigParser()
        configfile.read("config.ini")

        sqlserver = configfile["sqlserver"]
        server = sqlserver["server"]
        user = sqlserver["user"]
        password = sqlserver["password"]
        database = sqlserver["database"]
        driver = sqlserver["driver"]

        return f'DRIVER={driver};SERVER={server};DATABASE={database};UID={user};PWD={password}'
    
    def __init__(self):
        self.connectionstring = self._getsqlconnection()

    def _opencursor(self):
        # open connection with database
        self.connection = pyodbc.connect(self.connectionstring)
        # open the cursor
        self.cursor = self.connection.cursor()

    def _closecursor(self):
        # close the cursor and the database connection
        self.cursor.close()
        self.connection.close()


Now, in repository.py, we will add all methods to manage the database:

SELECT ALL USERS:

    def getallusers(self):
        try:
            # open cursor
            self._opencursor()

            # define the SQL SELECT statement
            select_query = "select Id, Username, Password from TabUser"

            # execute the Select and Fetch all records
            self.cursor.execute(select_query)

            # fetch all records
            result = self.cursor.fetchall()

            # create a list of user object
            lstuser = []

            for item in result:
                useritem = User(item[0], item[1], item[2])
                lstuser.append(useritem)

            return lstuser

        except Exception as error:
            print("Something went wrong in the method: getallusers")
            print(error)
        finally:
            self._closecursor()


[MAIN.PY]

from repository import Repository

repo = Repository()

lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


If we run main.py, the following will be the result:


INSERT USER:

    def addnewuser(self, username, password):
        try:
            # open cursor
            self._opencursor()

            # define the SQL INSERT statement
            insert_query = "INSERT INTO TabUser (Username, Password) VALUES (?,?)"

            # Define the values we want to insert
            values = (username, password)

            # execute the INSERT statement
            self.cursor.execute(insert_query, values)

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: addnewuser")
            print(error)
        finally:
            self._closecursor()


[MAIN.PY]

from repository import Repository

repo = Repository()
lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


repo.addnewuser('User5', 'pass765')
print("\n")
print('A new user has been added \n')


lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


If we run main.py, the following will be the result:


UPDATE USER:

    def updateuser(self, id, username, password):
        try:
            # open cursor
            self._opencursor()

            # define the SQL UPDATE statement
            update_query = "UPDATE TabUser set Username =?, Password=? where Id = ?"

            # execute the UPDATE statement
            self.cursor.execute(update_query, (username, password, id))

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: updateuser")
            print(error)
        finally:
            self._closecursor()


[MAIN.PY]

from repository import Repository

repo = Repository()
lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


repo.updateuser(13,'User51', 'pass1111')
print("\n")
print('A user with id=13 has been modified \n')


lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


If we run main.py, the following will be the result:


DELETE USER:

    def deleteuser(self, id):
        try:
            # open cursor
            self._opencursor()

            # define the SQL DELETE statement
            delete_query = "DELETE TabUser WHERE ID = ?"

            # execute the DELETE statement
            self.cursor.execute(delete_query, (id,))

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: deleteuser")
            print(error)
        finally:
            self._closecursor()


[MAIN.PY]

from repository import Repository

repo = Repository()
lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


repo.deleteuser(13)
print("\n")
print('The user with id=13 has been deleted \n')


lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


If we run main.py, the following will be the result:


HOW TO MANAGE A TRANSACTION:

    def managetransaction(self, id):
        try:
            print('Try to delete an item')
            # open cursor
            self._opencursor()

            # Start a transaction
            self.connection.autocommit = False  # Disable autocommit mode

            # define the SQL DELETE statement
            delete_query = "DELETE TabUser WHERE ID = ?"

            # execute the DELETE statement
            self.cursor.execute(delete_query, (id,))

            # Commit the transaction
            # In this case we have called the rollback but, in a real project, we should have called the commit
            # self.connection.commit()
            self.connection.rollback()
        except Exception as error:
            self.connection.rollback()
            print("Something went wrong in the method: managetransaction")
            print(error)
        finally:
            self._closecursor()


[MAIN.PY]

from repository import Repository

repo = Repository()
lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")

print("\n")
print('We try to delete the user with Id=6 \n')
repo.managetransaction(6)

lstusers = repo.getallusers()

for item in lstusers:
    print(f"{item.Id} - {item.UserName} - {item.Password}")


If we run main.py, the following will be the result:


[REPOSITORY.PY]

import configparser
import pyodbc
from user import User


class Repository:
    connectionstring = None
    cursor = None
    connection = None

    def _getsqlconnection(self):
        # definition of the object used to read the file
        configfile = configparser.ConfigParser()
        configfile.read("config.ini")

        sqlserver = configfile["sqlserver"]
        server = sqlserver["server"]
        user = sqlserver["user"]
        password = sqlserver["password"]
        database = sqlserver["database"]
        driver = sqlserver["driver"]

        return f'DRIVER={driver};SERVER={server};DATABASE={database};UID={user};PWD={password}'

    def __init__(self):
        self.connectionstring = self._getsqlconnection()

    def _opencursor(self):
        # open connection with database
        self.connection = pyodbc.connect(self.connectionstring)
        # open the cursor
        self.cursor = self.connection.cursor()

    def _closecursor(self):
        self.cursor.close()
        self.connection.close()

    def getallusers(self):
        try:
            # open cursor
            self._opencursor()

            # define the SQL SELECT statement
            select_query = "select Id, Username, Password from TabUser"

            # execute the Select and Fetch all records
            self.cursor.execute(select_query)

            # fetch all records
            result = self.cursor.fetchall()

            # create a list of user object
            lstuser = []

            for item in result:
                useritem = User(item[0], item[1], item[2])
                lstuser.append(useritem)

            return lstuser

        except Exception as error:
            print("Something went wrong in the method: getallusers")
            print(error)
        finally:
            self._closecursor()

    def addnewuser(self, username, password):
        try:
            # open cursor
            self._opencursor()

            # define the SQL INSERT statement
            insert_query = "INSERT INTO TabUser (Username, Password) VALUES (?,?)"

            # Define the values we want to insert
            values = (username, password)

            # execute the INSERT statement
            self.cursor.execute(insert_query, values)

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: addnewuser")
            print(error)
        finally:
            self._closecursor()

    def updateuser(self, id, username, password):
        try:
            # open cursor
            self._opencursor()

            # define the SQL UPDATE statement
            update_query = "UPDATE TabUser set Username =?, Password=? where Id = ?"

            # execute the UPDATE statement
            self.cursor.execute(update_query, (username, password, id))

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: updateuser")
            print(error)
        finally:
            self._closecursor()

    def deleteuser(self, id):
        try:
            # open cursor
            self._opencursor()

            # define the SQL DELETE statement
            delete_query = "DELETE TabUser WHERE ID = ?"

            # execute the DELETE statement
            self.cursor.execute(delete_query, (id,))

            # Commit the transaction
            self.connection.commit()

        except Exception as error:
            print("Something went wrong in the method: deleteuser")
            print(error)
        finally:
            self._closecursor()

    def managetransaction(self, id):
        try:
            print('Try to delete an item')
            # open cursor
            self._opencursor()

            # Start a transaction
            self.connection.autocommit = False  # Disable autocommit mode

            # define the SQL DELETE statement
            delete_query = "DELETE TabUser WHERE ID = ?"

            # execute the DELETE statement
            self.cursor.execute(delete_query, (id,))

            # Commit the transaction
            self.connection.commit()
        except Exception as error:
            self.connection.rollback()
            print("Something went wrong in the method: managetransaction")
            print(error)
        finally:
            self._closecursor()



Leave a Reply

Your email address will not be published. Required fields are marked *