Convert IP to Integer and Integer To IP in Python How to Get IPv4/IPv6 Address Range from CIDR in Python? Compare Two Objects For Equality in Python How to find Duplicate Elements from a List in Python Convert Timestamp to datetime in Python Convert datetime to Timestamp in Python Generate Random String of Specific Length in Python Encryption and Decryption of Strings in Python The string module in Python Convert string to bytes in Python Convert bytes to string in Python Convert string to datetime and datetime to string in Python Call a function Asynchronously from within a Loop and wait for the Results in Python Remove Duplicate Elements from a List in Python Caching in Python with Examples How to Bulk Insert and Retrieve Data from Redis in Python How to Write Unit Test in Python Read and Write CSV Files in Python Read and Write Data to a Text File in Python How to Convert CSV to JSON in Python Create ICS Calendar File in Python Install Python on Windows 10/11 Install Python on Ubuntu 20.04 or 22.04.3 Python - Install Virtual Environment How to Find a Specific Field Value from a JSON list in Python Download and Unzip a Zipped File in Python Python Install PIP Python Install Virtual Environment How to Fix Python Error: string argument without an encoding Compare Two JSON files in Python How to Hash a Dictionary Object in Python? Create a Digital Clock in Python Create Multiple URLs Using Each Path of a URL in Python Send an Email with Multiple Attachments using Amazon SES in Python SQLAlchemy Query Examples for Effective Database Management SQLAlchemy Query to Find IP Addresses from an IP Range in Bulk How to Create and Use Configuration files in a Python Project Check if a Value Already Exists in a List of Dictionary Objects in Python How to Split Large Files by size in Python? Fixing - Running Scripts is Disabled on this System Error on Windows Generating QR Codes in Python Reading QR Codes in Python

SQLAlchemy Query Examples for Effective Database Management

Last updated Apr 25, 2024

SQLAlchemy is a Python Object Relational Mapper library that makes working with databases in Python easier and more user-friendly by simplifying SQL complexities.

To interact with a database, you must start by creating an SQLAlchemy engine, which defines the connection details. Additionally, you'll need to establish the structure of the table you wish to work with using SQLAlchemy's ORM. Finally, to interact with the database, you need to create a session. For example:

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, TIMESTAMP, INTEGER, String, text

Base = declarative_base()
metadata = Base.metadata

# Defining a Table
class User(Base):
    __tablename__ = "user"
    id = Column(INTEGER, primary_key=True)
    name = Column(String(500))
    age = Column(INTEGER)
    created_date = Column(TIMESTAMP, nullable=False,
                            server_default=text("CURRENT_TIMESTAMP"))
    
# Database configuration
db_conn = {
      "host": "localhost",
      "username": "root",
      "password": "Testing123",
      "port": 3360,
      "database_name": "my_database",
      "connection_timeout": 60,
      "pool_size": 20,
      "pool_recycle": 10,
      "pool_timeout": 60,
      "max_overflow": 60
    }

try:
    mysql_string = "mysql+pymysql://"+db_conn.get("username")+":"+db_conn.get("password")+"@"+db_conn.get("host")+":"+str(db_conn.get("port"))+"/"+db_conn.get("database_name")+"?charset=utf8"
    connect_args = {
                "connect_timeout": db_conn.get("connection_timeout")}
    engine = create_engine(mysql_string, pool_size=db_conn.get("pool_size"), pool_recycle=db_conn.get("pool_recycle"), pool_timeout=db_conn.get("pool_timeout"), max_overflow=db_conn.get("max_overflow"), connect_args=connect_args)
    Session = sessionmaker(bind=engine)
    session = Session()   
except Exception as ex:
        print("Error connecting to db: " + str(ex))

Here are some of the most commonly used operators in SQLAlchemy are given below with examples:

AND

from sqlalchemy import and_

session.query(User).filter(and_(User.name == "Danny", User.age == 20))

EQUALS ==

session.query(User).filter(User.id == 1)

NOT EQUALS !=

session.query(User).filter(User.id != 1)

IN

session.query(User).filter(User.id.in_([1, 2, 3, 4]))

NOT IN

session.query(User).filter(~User.id.in_([1, 2, 3, 4]))

IS NULL

session.query(User).filter(User.name == None)

IS NOT NULL

session.query(User).filter(User.name != None)

LIKE

session.query(User).filter(User.name.like("%Danny%"))

MATCH

session.query(User).filter(User.name.match("Danny"))

OR

from sqlalchemy import or_

session.query(User).filter(or_(User.name == "Danny", User.name == "Jenifer", User.name == "Peter"))

INSERT Single Object

session.add(user)
session.commit()
session.close()

INSERT List of Objects

session.add_all(user_list)
session.commit()
session.close()

UPDATE

session.query(User).filter(User.id == 1).update({User.name: "Lee", User.active: True}, synchronize_session=False)
session.flush()
session.commit()

Sorting

from sqlalchemy import desc

sorted_users = session.query(User).order_by(desc(User.created_date)).all()

JOIN Two Tables

session.query(User).join(Address, User.id == Address.user_id).filter(User.id == 1)

Python Examples