FastAPI后端篇之业务功能开发（二）

-

 一、功能开发

紧接上篇的功能开发，然后开发出版社、书籍功能。

（一）出版社功能开发

1、开发出版社功能的API

  • 根据出版社名称查找出版社信息
  • 创建一个出版社信息
  • 获取所有出版社信息

2、请求与响应模型设计

       设计请求响应模型对用户发送的请求体以及服务器返回的响应体进行校验，在schemas.py文件中：
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional


class PublishBase(BaseModel):
    name: str

# 请求体模型校验
class PublishCreate(PublishBase):
    pass

# 响应体模型校验
class Publish(PublishBase):
    id: int

    class Config:
        orm_mode = True

3、工具类开发 

对用户提交的请求体进行操作，增删查改，在crud.py文件中：
from sqlalchemy.orm import Session
import models, schemas
from typing import List

# 通过出版社名字获取具体的信息
def get_publish_by_name(db: Session, name: str):
    return db.query(models.Publish).filter(models.Publish.name == name).first()

# 创建一个出版社信息
def create_publish(db: Session, publish: schemas.PublishCreate):
    db_publish = models.Publish(**publish.dict())
    db.add(db_publish)
    db.commit()
    db.refresh(db_publish)
    return db_publish

# 获取所有的出版社信息
def get_all_publish(db: Session):
    return db.query(models.Publish).all()

4、创建视图函数

在main.py文件中： 
from fastapi import FastAPI, Depends, HTTPException
import uvicorn
import schemas, crud
from sqlalchemy.orm import Session
from dependencies import get_db
from typing import List

app = FastAPI()

# 创建一个出版社信息
@app.post("/publisher", response_model=schemas.Publish)
def create_publish(publish: schemas.PublishCreate, db: Session = Depends(get_db)):
    db_publish = crud.get_publish_by_name(db, publish.name)
    if db_publish:
        raise HTTPException(status_code=400, detail="publish already exists!")
    return crud.create_publish(db, publish)

# 获取所有出版社信息
@app.get("/publishers", response_model=List[schemas.Publish])
def get_all_publishs(db: Session = Depends(get_db)):
    return crud.get_all_publish(db)


if __name__ == '__main__':
    uvicorn.run(app=app, host="127.0.0.1", port=8080)
启动后就可以直接访问了。

（二）书籍功能开发

1、开发书籍功能的API 

  • 根据书籍名称查找出书籍信息 
  • 创建一本书籍信息 
  • 获取所有书籍信息 

2、请求与响应模型设计 

设计请求响应模型对用户发送的请求体以及服务器返回的响应体进行校验，在schemas.py文件中： 
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional


class BookBase(BaseModel):
    title: str
    price: float
    publish_date: Optional[datetime] = None

# 创建一个书籍信息
class BookCreate(BookBase):
    pass

# 获取所有书籍信息
class Book(BookBase):
    author_id: int

    class Config:
        orm_mode = True

3、工具类开发 

对用户提交的请求体进行操作，增删查改，在crud.py文件中： 
from sqlalchemy.orm import Session
import models, schemas
from typing import List

# 根据作者id、出版社id列表来创建书籍信息，包括外键以及多对多关系表
def create_book_by_author(db: Session, book: schemas.BookCreate, author_id: int, publish_id_list: List[int]):
    db_book = models.Book(**book.dict(), author_id=author_id)
    # 根据publish_id_list生成对象列表
    publish_obj_list = [db.query(models.Publish).filter(models.Publish.id == i).first() for i in publish_id_list]
    db_book.book_to_publish = publish_obj_list
    db.add(db_book)
    db.commit()
    db.refresh(db_book)
    return db_book

# 根据书籍名称查找书籍
def get_book_by_title(db: Session, title: str):
    return db.query(models.Book).filter(models.Book.title == title).first()

# 获取所有的书籍信息
def get_all_books(db: Session):
    books = db.query(models.Book).all()
    result = []
    for obj in books:
        result.append({
            "id": obj.id,
            "title": obj.title,
            "price": obj.price,
            "publish_date": obj.publish_date,
            "authors": obj.book_to_author,
            "publishs": obj.book_to_publish
        })

    return result

4、创建视图函数 

在main.py文件中： 
from fastapi import FastAPI, Depends, HTTPException
import uvicorn
import schemas, crud
from sqlalchemy.orm import Session
from dependencies import get_db
from typing import List

app = FastAPI()

# 根据作者以及出版署参数创建书籍信息
@app.post("/book/{author_id}", response_model=schemas.Book)
def create_book(author_id: int, publish_id_list: List[int], book: schemas.BookCreate, db: Session = Depends(get_db)):
    db_book = crud.get_book_by_title(db=db, title=book.title)
    if db_book:
        raise HTTPException(status_code=400, detail="book already exists!")
    return crud.create_book_by_author(db=db, book=book, author_id=author_id, publish_id_list=publish_id_list)

# 获取所有的书籍信息
@app.get("/books")
def get_all_books(db: Session = Depends(get_db)):
    return crud.get_all_books(db)



if __name__ == '__main__':
    uvicorn.run(app=app, host="127.0.0.1", port=8080)
启动后就可以直接访问了。

二、自定义响应模型 

       在之前的响应中，返回的就是数据库中的数据，没有带一些返回给前台的信息，所以响应信息不完善，那么正常的是返回一些信息然后加上数据。 

1、自定义Pydantic模型 

在schemas.py文件中新建三个模型：
# 书籍响应体单独定义
class BookAll(BaseModel):
    id: int
    title: str
    price: float
    publish_date: datetime
    authors: Author
    publishs: List[Publish]


# 自定义响应模型，获取所有的数据
class GeneralResDefine(BaseModel):
    message: str
    code: int
    data: List[Union[Author, Publish, BookAll]]


# 创建成功数据模型，返回创建成功的那个对象
class GeneralDefine(BaseModel):
    message: str
    code: int
    obj: Union[Author, Publish, BookAll]

2、视图函数 

 在main.py文件中：
  from fastapi import FastAPI, Depends, HTTPException
import uvicorn
import schemas, crud
from sqlalchemy.orm import Session
from dependencies import get_db
from typing import List, Union
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

# import models
# from database import engine
#
# models.Base.metadata.create_all(bind=engine)

app = FastAPI()


@app.get("/")
async def home():
    return {"username": "zhangsan"}


@app.post("/author", response_model=Union[schemas.Author, schemas.GeneralDefine])
def create_author(author: schemas.AuthorCreate, db: Session = Depends(get_db)):
    res = {
        "message": "创建成功",
        "code": 200,
        "obj": {}
    }
    db_author = crud.get_author_by_username(db, author.username)
    if db_author:
        raise HTTPException(status_code=400, detail="author already exists!")
    try:
        obj = crud.create_author(db, author)
        res["obj"] = obj
    except Exception as e:
        res["message"] = "创建失败"
        res["code"] = 208
    return res


@app.get("/authors", response_model=Union[List[schemas.Author], schemas.GeneralResDefine])
def get_all_authors(db: Session = Depends(get_db)):
    res = {
        "message": "获取成功",
        "code": 200,
        "data": []
    }
    try:
        result = crud.get_all_author(db)
        res["data"] = result
    except:
        res["message"] = "获取失败"
        res["code"] = 208
    return res


@app.post("/publisher", response_model=Union[schemas.Publish, schemas.GeneralDefine])
def create_publish(publish: schemas.PublishCreate, db: Session = Depends(get_db)):
    res = {
        "message": "创建成功",
        "code": 200,
        "obj": {}
    }
    db_publish = crud.get_publish_by_name(db, publish.name)
    if db_publish:
        raise HTTPException(status_code=400, detail="publish already exists!")
    try:
        obj = crud.create_publish(db, publish)
        res["obj"] = obj
    except Exception as e:
        res["message"] = "创建失败"
        res["code"] = 208
    return res


@app.get("/publishers", response_model=Union[List[schemas.Publish], schemas.GeneralResDefine])
def get_all_publishs(db: Session = Depends(get_db)):
    res = {
        "message": "获取成功",
        "code": 200,
        "data": []
    }
    try:
        result = crud.get_all_publish(db)
        res["data"] = result
    except:
        res["message"] = "获取失败"
        res["code"] = 208
    return res


@app.post("/book/{author_id}", response_model=Union[schemas.Book, schemas.GeneralDefine])
def create_book(author_id: int, publish_id_list: List[int], book: schemas.BookCreate, db: Session = Depends(get_db)):
    res = {
        "message": "创建成功",
        "code": 200,
        "obj": {}
    }
    db_book = crud.get_book_by_title(db=db, title=book.title)
    if db_book:
        raise HTTPException(status_code=400, detail="book already exists!")
    try:
        obj = crud.create_book_by_author(db=db, book=book, author_id=author_id, publish_id_list=publish_id_list)
        res["obj"] = obj
    except Exception as e:
        res["message"] = "创建失败"
        res["code"] = 208
    return res


@app.get("/books", response_model=schemas.GeneralResDefine)
def get_all_books(db: Session = Depends(get_db)):
    res = {
        "message": "获取成功",
        "code": 200,
        "data": []
    }
    try:
        result = crud.get_all_books(db)
        res["data"] = result
    except:
        res["message"] = "获取失败"
        res["code"] = 208
    return res


if __name__ == '__main__':
    uvicorn.run(app=app, host="127.0.0.1", port=8010)

3、创建书籍工具类 

在创建书籍函数中之前没有返回与之相关的出版社信息，这里进行返回出版社信息，在crud.py文件中： 
def create_book_by_author(db: Session, book: schemas.BookCreate, author_id: int, publish_id_list: List[int]):
    db_book = models.Book(**book.dict(), author_id=author_id)
    # 根据publish_id_list生成对象列表
    publish_obj_list = [db.query(models.Publish).filter(models.Publish.id == i).first() for i in publish_id_list]
    db_book.book_to_publish = publish_obj_list
    db.add(db_book)
    db.commit()
    db.refresh(db_book)
    obj = {
        "id": db_book.id,
        "title": db_book.title,
        "price": db_book.price,
        "publish_date": db_book.publish_date,
        "authors": db_book.book_to_author,
        "publishs": db_book.book_to_publish
    }
    return obj


更多文章合集前往：往期精彩文章

此博客中的热门博文

FastAPI后端篇之项目搭建

-
 一、环境搭建         在之前的博文中 Python本地虚拟环境搭建及Pycharm配置 讲述过虚拟环境搭建的过程，这里不再赘述，直接使用即可。 查看虚拟环境 conda info --env 创建虚拟环境 conda create -n FullStackEnv python=3.8.5 激活虚拟环境  conda activate FullStackEnv pip list 安装FastAPI pip install fastapi==0.65.1 二、创建后台项目 新建full-stack-backend项目 新建目录full-stack-backend，然后使用IDE打开即可：         接着需要在该目录中新建一些目录，不同的目录有不同的功能，此时需要新建alembic和app文件夹，alemic文件夹主要用于数据库的迁移版本管理，类似于Django中migrations，而app文件夹主要就是业务功能开发的内容，因为后台项目没有那么复杂，所以目录结构无需设计的过于复杂。 然后在各个目录中新建对应的.py文件： ├─FullStackEnv_1 │ │ alembic.ini 数据库地址配置 │ │ requirements.txt 依赖包导出 │ │ __init__.py │ │ │ ├─alembic │ │ │ env.py 与ORM进行关联配置 │ │ ├─app │ crud.py 数据库操作相关 │ database.py sqlalchemy相关配置 │ dependencies.py FastAPI框架依赖项 │ main.py │ models.py 数据库模型表 │ schemas.py Pydantic数据校验模型 │ __init__.py 项目的大概架子搭好了，接着就是将虚拟环境配置到该项目上。 安装依赖 将项目中需要的依赖包进行安装，也就是requirements.txt文件中的依赖包： pip install -r
Read more »

玩转虚拟机系列之如何高效创建虚拟机

-
一、如何高效创建虚拟机 在前两期的视频中，我们介绍了虚拟机是如何搭建的，同时大家也发现了通过系统镜像创建虚拟机带来的不便，各种配置参数的填写以及网络配置。那么当我们创建好一个虚拟机后是否可以反复利用，无需重复创建的过程，答案当然是可以。 当我们创建一个虚拟机后，如果以后还想多次复用，切记在创建一个虚拟机最原始的时候进行克隆，保持原有虚拟机的最少功能以及体积最小的状态。 有两种方式可以解决这个问题： 通过VMware克隆功能 通过配置文件互相拷贝 我们分别来说明上述两种方式。 二、VMware克隆功能 通过菜单栏上的 虚拟机--》管理--》克隆功能实现： 点击下一页： 克隆虚拟机的当前状态： 创建完整克隆： 然后给新克隆的虚拟机指定名称以及安装位置： 最后完成克隆即可。 虽然完成了克隆虚拟机，但是新的虚拟机种的ip与之前的虚拟机一样，所以需要修改ip配置。启动新的虚拟机： 我们使用被克隆的虚拟机的账号登陆，然后进入终端: vi /etc//sysconfig/network-scripts/ifcfg-ens33 TYPE="Ethernet" BOOTPROTO="static" DEFROUTE="yes" PEERDNS="yes" PEERROUTES="yes" IPV4_FAILURE_FATAL="no" IPV6INIT="yes" IPV6_AUTOCONF="yes" IPV6_DEFROUTE="yes" IPV6_PEERDNS="yes" IPV6_PEERROUTES="yes" IPV6_FAILURE_FATAL="no" IPV6_ADDR_GEN_MODE="stable-privacy" NAME="ens33" UUID="5a09ae90-c928-4a18-a3ba-9b5b4b0e9ae1" DEVICE="ens33" ONBOOT="yes" IPADDR=192.168.111.5
Read more »

玩转虚拟机系列之如何搭建虚拟机

-
一、虚拟机的作用 进行一些软件的测试 安装不同的操作系统，比如Linux系列的系统         对于一些未知软件是否可用，没有把握性，不妨到虚拟机中进行测试，这样不会对宿主机造成影响。 而Linux系列的操作系统，比如：Centos、Ubuntu等操作系统 二、虚拟机的安装         如果要安装不同系统的镜像，那么我们需要为这些系统找一个壳子，比较好用的有VMware等，我们首先去官网下载： https://www.vmware.com/         这里我是用的是VMware Workstation Pro，你也可以下载VMware Workstation Player进行使用。 下载好的已经上传到网盘： OneDrive https://iveboy-my.sharepoint.com/:f:/g/personal/shenjianping0307_iveboy_onmicrosoft_com/EmIJNgpGLc5OrYlnAEtIb-oBw6Pfiwy8Pt1oPwQalf_ccA 百度网盘 链接： https://pan.baidu.com/s/1srAU-Y6nX42ijhwWQKeKcQ?pwd=zwis   提取码：zwis          首先新建一个目录用于存放安装的镜像目录，比如在D盘下新建一个Virtual Workstation Machines目录，然后在其中新建一个base目录： 然后我们点击软件进行安装： 然后下一步： 安装位置可以自己更改，然后下一步： 可以将1、2步中的勾选去掉，然后直接下一步： 直接下一步： 直接安装，等待安装完成即可。 这里点击许可证，输入密匙，网上分享的有很多，比如下面3个： ZF3R0-FHED2-M80TY-8QYGC-NPKYF YF390-0HF8P-M81RQ-2DXQE-M2UT6 ZF71R-DMX85-08DQY-8YMNC-PPHV8 输入其中任何一个： 点击输入即可完成。 三、安装系统 打开VMware然后创建新的虚拟机： 按照下图选择自定义： 点击下一步： 然后选择系统镜像： 填入对应的信息： 这里填入信息没有固定的，根据自己的情况填写即可。 修改虚拟机名称以及新建系统文件的存放位置。 根据自己电脑配置来填写合适的数量。 修改内存，根据电脑配置来进行选择。 网络类型就选择
Read more »