SQLAlchemy Query Examples

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)