python：Fastapi - 更大的应用

书接上文，上篇文章唠了SQL(关系)数据库，今天主要是唠下多文件，也就是对应的工程目录的应用开发。

工程目录概述

admin - 文件夹是作者规划的后台接口，可忽略
model - 文件夹是主要存放数据库表
routers - 文件夹也是作者规划的存放接口的，后续的实践接口也写在这里
schemas - 文件夹是存放的一些接口的模型
sql - 文件夹是存放一些基于sqlalchemy的查询方法
databases.py - 文件是数据库的连接操作，这里是使用的sqlite.db
dependencies.py - 文件是鉴权，也就是验证登录的，这里是写死的，主要是用于实践
main.py - 文件是程序主入口，也就是运行这个文件，整个应用就运行起来了
sc_database.db - 就是数据库

代码简单解读

./sc_app/model/user_models.py

from sqlalchemy.orm import relationship
from sc_app.databases import BASE
from sqlalchemy import Boolean, Column, Integer, String, ForeignKey

__all__ = [
    "Users",
    "Items"
]


class Users(BASE):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Items", back_populates="owner")


class Items(BASE):
    __tablename__ = "sc_items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("Users", back_populates="items")

信息注释：

这里是定义了两个表，也是基于sqlalchemy操作的，后续会总结一篇sqlalchemy的操作文章

./sc_app/routers/ro_user.py

from fastapi import Depends
from fastapi import APIRouter
from fastapi import HTTPException
from sc_app.schemas import sc_user
from sqlalchemy.orm import Session
from sc_app.sql import sq_user
from sc_app.databases import get_db
from typing import List

router = APIRouter()


@router.post("/add/", response_model=sc_user.User)
def sc_user_add(user: sc_user.UserPwd, db: Session = Depends(get_db)):
    """
    新增学员接口
    :param user:
    :param db:
    :return:
    """
    data = sq_user.get_user_username(db, username=user.username)
    if data:
        raise HTTPException(status_code=400, detail="Data does not exist or the name is duplicate !")
    return sq_user.create_user(db, user=user)


@router.put("/users/{user_id}", response_model=sc_user.User)
def sc_user_update(user_id: int, user: sc_user.UserPwd, db: Session = Depends(get_db)):
    """
    修改学员接口
    :param user_id:
    :param user:
    :param db:
    :return:
    """
    data = sq_user.update_user(db, user_id=user_id, user=user)
    if data is None:
        raise HTTPException(status_code=404, detail="user is not found")
    return data


@router.delete("/users/{user_id}", response_model=List[sc_user.User])
def sc_user_update(user_id: int, db: Session = Depends(get_db)):
    """
    删除接口
    :param user_id:
    :param db:
    :return:
    """
    data = sq_user.delete_user(db, user_id=user_id)
    if data is None:
        raise HTTPException(status_code=404, detail="user is not found")
    return data


@router.get("/users/", response_model=List[sc_user.User])
def sc_user_list(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    列表接口
    :param skip:
    :param limit:
    :param db:
    :return:
    """
    return sq_user.get_users(db, skip=skip, limit=limit)

信息注释：

定义了四个接口，分别是增删改查操作

./sc_app/schemas/sc_user.py

from pydantic import BaseModel


class BaseUser(BaseModel):
    username: str

    def __repr__(self):
        return self.username


class UserPwd(BaseUser):
    password: str

    def __repr__(self):
        return self.password


class User(BaseUser):
    id: int
    is_active: bool

    def __repr__(self):
        return self.id, self.is_active

    class Config:
        orm_mode = True

信息注释：

定义的接口的模型，这里的字段要和建表的字段一致，不然会报错
__repr__是查看显示原始属性
class Config: orm_mode = True 是否允许使用ORM模式，Config属性orm_mode必须设置True，才能创建模型实例

./sc_app/sql/sq_user.py

from sqlalchemy.orm import Session
from sc_app.model import user_models
from sc_app.schemas import sc_user


def get_user_username(db: Session, username: str):
    return db.query(user_models.Users).filter(user_models.Users.username == username).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(user_models.Users).offset(skip).limit(limit).all()


def delete_user(db: Session, user_id: int):
    user = db.query(user_models.Users).filter(user_models.Users.id == user_id).first()
    db.delete(user)
    db.commit()
    return db.query(user_models.Users).all()


def create_user(db: Session, user: sc_user.UserPwd):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = user_models.Users(username=user.username, password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def update_user(db: Session, user_id: int, user: sc_user.UserPwd):
    data = db.query(user_models.Users).filter(user_models.Users.id == user_id).first()
    data.username = user.password
    data.password = user.password + "notreallyhashed"
    db.commit()
    return db.query(user_models.Users).filter(user_models.Users.id == user_id).first()

信息注释：

这里使用的是sqlalchemy中提供的一些操作数据库的方法，后续会补一篇操作sqlalchemy操作数据库的文章。

./sc_app/databases.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

__all__ = ["BASE", "get_db"]


class _DBConnect:
    # 定义一个变量
    _SQLALCHEMY_DATABASE_URL = "sqlite:///./sc_database.db"
    # _SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@10.10.12.120:3307/fastapi"
    # _SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@192.168.0.110:3306/fastapi-dev?charset=utf8"
    # _SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

    # 创建一个连接
    _engine = create_engine(
        # echo=True参数表示连接发出的 SQL 将被记录到标准输出
        # future=True是为了确保我们使用最新的风格的API
        # _SQLALCHEMY_DATABASE_URL, pool_pre_ping=True, encoding="utf-8"
        _SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
    )

    # 创建一个持久对象，好处就是可以一次添加多个对象
    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=_engine)

    # 定义一个ORM模型基类
    Base = declarative_base()
    # 绑定连接，使用表元数据和引擎
    Base.metadata.create_all(bind=_engine)


# _DBConnect.Base赋值给BASE变量
BASE = _DBConnect.Base


def get_db():
    """
    初始化持久对象，并yield返回
    无论失败与否最后都会调用关闭
    :return:
    """
    db = _DBConnect.SessionLocal()
    try:
        yield db
    finally:
        db.close()

注意点：

connect_args={"check_same_thread": False}是只有你使用的数据库是sqlite时，才使用这个参数，如使用其他数据库则不需要

./sc_app/dependencies.py

from fastapi import Header, HTTPException


def verify_token(x_token: str = Header(...), token: str = Header(...)):
    """
    校验鉴权
    :param x_token:
    :param token:
    :return:
    """
    if x_token != "debugfeng" and token != "123456789":
        raise HTTPException(status_code=400, detail="not x_token and token in header !")

信息注释：

verify_token函数是设置信息头中必须要有x_token和token参数，如果不满足写死的值，则抛出错误

注意点：

x_token参数在postman工具中传参时，要变成x-token

./sc_app/main.py

import uvicorn
from fastapi import Depends
from fastapi import FastAPI
from sc_app.routers import ro_user
from sc_app.dependencies import verify_token

app = FastAPI()
app.include_router(
    router=ro_user.router, prefix="/user", dependencies=[Depends(verify_token)]
)


@app.get("/")
def index():
    """
    首页
    :return:
    """
    return {"message": "Welcome to the home page !"}


if __name__ == '__main__':
    uvicorn.run(app="main:app", reload=True, debug=True)

信息注释：

include_router添加所有其他路径操作一起工作
router=ro_user.router是调用ro_user文件中router去运行
prefix="/user"是给接口路径定义一个前缀
dependencies=[Depends(verify_token)是关联鉴权值，依赖项

启动服务/请求接口
直接运行main.py文件，开始进行接口访问

今天先聊到这里吧，以上总结或许能帮助到你，或许帮助不到你，但还是希望能帮助到你，如有疑问、歧义，直接私信留言会及时修正发布；非常期待你的一键 3 连【点赞、收藏、分享】哟，谢谢！

未完成，待续……

一直在努力，希望你也是！

微信搜索公众号：就用python

python：Fastapi - 更大的应用 - 多个文件

Python相关栏目本月热门文章