Connect Python to PostgreSQL
Python has lot of libraries one of them is SQLAlchemy. It is a Python SQL toolkit and Object Relational Mapper (ORM) that helps you work with relational databases like PostgreSQL, MySQL, SQLite, etc. We can write queries using Python, instead of raw SQL
Two Main Layers of SQLAlchemy:
Core (low-level)
Gives you direct SQL expression language and schema management.
Great for writing raw or dynamic SQL in Python.
ORM (high-level)
Maps database tables to Python classes and handles relationships.
Useful when working with structured, object-oriented code.
Installation
There are two ways to install this python tool.
1. Execute the below command in the Jupiter NoteBook terminal
!pip install sqlalchemy
2. In the case of anaconda distribution of Python then you can install it from the conda terminal:
conda install -c anaconda sqlalchemySQLAlchemy Core
Establish Connection with Database
The first step in establishing a connection with the PostgreSQL database is creating an engine object using the create_engine() function of SQLAlchemy. This method takes in the connection URL and returns a SQLAlchemy engine that references both a Dialect and a Pool.
engine = create_engine(dialect+driver://username:password@host:port/database_name)
Replace user, password, host, port, and database with your actual PostgreSQL credentials. If the database is running on the default port (5432), the :port part can be omitted. The postgresql+psycopg2 part specifies the database dialect and driver.
Example:
engine = create_engine("postgresql+psycopg2://postgres:Data%40123@localhost:5433/postgres")
dialect = postgresql
driver = psycopg2
username = postgres(PostgreSQL username)
password= Data@12/3(PostgreSQL password)
Port = 5433( Server runs on different port not on the default port 5432)
Database name = postgres
This engine object acts as a central source of connections to a particular database, providing both a factory as well as a holding space called a connection pool for these database connections.
The engine is typically a global object created just once for a particular database server
Escaping Special Characters such as @ signs in Passwords
When constructing the URL for create_engine(), the special characters in username and password should be URL encoded to be parsed correctly.
Below is the example URL
url = “postgresql+psycopg2://postgres:Data@12/3@localhost:5433/postgres"
1)The password given is ‘Data@12/3 ‘ here the “@” sign and ‘/’ characters should be encoded as %40 and %2F, respectively:
Now the URL is:
url = "postgresql+psycopg2://postgres:Data%40123@localhost:5433/postgres"
2) One more alternative way to escape special character is create the URL as object , don’t create the url as string . We can see how to do that.
The URL object is created using the URL.create() constructor method, passing all fields individually. Special characters such as those within passwords may be passed without any modification:
Example Code :
from sqlalchemy import URL
url_object = URL.create(
"postgresql+ psycopg2",
username=" postgres ",
password="Data@12/3", # plain (unescaped) text
host="localhost”
port = 5433
database="postgres",
)
Pass this URL object in create_engine() function :
from sqlalchemy import create_engine
engine = create_engine(url_object)
3)The best way to encode the special character is to use Python's urllib.parse.quote_plus function:
Example of Urllib
from urllib.parse import quote_plus
from sqlalchemy import create_engine
password = 'Data@12/3'
encoded_password = quote_plus(password)
engine = create_engine(f"postgresql+psycopg2:// postgres: {encoded_password}@localhost:5433/ postgres ")
Database API(DBAPI):
The Actual drivers that talk to the database.
The PostgreSQL dialect uses psycopg2 as the default DBAPI. Other PostgreSQL DBAPIs include pg8000 and asyncpg:
Example code:
· engine = create_engine(f"postgresql+psycopg2:// postgres: {encoded_password}@localhost:5433/ postgres ")
· engine = create_engine(f"postgresql+ asyncpg:// postgres: {encoded_password}@localhost:5433/ postgres ")
· engine = create_engine(f"postgresql+ pg8000:// postgres: {encoded_password}@localhost:5433/ postgres ")
Getting a Connection
The methods engine.connect() opens a DBAPI connection. And Pandas read_sql() method is more flexible — it accepts raw SQL strings directly and handles them internally.
Example code
with engine.connect() as conn:
dfResult = pd.read_sql(query, conn)
print(dfResult)
Dataset Used :
Treadmill Maximal Exercise Tests from the Exercise Physiology and Human Performance Lab of the University of Malaga.
Sample Code :
I did an analysis on Maximal Exercise Tests from the Exercise Physiology and Human Performance Lab of the University of Malaga.
Connected to database using PostgreSQL and loaded the csv data in to the PostgreSQL and queried the details of participants in test 1 and age > 50
import psycopg2 # it is a driver for postgres
import pandas as pd
from sqlalchemy import create_engine
df1=pd.read_csv('subject-info.csv')
df2=pd.read_csv('test_measure.csv')
merged_df=pd.merge(df1, df2, on=['ID','ID_test'],how='outer')
# Connect to PostgreSQL
engine = reate_engine("postgresql+psycopg2://postgres:
Data%40123@localhost:5433/postgres")
# Upload to database
df2.to_sql("test_measure", engine, if_exists="replace", index=False)
df1.to_sql('subject_inform',engine,if_exists='replace',index=False)
merged_df.to_sql('subject_test',engine,if_exists='replace',index=False)
query = """SELECT * FROM subject_test
WHERE "ID_test" LIKE '%%1' AND "Age" > 50;"""
with engine.connect() as conn:
dfResult = pd.read_sql(query, conn)
print(dfResult)
SQLAlchemy ORM:
"Object Relational Mapping (ORM) lets you interact with databases using Python classes instead of raw SQL."
This structure is known as a Declarative Mapping, defines at once both a Python object model, as well as database metadata that describes real SQL tables that exist in a particular database.
· The mapping starts with a base class and is created by calling upon the declarative_base() function, which produces a new base class.
· Create the Individual mapped classes by making subclasses of Base. A mapped class typically refers to a single particular database table. The name of the table should be declared as tablename class-level attribute.
· Next, columns that are part of the table are declared, this database column, should be declared such as Integer and String
· All ORM mapped classes require at least one column be declared as part of the primary key, typically by using the Column.primary_key .
Installation :
pip install sqlalchemy psycopg2
Import and Base Setup:
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
Create Base Clase, Table and Columns:
Base = declarative_base()
class User(Base): tablename = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
Connecting to a Database :
Create_engine() function will create the engine object , Base.metadata collects table and schema definitions, Base.metadata.create_all() method tells SQLAlchemy to create all tables in the database that are defined by the ORM classes.
engine = create_engine("postgresql+psycopg2://postgres:Data%40123@
Base.metadata.create_all(engine) # Creates the table
Creating a Session :
Session = sessionmaker(bind=engine)
session = Session()
sessionmaker() function that configures sessions and bind=engine, tells SQLAlchemy which database engine the sessions should connect to.
Query Execution:
· Create Operation:
Creating new user in the database with name is Alice and the email id is alice@example.com.
new_user = User(name="Alice", email="alice@example.com")
session.add(new_user)
session.commit()
users = session.query(User).all()
· Read Operation:
Read the user from the database whose name is ‘Alice’
user = session.query(User).filter_by(name="Alice").first()
session.commit()
· Update Operation:
Find the record with name Alice and update the email id of the particular record
user = session.query(User).filter_by(name="Alice").first()
session.commit()
· Delete Operation
session.delete(user)
session.commit()
Dataset Used :
Treadmill Maximal Exercise Tests from the Exercise Physiology and Human Performance Lab of the University of Malaga.
Sample Code(ORM) :
Connected to database using PostgreSQL and retrieved the details of participants in test 1 and age > 50
# Step 1: ORM base setup
Base = declarative_base()
# Step 2: Define ORM class for subject_test table
class SubjectTest(Base): tablename = 'subject_test'
ID = Column(Integer, primary_key=True)
ID_test = Column(String, primary_key=True)
Age = Column(Integer)
Temperature = Column(Float)
Humidity = Column(Float)
# Add more fields based on your actual table structure
# Step 3: Database connection
engine = create_engine("postgresql+psycopg2://postgres:Data%40123@localhost:5433/postgres")
Session = sessionmaker(bind=engine)
session = Session()
# Step 4: ORM-style query (equivalent to WHERE "ID_test" LIKE '%1' AND Age > 50)
results = session.query(SubjectTest).filter(SubjectTest.ID_test.like('%1'), SubjectTest.Age > 50).all()
# Step 5: Display results
for row in results:
print(f"ID: {row.ID}, ID_test: {row.ID_test}, Age: {row.Age}, Temp: {row.Temperature}, Humidity: {row.Humidity}")
session.close()
