FastApi SqlModel入门

环境准备

1. 安装和配置开发环境

我们需要以下工具和库：

Python 3.10+: 确保你已经安装了 Python。可以通过以下命令检查版本：
```
python --version
```
VSCode: 下载并安装最新版本的 VSCode。
PostgreSQL: 确保你已经安装并运行 PostgreSQL 数据库。你可以从 PostgreSQL 官网下载并安装。

安装 Python 扩展

打开 VSCode，点击左侧扩展图标，搜索并安装以下扩展：

Python: 提供 Python 开发支持。
Pylance: 提供智能代码补全。

2. 设置虚拟环境（推荐）

为了避免依赖冲突，我们建议使用虚拟环境。执行以下命令创建并激活虚拟环境：

python -m venv venv
source .venv/bin/activate  # Windows 用户使用 `.venv\Scripts\activate`

或者使用pyenv

3.安装依赖包

pip install -r requirements.txt -i https://mirrors.aliyun.com/pypi/simple/
或手动安装依赖
pip install "fastapi[standard]" -i https://mirrors.aliyun.com/pypi/simple/

pip install sqlmodel -i https://mirrors.aliyun.com/pypi/simple/

项目启动

1.初始化main文件

创建main.py:

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}

2.启动项目

fastapi dev main.py --reload

数据库配置

1. 配置 SQLite 数据库连接

在项目根目录下创建一个文件 database.py，并添加以下代码以配置 SQLite 数据库连接：

from sqlmodel import SQLModel, create_engine, Session

DATABASE_URL = "sqlite:///./data.db"
engine = create_engine(DATABASE_URL)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


# 依赖项，用于获取数据库会话
def get_session():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()

修改main.py：

from fastapi import FastAPI

from database import create_db_and_tables


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.get("/")
def read_root():
    return {"Hello": "World"}

开发CRUD

1.定义模型

定义作者和文章模型

在 models.py 中定义作者和文章的模型，并建立一对多的关系：

from typing import List, Optional
from sqlmodel import Field, Relationship

class Author(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    articles: List["Article"] = Relationship(back_populates="author")

class Article(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    author_id: Optional[int] = Field(default=None, foreign_key="author.id")
    author: Optional[Author] = Relationship(back_populates="articles")

2.增删改查接口实现

(1) 创建作者

@app.post("/authors/", response_model=Author)
def create_author(author: Author, session: Session = Depends(get_session)):
    """
    创建作者
    """
    session.add(author)
    session.commit()
    session.refresh(author)
    return author

(2) 获取所有作者

@app.get("/authors/", response_model=List[Author])
def read_authors(session: Session = Depends(get_session)):
    """
    获取所有作者
    """
    authors = session.query(Author).all()
    return authors

(3) 获取单个作者

@app.get("/authors/{author_id}", response_model=Author)
def read_author(author_id: int, session: Session = Depends(get_session)):
    """
    获取单个作者
    """
    author = session.get(Author, author_id)
    if not author:
        raise HTTPException(status_code=404, detail="未找到该作者")
    return author

(4) 更新作者

@app.put("/authors/{author_id}", response_model=Author)
def read_author(author_id: int, update_author: Author, session: Session = Depends(get_session)):
    """
    更新作者信息
    """
    author = session.get(Author, author_id)
    if not author:
        raise HTTPException(status_code=404, detail="未找到该作者")
    author.name = update_author.name
    author.email = update_author.email
    session.commit()
    session.refresh(author)
    return author

(5) 删除作者

@app.delete("/authors/{author_id}")
def delete_author(author_id: int, session: Session = Depends(get_session)):
    """
    删除作者
    """
    author = session.get(Author, author_id)
    if not author:
        raise HTTPException(status_code=404, detail="未找到该作者")
    session.delete(author)
    session.commit()
    return {"message": "删除成功"}

(6) 查询一个作者及其所有文章

@app.get("/authors/{author_id}/articles", response_model=AuthorWithArticles)
def read_author_with_articles(author_id: int, session: Session = Depends(get_session)):
    """
    查询一个作者及其所有文章
    """
    author = session.get(Author, author_id)
    if not author:
        raise HTTPException(status_code=404, detail="未找到该作者")

    articles = session.query(Article).filter(Article.author_id == author_id).all()
    return author

3.实现文章的增删改查接口

(1) 创建文章

@app.post("/articles/", response_model=Article)
def create_article(article: Article, session: Session = Depends(get_session)):
    """
    创建文章
    """
    session.add(article)
    session.commit()
    session.refresh(article)
    return article

(2) 获取所有文章

@app.get("/articles/", response_model=List[Author])
def read_articles(session: Session = Depends(get_session)):
    """
    获取所有文章
    """
    articles = session.query(Article).all()
    return articles

(3) 获取单个文章

@app.get("/articles/{article_id}", response_model=Article)
def read_article(article_id: int, session: Session = Depends(get_session)):
    """
    获取单个文章
    """
    article = session.get(Article, article_id)
    if not article:
        raise HTTPException(status_code=404, detail="未找到该文章")
    return article

(4) 更新文章

@app.put("/articles/{article_id}", response_model=Article)
def update_article(article_id: int, updated_article: Article, session: Session = Depends(get_session)):
    """
    更新文章
    """
    article = session.get(Article, article_id)
    if not article:
        raise HTTPException(status_code=404, detail="未找到该文章")
    article.title = updated_article.title
    article.content = updated_article.content
    session.commit()
    session.refresh(article)
    return article

(5) 删除文章

@app.delete("/articles/{article_id}")
def delete_article(article_id: int, session: Session = Depends(get_session)):
    """
    删除文章
    """
    article = session.get(Article, article_id)
    if not article:
        raise HTTPException(status_code=404, detail="未找到该文章")
    session.delete(article)
    session.commit()
    return {"message": "删除成功"}

4.启动应用

在终端中运行以下命令启动 FastAPI 应用：

fastapi dev main.py --reload

访问并调试:http://127.0.0.1:8000/docs

环境准备#

1. 安装和配置开发环境#

安装 Python 扩展#

2. 设置虚拟环境（推荐）#

3.安装依赖包#

项目启动#

1.初始化main文件#

2.启动项目#

数据库配置#

1. 配置 SQLite 数据库连接#

开发CRUD#

1.定义模型#

2.增删改查接口实现#

3.实现文章的增删改查接口#

4.启动应用#

知识学习#