Python Database Programming Exercise:
***********************************************
This exercise covers the following three popular database servers. You can choose the database server
you are familiar with to solve this exercise.
MySQL
PostgreSQL
SQLite
You can use any driver (DB module) as per your wish, for example, there are more than 5 libraries are
available to communicate with MySQL. In this exercise, I am using the following libraries.
MySQL: mysql connector python
PostgreSQL: psycopg2
SQLite: sqlite3
This Python database programming exercise includes: –
Now it has 5 exercise questions, which simulate the real-time queries, and each question
contains a specific skill you need to learn. When you complete the exercise, you get more
familiar with database operations in Python.
Note:
The solution is provided at the end of each question. There are also tips and helpful learning
resources for each question, which will help you solve the exercise.
Exercise/mini Project: Hospital Information System
In this exercise, We are implementing the Hospital Information System. In this exercise, I have
created two tables, Hospital and Doctor. You need to create those two tables on your database
server before starting the exercise.
SQL Queries for data preparation
Please find below the SQL queries to prepare the required data for our exercise.
PostgreSQL:
CREATE database python_db;
CREATE TABLE Hospital (
Hospital_Id serial NOT NULL PRIMARY KEY,
Hospital_Name VARCHAR (100) NOT NULL,
Bed_Count serial
);
INSERT INTO Hospital (Hospital_Id, Hospital_Name, Bed_Count)
VALUES
('1', 'Mayo Clinic', 200),
('2', 'Cleveland Clinic', 400),
('3', 'Johns Hopkins', 1000),
('4', 'UCLA Medical Center', 1500);
CREATE TABLE Doctor (
Doctor_Id serial NOT NULL PRIMARY KEY,
Doctor_Name VARCHAR (100) NOT NULL,
Hospital_Id serial NOT NULL,
Joining_Date DATE NOT NULL,
Speciality VARCHAR (100) NOT NULL,
Salary INTEGER NOT NULL,
Experience SMALLINT
);
INSERT INTO Doctor (Doctor_Id, Doctor_Name, Hospital_Id,
Joining_Date, Speciality, Salary, Experience)
VALUES
('101', 'David', '1', '2005-2-10', 'Pediatric', '40000', NULL),
('102', 'Michael', '1', '2018-07-23', 'Oncologist', '20000',
NULL),
('103', 'Susan', '2', '2016-05-19', 'Garnacologist', '25000',
NULL),
('104', 'Robert', '2', '2017-12-28', 'Pediatric ', '28000', NULL),
('105', 'Linda', '3', '2004-06-04', 'Garnacologist', '42000',
NULL),
('106', 'William', '3', '2012-09-11', 'Dermatologist', '30000',
NULL),
('107', 'Richard', '4', '2014-08-21', 'Garnacologist', '32000',
NULL),
('108', 'Karen', '4', '2011-10-17', 'Radiologist', '30000', NULL);
These tables should look like this.
hospital table
Doctor table
SQL data model that we are using for this exercise
Exercise 1: Connect to your database server and print its version
#import psycopg2
import [Link]
def get_connection():
connection = [Link](user="postgres",
password="pynative@#29",
host="[Link]",
port="5432",
database="python_db")
return connection
def close_connection(connection):
if connection:
[Link]()
def read_database_version():
try:
connection = get_connection()
cursor = [Link]()
[Link]("SELECT version();")
db_version = [Link]()
print("You are connected to PostgreSQL version: ", db_version)
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting data", error)
print("Question 1: Print Database version")
read_database_version()
Question 2: Fetch Hospital and Doctor Information using hospital Id and doctor Id
#import psycopg2
import [Link]
def get_connection():
connection = [Link](user="postgres",
password="pynative@#29",
host="[Link]",
port="5432",
database="python_db")
return connection
def close_connection(connection):
if connection:
[Link]()
print("Postgres connection is closed")
def get_hospital_detail(hospital_id):
try:
connection = get_connection()
cursor = [Link]()
select_query = """select * from Hospital where Hospital_Id = %s"""
[Link](select_query, (hospital_id,))
records = [Link]()
print("Printing Hospital record")
for row in records:
print("Hospital Id:", row[0], )
print("Hospital Name:", row[1])
print("Bed Count:", row[2])
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting data", error)
def get_doctor_detail(doctor_id):
try:
connection = get_connection()
cursor = [Link]()
select_query = """select * from Doctor where Doctor_Id = %s"""
[Link](select_query, (doctor_id,))
records = [Link]()
print("Printing Doctor record")
for row in records:
print("Doctor Id:", row[0])
print("Doctor Name:", row[1])
print("Hospital Id:", row[2])
print("Joining Date:", row[3])
print("Specialty:", row[4])
print("Salary:", row[5])
print("Experience:", row[6])
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting data", error)
print("Question 2: Read given hospital and doctor details \n")
get_hospital_detail(2)
print("\n")
get_doctor_detail(105)
Exercise 3: Get the list of doctors as per the given specialty and salary
import psycopg2
def get_connection():
connection = [Link](user="postgres",
password="pynative@#29",
host="[Link]",
port="5432",
database="python_db")
return connection
def close_connection(connection):
if connection:
[Link]()
print("Postgres connection is closed")
def get_specialist_doctors_list(speciality, salary):
try:
connection = get_connection()
cursor = [Link]()
sql_select_query = """select * from Doctor where Speciality=%s and Salary > %s"""
[Link](sql_select_query, (speciality, salary))
records = [Link]()
print("Printing doctors whose specialty is", speciality, "and salary greater than", salary, "\n")
for row in records:
print("Doctor Id: ", row[0])
print("Doctor Name:", row[1])
print("Hospital Id:", row[2])
print("Joining Date:", row[3])
print("Specialty:", row[4])
print("Salary:", row[5])
print("Experience:", row[6], "\n")
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting data", error)
print("Question 3: Get Doctors as per given Speciality\n")
get_specialist_doctors_list("Garnacologist", 30000)
Exercise 4: Get a list of doctors from a given hospital
Note: Implement the functionality to fetch all the doctors as per the given Hospital Id. You
must display the hospital name of a doctor.
import psycopg2
def get_connection():
connection = [Link](user="postgres",
password="pynative@#29",
host="[Link]",
port="5432",
database="python_db")
return connection
def close_connection(connection):
if connection:
[Link]()
def get_hospital_name(hospital_id):
# Fetch Hospital Name using Hospital id
try:
connection = get_connection()
cursor = [Link]()
select_query = """select * from Hospital where Hospital_Id = %s"""
[Link](select_query, (hospital_id,))
record = [Link]()
close_connection(connection)
return record[1]
except (Exception, [Link]) as error:
print("Error while getting data from PostgreSQL", error)
def get_doctors(hospital_id):
# Fetch Hospital Name using Hospital id
try:
hospital_name = get_hospital_name(hospital_id)
connection = get_connection()
cursor = [Link]()
sql_select_query = """select * from Doctor where Hospital_Id = %s"""
[Link](sql_select_query, (hospital_id,))
records = [Link]()
print("Printing Doctors of ", hospital_name, "Hospital")
for row in records:
print("Doctor Id:", row[0])
print("Doctor Name:", row[1])
print("Hospital Id:", row[2])
print("Hospital Name:", hospital_name)
print("Joining Date:", row[3])
print("Specialty:", row[4])
print("Salary:", row[5])
print("Experience:", row[6], "\n")
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting doctor's data", error)
print("Question 4: Get List of doctors of a given Hospital Id\n")
get_doctors(2)
Operation 5: Update doctor experience in years
import psycopg2
import datetime
from [Link] import relativedelta
def get_connection():
connection = [Link](user="postgres",password="pynative@#29",
host="[Link]", port="5432",database="python_db")
return connection
def close_connection(connection):
if connection:
[Link]()
def update_doctor_experience(doctor_id):
# Update Doctor Experience in Years
try:
# Get joining date
connection = get_connection()
cursor = [Link]()
select_query = """select Joining_Date from Doctor where Doctor_Id = %s"""
[Link](select_query, (doctor_id,))
joining_date = [Link]()
# calculate Experience in years
joining_date_1 = [Link](''.join(map(str, joining_date)), '%Y-%m-%d')
today_date = [Link]()
experience = relativedelta(today_date, joining_date_1).years
# Update doctor's Experience now
connection = get_connection()
cursor = [Link]()
sql_select_query = """update Doctor set Experience = %s where Doctor_Id =%s"""
[Link](sql_select_query, (experience, doctor_id))
[Link]()
print("Doctor Id:", doctor_id, " Experience updated to ", experience, " years")
close_connection(connection)
except (Exception, [Link]) as error:
print("Error while getting doctor's data", error)
print("Question 5: Calculate and Update experience of all doctors \n")
update_doctor_experience(101)