Before We Begin

It is best to first read SQLAlchemy Basic and then come back to this section.

Flask-SQLAlchemy Basic

for flask-sqlalchemy 3.x and sqlalchemy 2.x

Initialize

1
2
3
4
5
6
7
8

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
  pass

db = SQLAlchemy(model_class=Base)

Detailed Explanation of the Initialization Code

flask VS Flask

flask (lowercase) is the name of the Python package. This is the standard naming style for a module/package.

Flask (capitalized) is the class name. In Python, class names typically start with an uppercase letter (PascalCase naming convention).

This is a common Python naming convention, not something confusing: - Package/module names use lowercase (for example: flask, os, sys) - Class names start with uppercase letters (for example: Flask, String, Dict)

SQLAlchemy in flask_sqlalchemy

The SQLAlchemy class is an integration class. Its main purpose is:

To act as a bridge between Flask and SQLAlchemy, providing a unified interface for managing:

database connections, session management, model definitions, and query construction

db = SQLAlchemy(model_class=Base)

Native SQLAlchemy Usage

In native SQLAlchemy, we define models like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 1. 定义基类
class Base(DeclarativeBase):
    pass

# 2. 定义模型类
class User(Base):
    __tablename__ = 'users'  # 必须指定表名
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str]

Flask-SQLAlchemy Usage

In Flask-SQLAlchemy:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 1. 创建基类和db实例
class Base(DeclarativeBase):
    pass
db = SQLAlchemy(model_class=Base)

# 2. 定义模型类
class User(db.Model):  # 使用 db.Model 而不是 Base
    # 不需要 __tablename__，会自动生成
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str]

Key Points Explained

What does db = SQLAlchemy(model_class=Base) do?

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Flask-SQLAlchemy 内部大致实现（简化版）
class SQLAlchemy:
    def __init__(self, model_class=None):
        # 将传入的 Base 类扩展，添加 Flask-SQLAlchemy 特定的功能
        self.Model = self.make_declarative_base(model_class)

    def make_declarative_base(self, model_class):
        # 如果没有提供 model_class，创建一个新的
        if model_class is None:
            model_class = DeclarativeBase

        # 添加 Flask-SQLAlchemy 特定的功能
        class Model(model_class):
            # 自动生成表名的功能
            @declared_attr
            def __tablename__(cls):
                return cls.__name__.lower()

            # 添加查询助手
            query = QueryProperty()

            # 添加其他 Flask-SQLAlchemy 特定的功能...

        return Model

Why use db.Model?

• db.Model is a base class enhanced by Flask-SQLAlchemy
• It inherits from the original Base class, but adds extra functionality:
• Automatically generates table names (no need for __tablename__)
• Integrates Flask's context management
• Provides a more convenient query interface
• Adds session management features

The relationship between Flask-SQLAlchemy and SQLAlchemy

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

Flask-SQLAlchemy
     │
     ├── 提供 Flask 集成
     ├── 自动管理数据库连接
     ├── 处理应用上下文
     └── 简化配置和使用
         │
         ▼
   SQLAlchemy (核心)
     │
     ├── 数据库抽象层
     ├── SQL 查询构建
     └── ORM 功能

Define and Create Table

Define TABLE

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

class Base(DeclarativeBase):
    pass

db = SQLAlchemy(model_class=Base)

# 文章元数据
class Article_Meta_Data(db.Model):
    # 指定数据模型在数据库中的表名称 如果未指定那么为类名称的小写
    __tablename__ = 'article_meta_date'
    # 主键 但是无需为其赋值 SQLite数据库会自动为其生成一个唯一的值
    id: Mapped[int] = mapped_column(Integer, primary_key=True)

    # 文章标题 最长不超过100个字 默认nullable=False
    title: Mapped[str] = mapped_column(String(100))

    # 文章发布时间
    rollout_date: Mapped[date] = mapped_column(Date)

    # 表示文章最后更新的日期 只精确到年月日
    ultimate_modified_date: Mapped[date] = mapped_column(Date)

    # 文章作者 最长不超过50个字符
    author: Mapped[str] = mapped_column(String(50))

    # 文章指导者 存在Optional 默认nullable=True
    instructor: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)

    # 文章内容简介
    brief_introduction: Mapped[str] = mapped_column(Text)

    # 文章封面链接
    cover_image_url: Mapped[str] = mapped_column(String(100))

    # 文章分类
    category: Mapped[str] = mapped_column(String(100))

    # 文章分类 使用 mptt 待开发和测试
    # category_id = db.Column(db.Integer, db.ForeignKey('category.id'), nullable=False)
    # category = db.relationship('Category')

    def __repr__(self):
        return f'<Article {self.title}>'

Because of how SQLAlchemy and Flask-SQLAlchemy work:

1. When your model class inherits from db.Model, that model is automatically registered in SQLAlchemy's metadata.
2. When you call db.create_all(), SQLAlchemy checks all registered model classes and creates their tables.

However, there is one important prerequisite here: your model class must be imported and executed by the Python interpreter before db.create_all() is called.

Create Table

1
2

with app.app_context():
    db.create_all()

This code will:

• Create tables that do not exist
• Not modify tables that already exist
• Not delete tables that already exist
• Not overwrite existing data

db.create_all()

Let's look at how create_all() works:

1
2
3
4
5
6
7
8

# 简化版的内部实现逻辑
def create_all(self):
    # 1. 获取所有表的元数据
    for table in self.metadata.sorted_tables:
        # 2. 检查表是否存在
        if not table.exists():
            # 3. 如果表不存在，创建表
            table.create()

Common scenario example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 初始模型
class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]

# 执行 db.create_all() -> 创建表

# 后来修改模型，添加新字段
class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str]
    email: Mapped[str]  # 新增字段

# 再次执行 db.create_all() -> 不会添加新字段！

note

db.create_all() will automatically create the database file based on app.config['SQLALCHEMY_DATABASE_URI']

How should schema updates be handled?

Delete and recreate (development environment):

1
2
3

with app.app_context():
    db.drop_all()   # 删除所有表
    db.create_all() # 重新创建

Use database migrations (recommended, production environment):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# 使用 Flask-Migrate
from flask_migrate import Migrate
migrate = Migrate(app, db)

# 初始化迁移
$ flask db init

# 创建迁移脚本
$ flask db migrate -m "Add email column"

# 应用迁移
$ flask db upgrade

app.app_context()

Application instance and its context

First, we need to understand that every Flask object represents an independent application instance. When you create a Flask application, you are actually creating an instance of that application:

1

app = Flask(__name__)

This app object is a specific application instance. And app.app_context() creates an application context, which is an independent execution environment where operations are associated with a specific application instance.

Flask uses _app_ctx_stack to track the currently active application context. When you use with app.app_context()::

The context of that application is pushed onto _app_ctx_stack.

Inside the with block, the current_app proxy points to the application context at the top of the stack. When the with block exits, the context is popped.

1
2
3
4

with app.app_context():
    # The context of this application is pushed to the top of the stack
    ...
# When the with block exits, the context is popped

In multi-application scenarios, each application has its own context. For example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

app1 = Flask('app1')
app2 = Flask('app2')

with app1.app_context():
    # Operations here belong to app1
    ...

with app2.app_context():
    # Operations here belong to app2
    ...

Let's look at a complete example to understand this process:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

from flask import Flask, current_app

app1 = Flask('app1')
app2 = Flask('app2')

@app1.route('/')
def index1():
    return f"This is {current_app.name}"

@app2.route('/')
def index2():
    return f"This is {current_app.name}"

# 在脚本中使用
with app1.app_context():
    print(current_app.name)  # 输出: app1

with app2.app_context():
    print(current_app.name)  # 输出: app2

# 嵌套使用
with app1.app_context():
    print(current_app.name)  # 输出: app1
    with app2.app_context():
        print(current_app.name)  # 输出: app2
    print(current_app.name)  # 输出: app1

note:

In CLI commands: if you use the @app.cli.command() decorator, Flask will also provide the context automatically.

In tests, you can use app.test_request_context() to simulate a request context.

Why does db.create_all() need an application context?

There are several reasons why db.create_all() needs an application context:

1. Database configuration: Database connection information is usually stored in the application configuration (such as app.config['SQLALCHEMY_DATABASE_URI']).

2. Multi-application support: If your project has multiple Flask applications and each uses a different database, the context ensures that the correct configuration is used.

3. Lazy initialization: Flask-SQLAlchemy uses lazy initialization, and some settings can only be completed within an application context.

Automatic handling in view functions

When handling web requests, Flask automatically creates and manages an application context for each request, so there is no need to explicitly use with app.app_context() inside view functions.

1
2
3
4

@app.route('/')
def index():
    # The application context already exists here
    return f"Current app: {current_app.name}"

CUDR

Flask-SQLAlchemy CUDR is very similar to SQLAlchemy ORM CUDR.

Insert

Instantiate the table class -> add -> commit

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# part code of import_articles_scripts
article = Article_Meta_Data(
            title=metadata.get('Title', 'Untitled'),
            author=metadata.get('Author', 'Unknown'),
            instructor=metadata.get('Instructor'),
            cover_image_url=metadata.get('CoverImage'),
            rollout_date=metadata.get('RolloutDate'),
            ultimate_modified_date=metadata.get('UltimateModifiedDate'),
            category=metadata.get('Category', 'Uncategorized'),
            brief_introduction=brief_intro_text
        )
db.session.add(article_metadata)
db.session.commit()

Query

The query statement stmt needs to be executed inside db.session.execute(). You only need to add db. before select, because db.select is essentially sqlalchemy.select.

For example, a complex query might look like this:

1
2
3
4
5
6

result = db.session.execute(
    db.select(User, Address)
    .join(Address)
    .where(User.name == "plain")
    .order_by(User.id)
)

Exploring the essence of db.select

From the source code of flask-sqlalchemy extension.py, we can see that the SQLAlchemy class does not directly define a select method. Instead, it handles undefined attribute access through the __getattr__ magic method. This mechanism is quite elegant. Let's see how it works:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import typing as t
import sqlalchemy as sa
import sqlalchmey.event as sa.event
import sqlalchemy.orm as sa.orm

class SQLALchemy:

    # ... igonre before code


    def __getattr__(self, name: str) -> t.Any:
    if name == "relation":
        return self._relation

    if name == "event":
        return sa_event

    if name.startswith("_"):
        raise AttributeError(name)

    # 关键在这里
    for mod in (sa, sa_orm):
        if hasattr(mod, name):
            return getattr(mod, name)

    raise AttributeError(name)

The workflow of this code is:

1. When we access db.select, Python first checks whether the SQLAlchemy class has this attribute
2. If it does not find it, it calls the __getattr__ method
3. Inside __getattr__, it looks for the attribute in the sa (SQLAlchemy) and sa_orm (SQLAlchemy ORM) modules in sequence
4. Since the select function exists in the sqlalchemy module, hasattr(sa, 'select') returns True
5. It then returns the original SQLAlchemy select function through getattr(sa, 'select')

This explains why:

1. db.select is actually sqlalchemy.select
2. You cannot find a direct definition of select in the source code
3. type(db.select()) displays <class 'sqlalchemy.sql.selectable.Select'>, which is SQLAlchemy's Select object

This design pattern is called the Proxy Pattern. In this way, Flask-SQLAlchemy proxies most of SQLAlchemy's functionality onto the db object, allowing us to access SQLAlchemy features directly through db.

sqlalchemy.select VS db.select

Since we now know that db.select is actually sqlalchemy.select, would directly using sqlalchemy.select be faster because it avoids loading through __getattr__()?

In practice, the execution speed of the two approaches is exactly the same. There is no performance difference at all.

1
2
3
4
5
6

# 方式1
from sqlalchemy import select
select(User)  # 直接调用sqlalchemy.select

# 方式2
db.select(User)  # 通过__getattr__获取后调用sqlalchemy.select

Although db.select needs to obtain the function through the __getattr__ magic method, this process:

• Only happens the first time db.select is accessed
• Later accesses directly use the already cached attribute
• This extra overhead is negligible compared with the actual query execution time

But db.select is still recommended because:

• All database-related operations go through the db object. If Flask-SQLAlchemy ever needs to extend or modify select in the future, your code does not need to change

• From a software engineering perspective, using db.select is better practice. This follows the principles of dependency injection and separation of concerns, and it also makes the code easier to maintain and test.

Delete & Update

First, use a query statement to select a specific entry, for example:

1
2
3
4

article = db.session.execute(
    db.select(Article_Meta_Data)
    .where(Article_Meta_Data.title == "test-article")
)

Delete

1
2

db.session.delete(article)
db.session.commit()

Update

1
2

article.author = Plain
db.session.commit()

Reference

Flask-SQLAlchemy — Flask-SQLAlchemy Documentation (3.1.x)