Using SQLAlchemy

Python ORM in practice: Use SQLAlchemy to operate the database elegantly

In application development, almost every project cannot bypass the database. Although it is straightforward to directly use SQL statements to operate the database, when the project scale becomes larger and the table structure becomes more complex, it will become more and more difficult to maintain these scattered SQL codes. The emergence of ORM (Object Relational Mapping) is to solve this problem. It allows us to operate the database in a familiar object-oriented way, and SQLAlchemy is one of the most mature and powerful ORM frameworks in the Python ecosystem. This article will take you from installation and configuration to advanced-features, and master the core usage of SQLAlchemy step by step.

1. What is ORM? Why is it needed?

The core idea of ORM is: database table ↔ Python class, a row of records in the table ↔ an instance of the class, columns in the table ↔ attributes of the class.

Traditional method vs ORM method

When not using ORM, the results we query from the database are usually a list of tuples:

# 传统方式：返回原始元组
[
    ('1', 'Michael'),
    ('2', 'Bob'),
    ('3', 'Adam')
]

This method requires us to manually handle the order and type of fields, which is especially error-prone when involving multi-table related queries. Through ORM, we can process data in the form of objects:

# ORM 方式：返回对象列表
[
    User(id=1, name='Michael'),
    User(id=2, name='Bob'),
    User(id=3, name='Adam')
]

In this way, when we access the data, we can useuser.nameinstead ofuser[1], the readability and maintainability of the code have been greatly improved.

2. Installation and basic configuration

2.1 Install SQLAlchemy

The core package of SQLAlchemy can be installed directly through pip:

pip install sqlalchemy

If you use MySQL database, you also need to install the corresponding database driver, for examplepymysqlormysql-connector-python：

pip install pymysql
# 或者
pip install mysql-connector-python

2.2 Create database connection and session

The following is a standard basic configuration, including the definition of the connection engine, session factory and base class:

from sqlalchemy import create_engine, Column, String, Integer, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship, declarative_base

# 声明基类，所有的模型都会继承自它
Base = declarative_base()

# 数据库连接地址，格式为：数据库类型+驱动://用户名:密码@主机:端口/数据库名
DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/your_database"
engine = create_engine(DATABASE_URL, echo=True)  # echo=True 可打印 SQL 语句，调试时很有用

# 创建会话类，用于后续的数据库操作
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

hereSessionLocalIt is the factory for our subsequent CRUD operations. Each request or operation should create an independent session.

3. Define database model (table structure)

In SQLAlchemy, a model class corresponds to a table in the database. Let's first look at a simple user model:

3.1 Single table model

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)

Some common parameters in model definition:

primary_key=True: Primary key.
index=True: Create an index for this column to improve query efficiency.
unique=True: Ensure that the column value is unique.
nullable=False: Equivalent to databaseNOT NULLconstraint.

3.2 Define one-to-many and many-to-one relationships

Tables in reality often have related relationships, such as one user can publish multiple articles (one-to-many). In SQLAlchemy we can passForeignKeyandrelationshipTo establish this relationship:

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    email = Column(String(100), unique=True, index=True)

    # 定义与 Post 的一对多关系，back_populates 用于双向绑定
    posts = relationship("Post", back_populates="author")

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(100), nullable=False)
    content = Column(String(2000))
    author_id = Column(Integer, ForeignKey('users.id'))

    # 定义多对一关系
    author = relationship("User", back_populates="posts")

ForeignKey('users.id')Specifies the foreign key constraint, corresponding touserstableidfield.
relationshipIt is a relationship declaration at the ORM level, allowing us to directly passuser.postsGet all articles of this user, or bypost.authorGet the author of the article.

4. Basic database operations (CRUD)

After defining the model, we usually perform a table creation operation when the application starts:

# 创建所有继承自 Base 的表（如果表已存在会跳过）
Base.metadata.create_all(bind=engine)

Next, let's see how to perform add, delete, modify and check operations. In order to keep the code clear, each operation is encapsulated into a function and passedtry...finallyMake sure the session is closed properly.

4.1 Create records

def create_user(name: str, email: str):
    session = SessionLocal()
    try:
        user = User(name=name, email=email)
        session.add(user)       # 添加到会话
        session.commit()        # 提交事务
        session.refresh(user)   # 刷新实例，获取数据库生成的 id 等字段
        return user
    finally:
        session.close()

4.2 Query records

SQLAlchemy provides a very flexible way to build queries:

# 获取所有用户
def get_all_users():
    session = SessionLocal()
    try:
        return session.query(User).all()
    finally:
        session.close()

# 按主键查询单个用户
def get_user_by_id(user_id: int):
    session = SessionLocal()
    try:
        return session.query(User).filter(User.id == user_id).first()
    finally:
        session.close()

# 条件查询（例如按邮箱查找）
def get_user_by_email(email: str):
    session = SessionLocal()
    try:
        return session.query(User).filter(User.email == email).first()
    finally:
        session.close()

Common query methods:

all(): Returns a list of all results.
first(): Return the first result, if not, returnNone。
filter(): Add filter conditions, multiple calls can be made in a chain.
filter_by(): Easier, use keyword arguments for filtering (e.g.filter_by(email=email)）。

4.3 Update records

Updates generally require first finding the record to be modified, then modifying the attributes, and then submitting the transaction:

def update_user(user_id: int, name: str, email: str):
    session = SessionLocal()
    try:
        user = session.query(User).filter(User.id == user_id).first()
        if user:
            user.name = name
            user.email = email
            session.commit()
            session.refresh(user)
        return user
    finally:
        session.close()

4.4 Delete records

The deletion operation is similar, query first and then calldelete()method:

def delete_user(user_id: int):
    session = SessionLocal()
    try:
        user = session.query(User).filter(User.id == user_id).first()
        if user:
            session.delete(user)
            session.commit()
        return user
    finally:
        session.close()

5. Manipulate associated objects

One of the advantages of ORM is that it can naturally handle the relationships between objects. Suppose we want to create a user and add several articles to it at the same time. We can do this:

def create_user_with_posts(name: str, email: str, post_data: list[dict]):
    session = SessionLocal()
    try:
        user = User(name=name, email=email)
        for data in post_data:
            post = Post(**data)
            user.posts.append(post)   # 利用 relationship 自动处理外键
        session.add(user)
        session.commit()
        session.refresh(user)
        return user
    finally:
        session.close()

When querying a user, if we want all his articles to be loaded together (to avoid the N+1 query problem), we can usejoinedloadTo preload:

from sqlalchemy.orm import joinedload

def get_user_with_posts(user_id: int):
    session = SessionLocal()
    try:
        user = session.query(User).filter(User.id == user_id).options(
            joinedload(User.posts)
        ).first()
        return user
    finally:
        session.close()

6. advanced-features

6.1 Transaction Management

For update operations involving multiple records, transactions can guarantee data consistency. SQLAlchemy enables transactions by default, we need to manually commit or rollback:

def transfer_amount(from_user_id: int, to_user_id: int, amount: float):
    session = SessionLocal()
    try:
        from_user = session.query(User).filter(User.id == from_user_id).first()
        to_user = session.query(User).filter(User.id == to_user_id).first()

        if not from_user or not to_user:
            raise ValueError("用户不存在")

        # 假设 User 有一个 balance 字段
        if from_user.balance < amount:
            raise ValueError("余额不足")

        from_user.balance -= amount
        to_user.balance += amount

        session.commit()   # 全部成功，提交事务
    except Exception as e:
        session.rollback() # 出现异常，回滚所有更改
        raise e
    finally:
        session.close()

6.2 Asynchronous support (for SQLAlchemy 2.0+)

In high-concurrency scenarios, asynchronous operations can significantly improve performance. SQLAlchemy 2.0 provides complete asynchronous support:

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select

DATABASE_URL_ASYNC = "mysql+asyncmy://user:pass@localhost/your_database"
async_engine = create_async_engine(DATABASE_URL_ASYNC, echo=True)
AsyncSessionLocal = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)

async def async_get_user(user_id: int):
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User).where(User.id == user_id))
        user = result.scalars().first()
        return user

When using asynchronous, pay attention to the matchingasyncmyoraiomysqlWait for asynchronous driver.

7. Some practical suggestions

Manage session life cycle: It is recommended to use a context manager to encapsulate the session, or use dependency injection (such as FastAPI'sDepends) automatically handles the creation and closing of sessions.
Avoid long transactions: Keep transactions as short as possible to reduce the time of locking resources, especially in high concurrent write scenarios.
Batch Operation: Used when a large amount of data needs to be inserted or updatedsession.bulk_insert_mappings()orsession.bulk_save_objects()can get better performance.
Reasonable selection of loading strategies: For associated attributes, choose according to business scenariosjoinedload(preloaded),selectinload(Secondary query preloading) or keep the default lazy loading.
Make good use of type hints: With modern IDEs, type hints allow you to get better auto-completion and error checking when writing queries.

8. Summary

As the most mature ORM framework in Python, SQLAlchemy not only retains the expressive power of original SQL, but also provides an object-oriented operation interface. It can handle everything from simple single-table CRUD to complex relationship processing, transaction control, and asynchronous operations. Through the introduction of this article, I believe you have mastered the core usage of SQLAlchemy. Next, try replacing original SQL operations with SQLAlchemy in your project to experience the improvement in development efficiency it brings.

If you want to learn more, I recommend reading SQLAlchemy 官方文档, which contains a lot of advanced usage and best practices.

#Using SQLAlchemy

#Python ORM in practice: Use SQLAlchemy to operate the database elegantly

#1. What is ORM? Why is it needed?

#Traditional method vs ORM method

#2. Installation and basic configuration

#2.1 Install SQLAlchemy

#2.2 Create database connection and session

#3. Define database model (table structure)

#3.1 Single table model

#3.2 Define one-to-many and many-to-one relationships

#4. Basic database operations (CRUD)

#4.1 Create records

#4.2 Query records

#4.3 Update records

#4.4 Delete records

#5. Manipulate associated objects

#6. advanced-features

#6.1 Transaction Management

#6.2 Asynchronous support (for SQLAlchemy 2.0+)

#7. Some practical suggestions

#8. Summary