In this post, we will see what PostgreSQL is and how to use it in Python.
PostgreSQL, often referred to as Postgres, is a powerful, open-source object-relational database system with a strong reputation for reliability, feature robustness, and performance. It has been in active development for over 30 years, making it a highly mature database choice. PostgreSQL is known for its standards compliance, extensibility, and its ability to handle a wide variety of workloads.
The key characteristics of PostgreSQL are:
- Open Source and Free: PostgreSQL is open source and released under the PostgreSQL License, a liberal Open Source license, which means we can use, modify, and distribute PostgreSQL in any form we wish.
- Standards-Compliant: PostgreSQL adheres closely to the SQL:2008 standard, which includes advanced features like complex queries, foreign keys, triggers, views, transactional integrity, and multiversion concurrency control (MVCC).
- Extensibility: One of PostgreSQL’s most notable features is its extensibility. It allows users to define their own data types, operators, and index types, making it a very flexible and powerful database system.
- Advanced Data Types: PostgreSQL supports a variety of advanced data types including arrays, hstore (key-value pairs), JSON, XML, and even custom data types created by users.
- Robust Performance: PostgreSQL includes features like indexing, sophisticated query optimization, and parallelization that ensure high performance for both read and write operations.
One of the most important rival of PostgreSQL is obviously is Microsoft SQL Server but, there are many differences between the two:
- Licensing and Cost:
PostgreSQL: Open-source and free to use.
MS SQL Server: Proprietary software with various paid editions, though a free Express edition is available with limited features. - Operating System Compatibility:
PostgreSQL: Cross-platform, runs on various operating systems including Windows, Linux, and macOS.
MS SQL Server: Primarily Windows-based, though Linux versions are available. - Extensibility:
PostgreSQL: Highly extensible with support for custom data types, operators, and functions.
MS SQL Server: Less extensible compared to PostgreSQL, though it supports CLR integration for custom functions. - Language Support:
PostgreSQL: Supports a variety of procedural languages like PL/pgSQL, PL/Python, PL/Perl, and PL/Tcl.
MS SQL Server: Primarily uses T-SQL (Transact-SQL), with support for CLR integration. - Performance and Scalability:
PostgreSQL: Known for its excellent concurrency and performance, especially in read-heavy and mixed-load environments.
MS SQL Server: Offers strong performance and scalability features, particularly in write-heavy environments and enterprise scenarios.
For all information, the official web site is: https://www.postgresql.org
To quickly get started with PostgreSQL, we can run it in a Docker container. Below is the command to run a PostgreSQL container with the user and password equal to: Admin/Pass123:
docker run --name db_postgres -e POSTGRES_USER=Admin -e POSTGRES_PASSWORD=Pass123 -p 5432:5432 -d postgres
Finally, with a software like DBeaver, we can connect and manage our PostgreSQL instance for example, creating a database called TestDB:
Now that our database has been created, we will see how to implement CRUD operations in Python, for an entity called User that we are going to create in the TestDB database.
First of all, in DBeaver, we run this script to create the table User:
CREATE TABLE "User" (
"Id" SERIAL PRIMARY KEY,
"Username" VARCHAR(50) NOT NULL UNIQUE,
"Password" VARCHAR(100) NOT NULL,
"Email" VARCHAR(100) NOT NULL UNIQUE
);
Then, we will define our Python program.
First of all, we have to install the library ‘psycopg2’, that is a PostgreSQL database adapter.
python3 -m pip install psycopg2-binary
Then, we define a file config.ini where we will put all database parameters:
[CONFIG.INI]
[postgresql]
server = localhost
user = Admin
password = Pass123
database = TestDB
port = 5432
Finally, we will define a file called repository.py where we will insert all method to manage the CRUD operations.
We start defining the methods to open and close the cursor and the connection:
[REPOSITORY.PY]
import configparser
import psycopg2
from psycopg2 import sql
class Repository:
_cursor = None
_connection = None
# Database connection parameters
_db_name = None
_db_user = None
_db_password = None
_db_server = None
_db_port = None
def __init__(self) -> None:
# definition of the object used to read the config file
configfile = configparser.ConfigParser()
configfile.read("config.ini")
# get all db parameters
postgresql = configfile["postgresql"]
self._db_name = postgresql["database"]
self._db_user = postgresql["user"]
self._db_password = postgresql["password"]
self._db_server = postgresql["server"]
self._db_port = postgresql["port"]
def _open_cursor(self):
self._connection = psycopg2.connect(
dbname=self._db_name,
user=self._db_user,
password=self._db_password,
host=self._db_server,
port=self._db_port
)
self._cursor = self._connection.cursor()
def _close_cursor(self):
self._cursor.close()
self._connection.close()
And now, step by step, we will add the methods:
INSERT & SELECT ALL
def add_user(self, username, password, email):
try:
# Open a cursor to perform database operations
self._open_cursor()
# Execute the SQL INSERT statement to add a new user to the Users table
# The statement uses parameterized queries to prevent SQL injection attacks
# The RETURNING clause fetches the 'Id' of the newly inserted user
self._cursor.execute(
"INSERT INTO public.\"User\" (\"Username\", \"Password\", \"Email\") VALUES (%s, %s, %s) RETURNING \"Id\"",
(username, password, email)
)
# Commit the transaction to save the changes in the database
self._connection.commit()
# Fetch the 'Id' of the newly inserted user from the result of the RETURNING clause
user_id = self._cursor.fetchone()[0]
# Return the 'Id' of the newly inserted user
return user_id
except Exception as e:
# Print an error message if any exception occurs during the database operation
print(f"An error occurred: {e}")
finally:
# Ensure that the cursor and connection are properly closed
self._close_cursor()
def get_all(self):
self._open_cursor()
self._cursor.execute("SELECT * FROM public.\"User\"")
lst_users = self._cursor.fetchall()
self._close_cursor()
if lst_users:
return lst_users
else:
print("No users found.")
return []
[MAIN.PY]
from repository import Repository
objRepo = Repository()
lst_users = objRepo.get_all()
print("List of Users")
for item in lst_users:
print(f"Id:{item[0]}, Username:{item[1]}, Password:{item[2]}, Email:{item[3]}")
print(f"A new user has been inserted and with th Id:{objRepo.add_user("User1", "Pass1", "user1@test.com")}")
print(f"A new user has been inserted and with th Id:{objRepo.add_user("User2", "Pass2", "user2@test.com")}")
print(f"A new user has been inserted and with th Id:{objRepo.add_user("User3", "Pass3", "user3@test.com")}")
print(f"A new user has been inserted and with th Id:{objRepo.add_user("User4", "Pass4", "user4@test.com")}")
print(f"A new user has been inserted and with th Id:{objRepo.add_user("User5", "Pass5", "user5@test.com")}")
print()
print("List of Users")
lst_users = objRepo.get_all()
for item in lst_users:
print(f"Id:{item[0]}, Username:{item[1]}, Password:{item[2]}, Email:{item[3]}")
SELECT BY ID
def get_user_by_id(self, id):
self._open_cursor()
self._cursor.execute("SELECT * FROM public.\"User\" WHERE \"Id\" = %s", (id,))
user = self._cursor.fetchone()
self._close_cursor()
if user:
return user
else:
print(f"No User found with Id:{id}")
return None
[MAIN.PY]
from repository import Repository
objRepo = Repository()
print("Try to select User with Id=7")
user = objRepo.get_user_by_id(7)
if user!=None:
print(f"The selected User is -> Id:{user[0]}, Username:{user[1]}, Password:{user[2]}, Email:{user[3]}")
print()
print("Try to select User with Id=3")
user = objRepo.get_user_by_id(3)
if user != None:
print(f"The selected User is -> Id:{user[0]}, Username:{user[1]}, Password:{user[2]}, Email:{user[3]}")
UPDATE
def update_user_email(self, id, new_email):
try:
self._open_cursor()
self._cursor.execute("UPDATE public.\"User\" set \"Email\" = %s where \"Id\" = %s", (new_email, id))
self._connection.commit()
except Exception as e:
print(f"An error occurred: {e}")
finally:
self._close_cursor()
[MAIN.PY]
from repository import Repository
objRepo = Repository()
print("Try to select User with Id=7")
user = objRepo.get_user_by_id(7)
if user!=None:
print(f"The selected User is -> Id:{user[0]}, Username:{user[1]}, Password:{user[2]}, Email:{user[3]}")
print()
print("Update email")
objRepo.update_user_email(7, "newUser7@test.com")
print("Try to select User with Id=7")
user = objRepo.get_user_by_id(7)
if user!=None:
print(f"The selected User is -> Id:{user[0]}, Username:{user[1]}, Password:{user[2]}, Email:{user[3]}")
DELETE
def delete_user(self, id):
try:
self._open_cursor()
self._cursor.execute("DELETE FROM public.\"User\" WHERE \"Id\" = %s", (id,))
self._connection.commit()
except Exception as e:
print(f"An error occurred: {e}")
finally:
self._close_cursor()
[MAIN.PY]
from repository import Repository
objRepo = Repository()
print("List of Users")
lst_users = objRepo.get_all()
for item in lst_users:
print(f"Id:{item[0]}, Username:{item[1]}, Password:{item[2]}, Email:{item[3]}")
print()
print("Delete User with Id:7")
objRepo.delete_user(7)
print()
print("List of Users")
lst_users = objRepo.get_all()
for item in lst_users:
print(f"Id:{item[0]}, Username:{item[1]}, Password:{item[2]}, Email:{item[3]}")
[REPOSITORY.PY]
import configparser
import psycopg2
from psycopg2 import sql
class Repository:
_cursor = None
_connection = None
# Database connection parameters
_db_name = None
_db_user = None
_db_password = None
_db_server = None
_db_port = None
def __init__(self) -> None:
# definition of the object used to read the config file
configfile = configparser.ConfigParser()
configfile.read("config.ini")
# get all db parameters
postgresql = configfile["postgresql"]
self._db_name = postgresql["database"]
self._db_user = postgresql["user"]
self._db_password = postgresql["password"]
self._db_server = postgresql["server"]
self._db_port = postgresql["port"]
def _open_cursor(self):
self._connection = psycopg2.connect(
dbname=self._db_name,
user=self._db_user,
password=self._db_password,
host=self._db_server,
port=self._db_port
)
self._cursor = self._connection.cursor()
def _close_cursor(self):
self._cursor.close()
self._connection.close()
def add_user(self, username,password, email):
try:
self._open_cursor()
self._cursor.execute("INSERT INTO public.\"User\" (\"Username\",\"Password\", \"Email\") VALUES (%s, %s, %s) RETURNING \"Id\"", (username, password, email))
self._connection.commit()
user_id = self._cursor.fetchone()[0]
return user_id
except Exception as e:
print(f"An error occurred: {e}")
finally:
self._close_cursor()
def get_all(self):
self._open_cursor()
self._cursor.execute("SELECT * FROM public.\"User\"")
lst_users = self._cursor.fetchall()
self._close_cursor()
if lst_users:
return lst_users
else:
print("No users found.")
return []
def get_user_by_id(self, id):
self._open_cursor()
self._cursor.execute("SELECT * FROM public.\"User\" WHERE \"Id\" = %s", (id,))
user = self._cursor.fetchone()
self._close_cursor()
if user:
return user
else:
print(f"No User found with Id:{id}")
return None
def delete_user(self, id):
try:
self._open_cursor()
self._cursor.execute("DELETE FROM public.\"User\" WHERE \"Id\" = %s", (id,))
self._connection.commit()
except Exception as e:
print(f"An error occurred: {e}")
finally:
self._close_cursor()
def update_user_email(self, id, new_email):
try:
self._open_cursor()
self._cursor.execute("UPDATE public.\"User\" set \"Email\" = %s where \"Id\" = %s", (new_email, id))
self._connection.commit()
except Exception as e:
print(f"An error occurred: {e}")
finally:
self._close_cursor()
The last thing I want to higlight again, is that this post was created to show how to handle CRUD operations in PostgreSQL with Python and, it is not a post about PostgreSQL.
This database is very powerful and there would be a lot of things to talk about.
Maybe, in the future, in another post…