一、功能开发
紧接上篇的功能开发，然后开发出版社、书籍功能。
（一）出版社功能开发
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
更多文章合集前往：
往期精彩文章