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()