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)