Python中CRUD增删改查教程

CRUD 是一个缩写，代表创建（Create）、读取（Read）、更新（Update）和删除（Delete），这四个操作是几乎所有应用程序的核心，它们允许开发者在应用程序中创建、检索、更新和删除数据。

以下是关于 CRUD 操作的一些关键点：

  1. 创建（Create）：向数据库添加新条目。
  2. 读取（Read）：检索或查看数据库中现有的条目。
  3. 更新（Update）：修改数据库中已有的数据详细信息。
  4. 删除（Delete）：从数据库中移除现有的条目。

CRUD 操作是应用程序功能的基础，它们影响应用程序如何存储、检索和管理数据的每一个方面。理解 CRUD 对于构建应用程序和理解你所使用的应用程序的后台工作至关重要。

CRUD 操作在 SQL 中几乎一一对应于 SQL 命令：

  • 创建 对应于 INSERT 命令。
  • 读取 对应于 SELECT 命令。
  • 更新 对应于 UPDATE 命令。
  • 删除 对应于 DELETE 命令。

使用 SQLAlchemy 执行 CRUD 操作
SQLAlchemy是 Python 的一个流行对象关系映射器 (ORM)。ORM 允许您使用 Python 对象与数据库交互，而不是像上一节中那样编写原始 SQL 查询。这让 Python 开发人员编写代码更加方便。

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, sessionmaker

class Base(DeclarativeBase):
    pass

class Bird(Base):
    __tablename__ = "bird"
    id = Column(Integer, primary_key=True)
    name = Column(String)

    def __repr__(self):
        return f"Bird(id={self.id}, name={self.name!r})"

engine = create_engine("sqlite:///birds.db")
Session = sessionmaker(bind=engine)

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

使用 SQLAlchemy 的一个重要优势是可以声明模型。在第 7 行中，您创建了一个 Bird 类，它既描述了鸟类数据库表，又描述了稍后将与表项交互的 Python 对象模型。

通过第 18 行的 init_db()，如果 birds.db 数据库不存在，则创建它；如果它存在，则连接它。这意味着您可以继续使用前一节中已有的 birds.db 数据库来研究 SQLAlchemy 的 CRUD 操作。

从您保存 crud_sql_alchemy.py 的同一文件夹中启动另一个 Python REPL 会话。然后，使用 SQLAlchemy 执行第一次 CRUD 操作：

>>> from crud_sql_alchemy import Session, Bird, init_db
>>> init_db()
>>> session = Session()
>>> new_bird = Bird(name="Test Bird")
>>> session.add(new_bird)
>>> session.commit()

通过 SQLAlchemy，您可以使用会话的 .add() 方法创建新的数据库条目。在上面的示例中，您将创建一个名称为 "Test Bird "的新条目。

要检查数据库中是否存在 "Test Bird"，可以查询数据库：

>>> from sqlalchemy import select
>>> query = select(Bird).where(Bird.name == "Test Bird")
>>> bird = session.execute(query).scalar_one()
>>> bird
<Bird(id=3, name='Test Bird')>

要使用 SQLAlchemy 检索数据库条目，需要使用 select() 和 .where() 方法。查询类似于原始 SQL 语句，但 Python 代码看起来更熟悉。在这里，您要选择名称等于 "Test Bird "的所有 Bird 对象。

返回将得到一个 Select 对象，并将其保存在查询变量中。

将 query 传递到 .execute() 时，就会执行查询。虽然数据库中只有一个 "Test Bird "符合您的查询，但您需要使用 .scalar_one() 将这个准确的数据库条目保存为 bird。

现在，您已经捕捉到了 "Test Bird"，并将对象保存到了 bird 变量中，您可以继续更新它的名称了：

>>> bird.name = "Example Bird"
>>> session.commit()
>>> session.execute(select(Bird)).scalars().all()
[<Bird(id=1, name='Hummingbird')>, <Bird(id=3, name='Example Bird')>]

在 SQLAlchemy 中，更新操作就像更改对象的属性然后提交更改一样简单。为了验证一切正常，您可以再次查询数据库并查看所有数据库条目。

在使用 SQLAlchemy 创建、读取和更新数据库后，您的 CRUD 例程中只缺少一个操作：删除条目。

只要您还在会话中，您就可以继续使用 bird：

>>> session.delete(bird)
>>> session.commit()
>>> session.close()

您可以通过调用会话的 .delete() 方法来准备使用 SQLAlchemy 删除数据库条目。同样，您需要调用 .commit() 来执行删除 CRUD 操作。

最后，重置会话是一种好的做法。您可以将 session.close() 视为您的善后工作。

使用 SQLAlchemy 可以通过添加一层抽象层来执行 CRUD 操作，从而降低编写原始 SQL 的复杂性。在使用网络应用程序时，在执行 CRUD 操作时还可以再增加一层抽象层。

接下来，您将探索 CRUD 操作在网络中的作用，然后构建自己的 CRUD 驱动的 REST API，使用 HTTP 请求方法和 SQLAlchemy 与数据库交互。

Http方法
CRUD 操作也与 HTTP 请求方法相对应：

  • 创建 对应于 POST 请求。
  • 读取 对应于 GET 请求。
  • 更新 可以使用 PUT 或 PATCH 请求。
  • 删除 对应于 DELETE 请求。

使用 FastAPI 执行 CRUD 操作
FastAPI是一个使用 Python 构建 API 的 Web 框架。继续将 FastAPI 安装到您之前创建的相同虚拟环境中，并在其中安装了 SQLAlchemy：

(venv) $ python -m pip install fastapi "uvicorn[standard]"

除了 FastAPI，您还可以使用上述命令安装Uvicorn。Uvicorn将是您运行 FastAPI 的服务器。

此外，FastAPI 安装Pydantic作为其依赖项。Pydantic 将帮助您对数据进行建模。

您将使用 FastAPI 创建要向其发送 HTTP 请求的端点。端点收到请求后，您将对数据库执行 CRUD 操作birds.db。

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import select
from sqlalchemy.orm import Session
from pydantic import BaseModel, ConfigDict

from crud_sql_alchemy import Bird, init_db
from crud_sql_alchemy import Session as SessionLocal

app = FastAPI()
init_db()

class BirdCreate(BaseModel):
    name: str

class BirdUpdate(BaseModel):
    name: str

class BirdResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

首先，导入 FastAPI 来创建 Web 应用程序并HTTPException处理潜在错误。使用PydanticBaseModel进行数据验证和设置管理。

在试用 API 之前，您需要定义 CRUD 操作的端点。继续编辑crud_fastapi.py并添加以下代码来定义您的第一个端点：

# ...

@app.post("/birds/", response_model=BirdResponse)
def create_bird(bird: BirdCreate, db: Session = Depends(get_db)):
    new_bird = Bird(name=bird.name)
    db.add(new_bird)
    db.commit()
    db.refresh(new_bird)
    return new_bird

读取当前存储在数据库中的所有鸟类的端点：

# ...

@app.get("/birds/", response_model=list[BirdResponse])
def read_birds(db: Session = Depends(get_db)):
    birds = db.execute(select(Bird)).scalars().all()
    return birds

read_birds()函数返回存储在bird数据库表中的所有鸟类。此端点响应GET请求，允许您检索所有鸟类的列表。

如果您只想接收特定的鸟，则需要在响应中告诉 API 您想要获取哪只鸟。为此，添加另一个接受鸟id作为参数的端点：

# ...

@app.get("/birds/{bird_id}", response_model=BirdResponse)
def read_bird(bird_id: int, db: Session = Depends(get_db)):
    query = select(Bird).where(Bird.id == bird_id)
    found_bird = db.execute(query).scalar_one()
    if found_bird is None:
        raise HTTPException(status_code=404, detail="Bird not found")
    return found_bird

当您将bird_id作为请求的端点 URL 的一部分传入时GET，该read_bird()函数会尝试返回具有给定 ID 的鸟。

上面的 SQLAlchemy 代码看起来与您之前在 Python REPL 中执行的步骤类似。

CRUD​​​​​​​更新操作
只有当具有请求 ID 的鸟存在时，您才会继续更改鸟的名称：

# ...

@app.put("/birds/{bird_id}", response_model=BirdResponse)
def update_bird(bird_id: int, bird: BirdUpdate, db: Session = Depends(get_db)):
    query = select(Bird).where(Bird.id == bird_id)
    found_bird = db.execute(query).scalar_one()
    if found_bird is None:
        raise HTTPException(status_code=404, detail="Bird not found")
    found_bird.name = bird.name
    db.commit()
    db.refresh(found_bird)
    return found_bird

删除.delete()：

# ...

@app.delete("/birds/{bird_id}", response_model=dict)
def delete_bird(bird_id: int, db: Session = Depends(get_db)):
    query = select(Bird).where(Bird.id == bird_id)
    found_bird = db.execute(query).scalar_one()
    if found_bird is None:
        raise HTTPException(status_code=404, detail="Bird not found")
    db.delete(found_bird)
    db.commit()
    return {"message": "Bird deleted successfully"}


现在，您的所有 CRUD API 端点都已存在，是时候试用您的 API 了！通过运行命令启动 Uvicorn 服务器uvicorn。作为参数，您需要提供不带.py扩展名的文件名，后跟冒号 ( :) 和 FastAPI 应用程序的名称：

(venv) $ uvicorn crud_fastapi:app
INFO:     Will watch for changes in these directories: [...]
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [29889] using WatchFiles
INFO:     Started server process [29891]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

FastAPI 的一个很酷的功能是它自带了开箱即用的交互式 API 文档。这意味着您可以直接在浏览器中测试 REST API 端点。访问http://127.0.0.1:8000/docs并在您的 REST API 上执行一些 CRUD 操作

REST API 中的 CRUD 操作为构建可通过 Web 创建、读取、更新和删除资源的 Web 服务提供了标准化框架。

总结
在本教程中，您探索了数据库和 REST API 中的 CRUD 操作。在用原始 SQL 编写 CRUD 操作后，您利用 SQLAlchemy 以更直观的方式与数据库交互。然后，您使用 FastAPI 创建 REST API 来帮助您了解 CRUD 操作如何连接到 HTTP 请求方法。

  • FastAPI 是一款出色的软件包，可用于将 CRUD 操作与 HTTP 请求方法连接起来。借助端点，您可以将 CRUD 操作映射到与数据库交互的函数。
  • 同样，SQLAlchemy 是一款出色的工具，可帮助您直观地在数据库上执行 CRUD 操作。