This article explores two primary approaches for interacting with MySQL databases using Python: the low-level driver PyMySQL and the high-level ORM framework SQLAlchemy.

PyMySQL

PyMySQL is a pure-Python MySQL client library. Its API closely resembles that of MySQLdb, making it a straightforward tool for executing raw SQL.

Installation:

pip3 install pymysql

Basic Operations

1. Executing SQL Statements

import pymysql

connection = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor_obj = connection.cursor()

row_count = cursor_obj.execute("UPDATE hosts SET host = '1.1.1.2'")

# Parameterized query
# row_count = cursor_obj.execute("UPDATE hosts SET host = '1.1.1.2' WHERE nid > %s", (10,))

# Batch insert
# row_count = cursor_obj.executemany("INSERT INTO hosts (host, color_id) VALUES (%s, %s)", [("10.0.0.1", 1), ("10.0.0.2", 2)])

connection.commit()
cursor_obj.close()
connection.close()

2. Retrieving Auto-Incremented ID

import pymysql

connection = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor_obj = connection.cursor()
cursor_obj.executemany("INSERT INTO hosts (host, color_id) VALUES (%s, %s)", [("10.0.0.1", 1), ("10.0.0.2", 2)])
connection.commit()

last_id = cursor_obj.lastrowid

cursor_obj.close()
connection.close()

3. Fetching Query Results

import pymysql

connection = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor_obj = connection.cursor()
cursor_obj.execute("SELECT * FROM hosts")

first_record = cursor_obj.fetchone()
# Fetch the next three rows
# some_records = cursor_obj.fetchmany(3)
# Fetch all remaining rows
# all_records = cursor_obj.fetchall()

connection.commit()
cursor_obj.close()
connection.close()

Cursors fetch rows sequentially. To reposition, use scroll:

• cursor_obj.scroll(1, mode='relative') moves one row forward from the current position.
• cursor_obj.scroll(2, mode='absolute') jumps to the second row.

4. Fetching as Dictionaries

By default, rows are returned as tuples. For dictionary-style results, specify a DictCursor:

import pymysql

connection = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor_dict = connection.cursor(cursor=pymysql.cursors.DictCursor)
cursor_dict.execute("CALL p1()")

row_result = cursor_dict.fetchone()

connection.commit()
cursor_dict.close()
connection.close()

SQLAlchemy ORM

SQLAlchemy is a comprehensive ORM (Object Relational Mapper) for Python. It translates Python classes and their relationships into SQL statements executed via a database driver like PyMySQL.

Installation:

pip3 install SQLAlchemy

SQLAlchemy operates through pluggable Dialects that communicate with specific database APIs. The connection URL defines which driver to use:

• mysql+pymysql://<user>:<pass>@<host>/<db> — PyMySQL
• mysql+mysqldb://<user>:<pass>@<host>/<db> — MySQLdb
• oracle+cx_oracle://user:pass@host:port/dbname — cx_Oracle

Low-Level Engine Interaction

Using the Engine, ConnectionPooling, and Dialect directly:

from sqlalchemy import create_engine

db_engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

# A single insert
# result_proxy = db_engine.execute("INSERT INTO hosts (host, color_id) VALUES ('1.1.1.22', 3)")
# auto_id = result_proxy.lastrowid

# Execute a SELECT
# result_proxy = db_engine.execute('SELECT * FROM hosts')
# row_one = result_proxy.fetchone()
# row_many = result_proxy.fetchmany(3)
# all_rows = result_proxy.fetchall()

ORM Usage

The full ORM stack (Declarative Base, Session, Mapped Classes) provides a Pythonic interface.

1. Defining Models (Schema Creation)

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

db_engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
Base = declarative_base()

class UserProfile(Base):
    __tablename__ = 'user_profile'
    uid = Column(Integer, primary_key=True)
    username = Column(String(32))
    metadata_field = Column(String(16))

    __table_args__ = (
        UniqueConstraint('uid', 'username', name='uq_uid_user'),
        Index('ix_user_meta', 'username', 'metadata_field'),
    )

class Category(Base):
    __tablename__ = 'category'
    cid = Column(Integer, primary_key=True)
    title = Column(String(50), default='default', unique=True)

class Item(Base):
    __tablename__ = 'item'
    iid = Column(Integer, primary_key=True)
    label = Column(String(32), index=True, nullable=True)
    category_id = Column(Integer, ForeignKey("category.cid"))

class Cluster(Base):
    __tablename__ = 'cluster'
    cid = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port_num = Column(Integer, default=22)

class Node(Base):
    __tablename__ = 'node'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    host = Column(String(64), unique=True, nullable=False)

class NodeClusterMap(Base):
    __tablename__ = 'node_cluster_map'
    mapping_id = Column(Integer, primary_key=True, autoincrement=True)
    node_id = Column(Integer, ForeignKey('node.nid'))
    cluster_id = Column(Integer, ForeignKey('cluster.cid'))

def init_db():
    Base.metadata.create_all(db_engine)

def drop_db():
    Base.metadata.drop_all(db_engine)

Alternative ForeignKey definition: ForeignKeyConstraint(['column_local'], ['other_table.other_col'])

2. CRUD Operations

Ensure a session is created to manage transactions:

SessionMaker = sessionmaker(bind=db_engine)
db_session = SessionMaker()

Insert

entry = UserProfile(username="alex0", metadata_field='demo')
db_session.add(entry)
db_session.add_all([
    UserProfile(username="alex1", metadata_field='demo'),
    UserProfile(username="alex2", metadata_field='demo'),
])
db_session.commit()

Delete

db_session.query(UserProfile).filter(UserProfile.uid > 2).delete()
db_session.commit()

Udpate

db_session.query(UserProfile).filter(UserProfile.uid > 2).update({"username": "updated"})
db_session.query(UserProfile).filter(UserProfile.uid > 2).update(
    {UserProfile.username: UserProfile.username + "_suffix"},
    synchronize_session=False
)
db_session.commit()

Select

rows = db_session.query(UserProfile).all()
limited_columns = db_session.query(UserProfile.username, UserProfile.metadata_field).all()
filtered = db_session.query(UserProfile).filter_by(username='alex').all()
first_match = db_session.query(UserProfile).filter_by(username='alex').first()

# Using textual SQL with parameters
from sqlalchemy import text
result_textual = db_session.query(UserProfile).filter(
    text("uid < :bound_id AND username = :uname")
).params(bound_id=224, uname='fred').all()

Additional Query Patterns

# Conditions
result = db_session.query(UserProfile).filter(UserProfile.uid.between(1, 3), UserProfile.username == 'eric').all()
result = db_session.query(UserProfile).filter(UserProfile.uid.in_([1, 3, 4])).all()
result = db_session.query(UserProfile).filter(~UserProfile.uid.in_([1, 3, 4])).all()

from sqlalchemy import and_, or_
result = db_session.query(UserProfile).filter(or_(UserProfile.uid < 2, UserProfile.username == 'eric')).all()

# Wildcards
result = db_session.query(UserProfile).filter(UserProfile.username.like('e%')).all()
result = db_session.query(UserProfile).filter(~UserProfile.username.like('e%')).all()

# Slicing
result_slice = db_session.query(UserProfile)[1:3]

# Ordering
result_order = db_session.query(UserProfile).order_by(UserProfile.username.desc()).all()

# Aggregation & Grouping
from sqlalchemy.sql import func
result_agg = db_session.query(
    func.max(UserProfile.uid),
    func.min(UserProfile.uid)
).group_by(UserProfile.username).having(func.min(UserProfile.uid) > 2).all()

# Join examples
result_join = db_session.query(Item, Category).filter(Item.category_id == Category.cid).all()
result_outer = db_session.query(Item).join(Category, isouter=True).all()

# Union
q1 = db_session.query(UserProfile.username).filter(UserProfile.uid > 2)
q2 = db_session.query(Category.title).filter(Category.cid < 2)
combined = q1.union(q2).all()