开始免费使用
当前位置：首页 > Python > Python后端开发必学：数据库操作与RESTful API设计

Python后端开发必学：数据库操作与RESTful API设计

青禾
青禾
发布时间：2026-10-24| 9人浏览过
原创

构建稳定、可维护的 Python 后端服务，核心逻辑在于两项基础能力的配合：一是准确管理业务数据在数据库中的生命周期，二是通过结构清晰、遵循约定的 RESTful API 将数据安全地交付给客户端。本文以 Python 主流开发栈（FastAPI 与 SQLAlchemy）为基础，直接从分层架构、API 规范设计、数据库连接及端到端接口编写展开讲解，提供可落地的代码和操作步骤。

一、后端分层架构与数据流转原理

在编写业务逻辑之前，必须理清客户端请求在后端系统中的流转边界。清晰的分层能够避免把 SQL 查询或数据校验混杂在路由处理函数中，使系统易于扩展和维护。

典型的 Python 后端架构由四层组成：客户端协议交互层、API 路由控制层、业务与数据访问层（ORM / DAO），以及底层物理数据库。客户端发起的 HTTP 请求首先到达路由控制层，由校验器（如 Pydantic）完成入参格式校验；随后调用数据访问层通过连接池执行结构化查询；最后把处理后的领域对象序列化为标准 JSON 格式并返回给调用方。

Python后端开发必学：数据库操作与RESTful API设计
图 1：Python 后端分层架构与数据全生命周期流转图

二、RESTful API 设计的核心规范与约定

RESTful 并非具体框架，而是一种基于 HTTP 协议特性的接口设计风格。遵循 RESTful 规范的核心目标，是让 API 具备自解释性、一致性和可预测性。

1. 资源路径规范

REST 架构把所有可访问的实体视为“资源”（Resource）。设计路径时必须坚持使用名词复数，避免在 URI 中包含动词。操作的具体意图完全由 HTTP 请求方法表达。

  • 推荐形式：/api/v1/articles（资源集合）、/api/v1/articles/42（具体单条资源）。
  • 反面做法：/api/v1/getArticles、/api/v1/createArticle、/api/v1/article/delete?id=42。这种把动作写入 URL 的做法破坏了 HTTP 语义。
  • 层级关系表达：若某资源依附于上级资源，应采用层级嵌套形式，例如 /api/v1/articles/42/comments 表示获取第 42 篇博客下的所有评论。

2. HTTP 动词与状态码的精准映射

不同的客户端行为应当对应明确的 HTTP 动词，并且服务器在响应时必须返回语义相符的状态码，而非一律返回 200 OK 并用自定义 code 掩盖错误。

Python后端开发必学：数据库操作与RESTful API设计
图 2：RESTful 规范中的方法意图、幂等性与标准状态码对照

三、数据库操作基础：连接、会话与 ORM 模型

在 Python 后端开发中，原生 SQL（如直接调用 sqlite3 或 psycopg2）在复杂项目中容易带来 SQL 注入隐患和代码维护困难。采用对象关系映射（ORM，如 SQLAlchemy 2.0+）能够将 Python 类映射为数据库表，将实例对象映射为行记录。

1. 依赖安装与工程准备

首先准备必要的 Python 依赖包。在虚拟环境中执行以下安装命令：

pip install fastapi uvicorn sqlalchemy pydantic

2. 建立引擎与会话生命周期

数据库操作的核心是管理 Engine（连接池与底层驱动封装）与 Session（数据库事务单元）。必须确保每个 HTTP 请求拥有独立的数据库会话，并在请求结束时必定关闭连接。

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# 此处使用 SQLite 演示，生产环境可替换为 postgresql://user:pwd@host/dbname
SQLALCHEMY_DATABASE_URL = "sqlite:///./app_production.db"

# connect_args={"check_same_thread": False} 仅用于 SQLite
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False}
)

# 事务会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 模型基类
Base = declarative_base()

# FastAPI 依赖注入生成器：确保每个请求独立开启并在响应后释放连接
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. 声明式数据模型（ORM）

在 models.py 中定义与数据库表结构绑定的 Python 类：

# models.py
from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime
from datetime import datetime
from database import Base

class Article(Base):
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String(150), nullable=False, index=True)
    content = Column(Text, nullable=False)
    is_published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)

四、端到端接口实现全流程

实现一个完整的 RESTful API 端点包含四个明确阶段：定义数据传输契约（Schema）、封装底层数据库查询操作（CRUD）、挂载路由并进行参数依赖注入、以及错误统一处理。

Python后端开发必学：数据库操作与RESTful API设计
图 3：API 与数据库交互的四步开发落地步骤图

步骤一：定义输入与输出模式（Schemas）

ORM 模型用于跟数据库交互，而 Pydantic 模式用于约束 HTTP 请求体与响应体。将两者解耦可以严格保护数据安全（例如防止将密码哈希或内部敏感标识输出给前端）：

# schemas.py
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import Optional

# 客户端创建文章时的入参结构
class ArticleCreate(BaseModel):
    title: str
    content: str
    is_published: bool = False

# 服务端响应给客户端时的结构
class ArticleResponse(BaseModel):
    id: int
    title: str
    content: str
    is_published: bool
    created_at: datetime

    # 启用属性映射以直接读取 SQLAlchemy ORM 对象
    model_config = ConfigDict(from_attributes=True)

步骤二：编写 CRUD 数据库持久化层

在独立的 crud.py 模块中编写纯粹的数据库操作逻辑，绝不直接接触 HTTP Request 对象：

# crud.py
from sqlalchemy.orm import Session
from models import Article
from schemas import ArticleCreate
from typing import List, Optional

def get_article_by_id(db: Session, article_id: int) -> Optional[Article]:
    return db.query(Article).filter(Article.id == article_id).first()

def get_articles(db: Session, skip: int = 0, limit: int = 20) -> List[Article]:
    return db.query(Article).offset(skip).limit(limit).all()

def create_article(db: Session, article_in: ArticleCreate) -> Article:
    db_obj = Article(
        title=article_in.title,
        content=article_in.content,
        is_published=article_in.is_published
    )
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)  # 刷新以加载数据库自增的主键 id 和默认字段
    return db_obj

def delete_article(db: Session, article_id: int) -> bool:
    db_obj = db.query(Article).filter(Article.id == article_id).first()
    if not db_obj:
        return False
    db.delete(db_obj)
    db.commit()
    return True

步骤三：装配 RESTful 路由控制器

在 main.py 中绑定路由、注入数据库会话，并处理资源不存在时的 404 状态码：

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from database import engine, Base, get_db
import models, schemas, crud

# 在应用启动时创建表结构（生产环境推荐配合 Alembic 进行迁移）
Base.metadata.create_all(bind=engine)

app = FastAPI(title="文章管理系统 API", version="1.0.0")

# 1. 列表获取 (GET /articles)
@app.get("/articles", response_model=List[schemas.ArticleResponse], status_code=status.HTTP_200_OK)
def list_articles(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    return crud.get_articles(db, skip=skip, limit=limit)

# 2. 单条读取 (GET /articles/{id})
@app.get("/articles/{article_id}", response_model=schemas.ArticleResponse, status_code=status.HTTP_200_OK)
def read_article(article_id: int, db: Session = Depends(get_db)):
    article = crud.get_article_by_id(db, article_id)
    if article is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"文章 ID {article_id} 未找到"
        )
    return article

# 3. 新建资源 (POST /articles)
@app.post("/articles", response_model=schemas.ArticleResponse, status_code=status.HTTP_201_CREATED)
def create_new_article(article_in: schemas.ArticleCreate, db: Session = Depends(get_db)):
    return crud.create_article(db, article_in)

# 4. 删除资源 (DELETE /articles/{id})
@app.delete("/articles/{article_id}", status_code=status.HTTP_204_NO_CONTENT)
def remove_article(article_id: int, db: Session = Depends(get_db)):
    success = crud.delete_article(db, article_id)
    if not success:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"文章 ID {article_id} 未找到，无法执行删除"
        )
    return None

五、生产环境核心避坑指南与质量核查

在实际生产部署中，不规范的数据库操作和粗糙的 API 设计往往会导致性能雪崩或难以排查的并发隐患。以下总结了开发中必须落地的四项检查清单：

重点防线一：避免数据库连接耗尽

绝不要在路由函数或后台任务中直接调用 SessionLocal() 而忘记在 finally 块中 close()。推荐始终依赖框架的依赖注入机制（如上述 get_db），确保即使代码遇到未捕获异常，连接也能如期归还连接池。

重点防线二：控制分页查询与避免全表扫描

凡是列表查询端点，严禁返回无限制的 db.query(Model).all()。必须在参数中强制限制 limit 的最大边界（例如 limit: int = Query(20, le=100)），并对高频过滤条件（如 created_at、user_id）在数据库端建立索引。

Python后端开发必学：数据库操作与RESTful API设计
图 4：数据库操作与 RESTful API 生产级质量双向核查清单

六、总结

掌握数据库持久化与 RESTful API 设计是 Python 后端开发的核心基本功。通过 SQLAlchemy 完成物理数据表到领域对象的建模，借由 FastAPI 与 Pydantic 确立严谨的接口输入输出边界，再依托 HTTP 标准动词与状态码建立清晰契约，能够显著降低前后端联调成本，并为系统的水平扩展与微服务演进打下坚实的基础。

相关文章

相关标签: Python 数据库操作 FastAPI SQLAlchemy ORM

免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。

作者最新文章

全部文章

Python 相关产品和应用

全部下载
PythonSelenium
Selenium

Selenium

4.51.0 · Windows

用于浏览器自动化和 Web 测试的 Python 工具。

Python机器学习
scikit-learn

scikit-learn

1.9.1 · Windows

常用 Python 机器学习算法与模型评估工具库。

Python代码质量
Black

Black

26.10.1 · Windows

强调一致性的 Python 自动代码格式化工具。

Python代码质量
Ruff

Ruff

0.17.1 · macOS

由 Rust 编写的高速 Python 代码检查和格式化工具。

Pythonpytest
pytest

pytest

9.1.1 · Windows

简洁而强大的 Python 自动化测试框架。

PythonFastAPI
FastAPI

FastAPI

0.146.0 · Windows

高性能 Python API 框架，支持类型提示和自动接口文档。

PythonFlask
Flask

Flask

3.1.3 · Windows

轻量灵活的 Python Web 应用框架。

PythonDjango
Django

Django

6.1.2 · Windows

成熟的 Python Web 框架，适合构建内容站点和业务系统。

PythonIDE
Spyder

Spyder

6.1.7 · Windows

面向科学计算与数据分析的 Python 集成开发环境。

PythonSciPy
SciPy

SciPy

1.18.1 · Windows

面向科学、工程和技术计算的 Python 算法库。