使用fastapi完成一个车辆违章查询

FastAPI框架是一个高性能，易于学习，高效编码，生产可用的python，web开发框架。

但对于刚开始学习python不久的同学来说仅仅从官文学习，有一定的难度，并咩有那么易学。

本篇主要从一个车辆违章查询的项目，以项目来驱动fastapi的学习。

项目实现效果如下：
接口swagger ui界面：

查询页面展示：

fastapi的安装和简单案例

fastapi框架的安装

pip install fastapi

此外还需要一个能让项目跑起来的服务器，并且需要的是一个ASGI服务器

pip install uvicorn

最简单的fastapi只需要一个文件

创建main.py：这里使用官方代码：

from typing import Optional

from fastapi import FastAPI

app = FastAPI()


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


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}

保存好后在命令行运行服务器：

uvicorn main:app --reload

访问路径 http://127.0.0.1:8000/docs/，就能看到你的第一个接口的接口文档，如下

车辆查询项目的项目结构

本人也是对fastapi只是入门，对结构的理解暂时如下，该结构并不是一个完整web项目的需求。以后有机会再深入研究。

一一介绍每个部分：

• curd ：数据库的操作，使用sqlalchmey完成，curd即是创建（create），更新（update），读取（read），删除（delete）。
• database：完成对数据库的连接。
• main：路由函数的位置，可以和flask类比。
• models：对数据库表结构类化，和curd，database部分的实现主要参考sqlalchmey。
• pagnation：自定义的分页器，完成查询后数据的分页。
• schemas：规范约束，响应的或者查询到数据。

完成数据库连接和模型构建部分

这里使用的是mysql，并且使用mysqlclient库，如果使用的pymysql可参见第二种写法，这里创建了database.py文件。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
# 使用mysqlclient的情况
SQLALCHEMY_DATABASE_URL = "mysql://username:password@localhost:3306/数据库名?charset=utf8"
# 使用pymysql
# SQLALCHEMY_DATABASE_URL = "mysql+pymysql://username:password@localhost:3306/数据库名?charset=utf8"
from sqlalchemy.orm import sessionmaker

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

创建models.py，该项目中主要使用了record和car两个表，record关联到car的id上。

实现如下：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Date, MetaData
from sqlalchemy.orm import relationship

from database import Base, engine

metadata = MetaData(engine)


class Car(Base):
    __tablename__ = "car"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    carno = Column(String(15), index=True, )
    owner = Column(String(20), index=True, )
    brand = Column(String(20))
# 这里的配置存在疑惑，也是初次使用sqlalchemy。
    records = relationship('Record', back_populates='car')


class Record(Base):
    __tablename__ = "record"

    id = Column(Integer, primary_key=True, autoincrement=True)
    reason = Column(String(255))
    makedate = Column(Date)
    punish = Column(String(255))
    dealt = Column(Boolean, default=0)
    car_id = Column(Integer, ForeignKey('tb_car.id'))

    car = relationship("Car", back_populates="records")

curd操作

在该案例中对数据库主要有查询，更改和删除。查询条件包含时间范围，和对输入框的内容在车牌和车主姓名里进行查询。更改操作，完成对数据受理信息的更改，删除操作只是删除record记录，car表的信息不做改变。（在实际案例中，删除操作是对删除信息字段进行改变）。

下面代码中实现了五个接口：

from typing import List, Any

from pydantic.schema import date
from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

import pagnation
from models import Car, Record

PAGE_SIZE = 5

# 查询所有车辆信息
def read_car(db: Session):
    """

    :param db:
    :return:
    """
    cars: List[Any] = db.query(Car).all()
    return cars

# 查询所有记录信息，不包含车辆信息
def read_search_record(db: Session):
    """
    :param db:
    :return:
    """
    return db.query(Record).all()

# 按照条件进行查询
def read_all_record(db: Session, carno: str = '', start: date = None, end: date = None, page: int = 1):
    """
    :param db:
    :param carno:
    :param start:
    :param end:
    :param page:
    :return:
    """
    url = "/api/record/?"
    records = db.query(Record)
    if carno:
        records = records.join(Car).filter(or_(
            Car.carno.like("%" + carno + "%"),
            Car.owner.like("%" + carno + "%")))
        url += f"carno={carno}&"
    if all([start, end]):
        records = records.filter(and_(Record.makedate <= end, Record.makedate >= start))
        url += f"start={start}&end={end}"
    count = (len(records.all()) + PAGE_SIZE - 1) / PAGE_SIZE
    if page:
        records = records.limit(PAGE_SIZE).offset((page - 1) * PAGE_SIZE)
    records = records.all()
    return pagnation.pagenation(records, url, page, count)

# 更新record中的 dealt字段为True
def update_dealt(db: Session, record_id: int):
    """
    :param db:
    :param record_id:
    :return:
    """
    try:
        record = db.query(Record).get(record_id)
        record.dealt = True
        db.add(record)
        db.commit()
        return {'code': 10000, 'msg': "Success update car's record "}
    except Exception as e:
        print(e)
    return {'code': 10001, 'msg': " Fail update car's record "}

# 删除一条已受理的record记录
def del_record(db: Session, record_id: int):
    """
    :param db:
    :param record_id:
    :return:
    """
    try:
        record = db.query(Record).get(record_id)
        if record.dealt:
            db.delete(record)
            db.commit()
            return {'code': 10000, 'msg': " Success del car's record "}
    except Exception as e:
        print(e)
    return {'code': 10001, 'msg': " Fail del car's record "}

curd操作完成后，必不可少，会考虑到分页的需求。实现思路主要是，路由参数参数传递中添加page字段，传出参数中需要以下5个字段：

• count：总共页数
• currpage：当前页码
• nexturl：下一页接口
• preurl：前一页接口
• record：查询的记录数据

分页器实现代码

def pagenation(record, url, currage, count: int):
    nextnum = int(count // 1) if (currage + 1) >= count else currage + 1
    prenum = 1 if (currage - 1) == 0 else currage - 1
    nexturl = url + f"page={nextnum}"
    preurl = url + f"page={prenum}"
    if currage == 0:
        currage = 1
    if currage >= count:
        currage = count
    latest = {
        'count': count,
        'currpage': currage,
        'nexturl': nexturl,
        'preurl': preurl,
        "records": record,
    }
    return latest

schemas模块的实现

这个对于初学着看着有点摸不着头脑。这里简单描述：该部分完成的是对于路由函数（main.py里的函数）的接受和返回的数据进行规范。

对下面代码进行简单描述：

• CarBase包含了从car中需要查询到的字段，并规定了每个字段的类型
• Car在CarBase基础上添加了orm_mode,可以实现orm模型

from typing import List

from pydantic import BaseModel
from pydantic.schema import date


class CarBase(BaseModel):
    carno: str
    owner: str


class Car(CarBase):
    class Config:
        orm_mode = True


class RecordBase(BaseModel):
    reason: str
    makedate: date
    punish: str
    dealt: bool

# 这里car实现外键数据的Car化
class Record(RecordBase):
    id: int
    car: Car

    class Config:
        orm_mode = True

# latest是查询后接口返回的数据格式，recods这里获取到的数据是一个列表
class Latest(BaseModel):
    count: int
    currpage: int
    nexturl: str
    preurl: str
    records: List[Record]

    class Config:
        orm_mode = True


class Dealt(BaseModel):
    code: int
    msg: str

这里定义完成后的调用的关键字字段是response_model。

最终的mian.py

以上的工作完成的都是fastapi的周边工作，fastapi主要完成的是一个接口的实现，接口文档的自动生成（生成风格是swagger UI），在该文档中还可以实现接口的测试。

主要实现的步骤如下：

• 定义Fsatapi对象

• 在对象中以get，post，delete，patch +路由的形式定义接口

  例子：get（url ，name=“文档中的名字”，description=“具体的描述”，response_model=“schemas模块中的定义”）

from fastapi import FastAPI, Depends
from pydantic.schema import date

# 使用sqlalchemy对数据库进行crud操作
from sqlalchemy.orm import Session

# 导入支持访问静态文件的相关包
from starlette.requests import Request
from starlette.responses import FileResponse
from starlette.staticfiles import StaticFiles

import crud
import schemas
from database import SessionLocal


def get_db():
    """
    获取事务
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


app = FastAPI()

app.mount("/static", StaticFiles(directory="static"))


@app.get("/")
async def index(request: Request):
    """

    :param request:
    :return:
    """
    return FileResponse('./static/index.html')


@app.get("/api/car", name="获取车辆信息", description="获取所有车辆信息")
async def get_car(db: Session = Depends(get_db)):
    """

    :param db:
    :return:
    """
    car = crud.read_car(db)
    return car


@app.get("/api/record/", name="获取所有违章记录", description="获取所有违章记录", response_model=schemas.Latest)
async def get_all_record(page: int = 1, carno: str = None, start: date = None, end: date = None,
                         db: Session = Depends(get_db)):
    """
    :param page:
    :param carno:
    :param start:
    :param end:
    :param db:
    :return:
    """
    latest = crud.read_all_record(carno=carno, start=start, end=end, page=page, db=db)
    return latest


@app.patch("/api/record/{record_id}", name="update car record", description="处理违章记录", response_model=schemas.Dealt)
async def dealt_record(record_id: int, db: Session = Depends(get_db)):
    """
    :param record_id:
    :param db:
    :return:
    """
    dealt = crud.update_dealt(db=db, record_id=record_id)
    return dealt


@app.delete("/api/record/{record_id}", name="del car record", description="删除违章记录", response_model=schemas.Dealt)
async def dealt_record(record_id: int, db: Session = Depends(get_db)):
    """
    :param record_id:
    :param db:
    :return:
    """
    dealt = crud.del_record(db=db, record_id=record_id)
    return dealt


if __name__ == '__main__':
    import uvicorn

    # uvicorn main: app --reload
    uvicorn.run(app, host='127.0.0.1', port=8000)

想看到页面的必经之路

fastapi本身是做接口的，就像数据操作需要sqlalchemy来实现，这里实现使用到的库是starlette==0.13.4。

接口的页面的静态文件在我的仓库里，后端开发不需要为自己写前端文件啦。

项目源代码在这里。

"
dealt = crud.del_record(db=db, record_id=record_id)
return dealt

if name == ‘main’:
import uvicorn

# uvicorn main: app --reload
uvicorn.run(app, host='127.0.0.1', port=8000)

### 想看到页面的必经之路

[外链图片转存中...(img-kyUinaZm-1593692207800)]

fastapi本身是做接口的，就像数据操作需要sqlalchemy来实现，这里实现使用到的库是starlette==0.13.4。

接口的页面的静态文件在我的仓库里，后端开发不需要为自己写前端文件啦。

项目源代码在[这里](https://github.com/hhs44/fastapi_example)。

本文地址：https://blog.csdn.net/hsh969/article/details/107092312