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

 一、业务功能简介

本项目主要是开发图书馆管理系统，所以设计的业务需求主要包含：

• 作者功能开发
• 出版社功能开发
• 书籍功能开发

当然要完成上述的功能，还需要开发一些辅助功能，比如数据库依赖项、数据库操作工具集等。

二、功能开发

（一）数据库依赖项开发

       在FastAPI框架中操作数据库必然需要先得到一个数据库连接，这里得到数据库连接的方法也比较简单，通过依赖项的方式来进行获取。

• 数据库会话配置 

       对数据库的增删改查会涉及到数据库的连接以及创建会话，所以在database.py文件中添加生成会话的代码：

...
from sqlalchemy.orm import sessionmaker
...
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
...

• 数据据依赖项开发 

      通过上面的SessionLocal来生成db实例用来连接数据库进行增删改查的操作，在dependencies.py文件中：

from database import SessionLocal


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

（二）作者功能开发

1、开发作者功能的API

• 根据作者用户名查找作者信息
• 创建一个作者信息
• 获取所有作者的信息

2、请求与响应模型设计 

设计请求响应模型对用户发送的请求体以及服务器返回的响应体进行校验，在schemas.py文件中：

from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import Optional


class AuthorBase(BaseModel):
    username: str

# 请求体模型校验
class AuthorCreate(AuthorBase):
    email: EmailStr

# 响应体模型校验
class Author(AuthorBase):
    id: int
    email: EmailStr

    class Config:
        orm_mode = True

3、工具类开发 

对用户提交的请求体进行操作，增删查改，在crud.py文件中：

from sqlalchemy.orm import Session
import models, schemas
from typing import List

# 通过作者名字获取具体的信息
def get_author_by_username(db: Session, username: str):
    return db.query(models.Author).filter(models.Author.username == username).first()

# 创建一个作者信息
def create_author(db: Session, author: schemas.AuthorCreate):
    db_author = models.Author(**author.dict())
    db.add(db_author)
    db.commit()
    db.refresh(db_author)
    return db_author

# 获取所有的作者信息
def get_all_author(db: Session):
    return db.query(models.Author).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("/author", response_model=schemas.Author)
def create_author(author: schemas.AuthorCreate, db: Session = Depends(get_db)):
    db_author = crud.get_author_by_username(db, author.username)
    if db_author:
        raise HTTPException(status_code=400, detail="author already exists!")
    return crud.create_author(db, author)

# 获取所有作者信息
@app.get("/authors", response_model=List[schemas.Author])
def get_all_authors(db: Session = Depends(get_db)):
    return crud.get_all_author(db)

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

启动后就可以直接访问了。

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