Comprehensive Guide to SQLAlchemy ORM in Python

SQLAlchemy is a Python ORM framework built upon the DB API, translating classes and objects into SQL, executing it, and retrieving results. It can create and drop tables but not alter structures.

Install via pip:

pip3 install sqlalchemy

Core components:

Engine: the framework's core
Connection Pooling: manages database connections
Dialect: selects the DB API for the databace
Schema/Types: defines table structures and data types
SQL Expression Language: provides SQL-like expressions

SQLAlchemy relies on third-party drivers like pymysql. The Dialect communicates with the chosen DB API based on configuration, for example:

MySQL-Python:
    mysql+mysqldb://user:password@host[:port]/dbname

PyMySQL:
    mysql+pymysql://user:password@host/dbname[?options]

MySQL-Connector:
    mysql+mysqlconnector://user:password@host[:port]/dbname

cx_Oracle:
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

Defining Table Structures

Single Table

from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, UniqueConstraint, Index
from sqlalchemy.ext.declarative import declarative_base
import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    notes = Column(Text, nullable=True)

    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'email'),
    )

def create_tables():
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/mydb?charset=utf8",
        max_overflow=0, pool_size=5, pool_timeout=30, pool_recycle=-1
    )
    Base.metadata.create_all(engine)

def drop_tables():
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/mydb?charset=utf8",
        max_overflow=0, pool_size=5, pool_timeout=30, pool_recycle=-1
    )
    Base.metadata.drop_all(engine)

if __name__ == '__main__':
    drop_tables()
    create_tables()

Multiple Tables with ForeignKey and Many-to-Many

from sqlalchemy import create_engine, Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    email = Column(String(32), unique=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    extra = Column(Text, nullable=True)

class Host(Base):
    __tablename__ = 'hosts'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

class Hobby(Base):
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), default='Basketball')

class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.id"))
    hobby = relationship("Hobby", backref='persons')

class ServerGroup(Base):
    __tablename__ = 'server_group'
    id = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    servers = relationship('Server', secondary='server_group', backref='groups')

class Server(Base):
    __tablename__ = 'server'
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)

def init_db():
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
        max_overflow=0, pool_size=5, pool_timeout=30, pool_recycle=-1
    )
    Base.metadata.create_all(engine)

def drop_db():
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0, pool_size=5, pool_timeout=30, pool_recycle=-1
    )
    Base.metadata.drop_all(engine)

if __name__ == '__main__':
    drop_db()
    init_db()

Connecting to the Database

Basic Connection

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
import models

engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# Perform ORM operations
user = models.User(name="derek", email='derek@163.com')
session.add(user)
session.commit()

# Return connection to pool
session.close()

Thread-Safe Connection with scoped_session

from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import create_engine
import models

engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)

# Perform ORM operations
user = models.User(name="jack", email='jack@163.com')
session.add(user)
session.commit()

session.close()

Basic ORM Operations

Adding Records

# Single addition
user = models.User(name="jack", email='jack@163.com')
session.add(user)
session.commit()

# Batch addition
session.add_all([
    models.User(name="jack1", email='jack1@163.com'),
    models.User(name="jack2", email='jack2@163.com'),
])
session.commit()

Querying Data

user_list = session.query(models.User).all()
for user in user_list:
    print(user.id, user.name, user.email)

# Filtering
user_list = session.query(models.User).filter(models.User.id > 2).all()

Deleting Records

session.query(models.User).filter(models.User.id > 4).delete()
session.commit()

Updating Records

session.query(models.User).filter(models.User.id == 4).update({'name': 'Tom'})
session.query(models.User).filter(models.User.id == 1).update({'name': models.User.name + "099"}, synchronize_session=False)
session.query(models.User).filter(models.User.id == 3).update({"age": models.User.age + 1}, synchronize_session="evaluate")
session.commit()

Advanced Query Techniques

# Conditions
ret = session.query(User).filter_by(name='alex').all()
ret = session.query(User).filter(User.id > 1, User.name == 'eric').all()
ret = session.query(User).filter(User.id.between(1, 3), User.name == 'eric').all()
ret = session.query(User).filter(User.id.in_([1,3,4])).all()
ret = session.query(User).filter(~User.id.in_([1,3,4])).all()

# And/Or
from sqlalchemy import and_, or_
ret = session.query(User).filter(and_(User.id > 3, User.name == 'eric')).all()
ret = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()

# Wildcard
ret = session.query(User).filter(User.name.like('e%')).all()
ret = session.query(User).filter(~User.name.like('e%')).all()

# Limiting
ret = session.query(User)[1:2]

# Sorting
ret = session.query(User).order_by(User.name.desc()).all()
ret = session.query(User).order_by(User.name.desc(), User.id.asc()).all()

# Grouping
from sqlalchemy.sql import func
ret = session.query(func.max(User.id), func.sum(User.id), func.min(User.id)).group_by(User.name).all()
ret = session.query(func.max(User.id), func.sum(User.id), func.min(User.id)).group_by(User.name).having(func.min(User.id) > 2).all()

# Joins
ret = session.query(User, Hobby).filter(User.id == Hobby.id).all()
ret = session.query(Person).join(Hobby).all()
ret = session.query(Person).join(Hobby, isouter=True).all()  # Left join

# Set Operations
q1 = session.query(User.name).filter(User.id > 2)
q2 = session.query(Hobby.name).filter(Hobby.id < 2)
ret = q1.union(q2).all()
ret = q1.union_all(q2).all()

Using Raw SQL

from sqlalchemy import text

cursor = session.execute('select * from users')
result = cursor.fetchall()

cursor = session.execute('insert into users(name) values(:value)', params={"value":'example'})
session.commit()
print(cursor.lastrowid)

session.close()

Cross-Tible Relationships

Adding Related Data

session.add_all([
    models.Hobby(name="Reading"),
    models.Hobby(name="Swimming")
])
session.commit()

session.add_all([
    models.Person(name="Alice", hobby_id=2),
    models.Person(name="Bob", hobby_id=2),
    models.Person(name="Charlie", hobby_id=1),
])
session.commit()

Querying from One to Many

person = session.query(models.Person).filter(models.Person.id == 1).first()
print(person.hobby_id)

Querrying from Many to One

hobby = session.query(models.Hobby).filter(models.Hobby.id == 1).first()
for person in hobby.persons:
    print(person.name)

Inserting with Relationships

hobby = models.Hobby(name='Gaming')
hobby.persons = [models.Person(name='David'), models.Person(name='Emma')]
session.add(hobby)
session.commit()

Tags: SQLAlchemy Python ORM database MySQL Tutorial

Posted on Mon, 26 Oct 2026 16:16:12 +0000 by methodlessman

Freaks City