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

工程目录概述

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

./sc_app/model/user_models.py

from sqlalchemy.orm import relationshipfrom sc_app.databases import BASEfrom 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 Dependsfrom fastapi import APIRouterfrom fastapi import HTTPExceptionfrom sc_app.schemas import sc_userfrom sqlalchemy.orm import Sessionfrom sc_app.sql import sq_userfrom sc_app.databases import get_dbfrom typing import Listrouter = 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 BaseModelclass BaseUser(BaseModel):    username: str    def __repr__(self):        return self.usernameclass UserPwd(BaseUser):    password: str    def __repr__(self):        return self.passwordclass 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 Sessionfrom sc_app.model import user_modelsfrom sc_app.schemas import sc_userdef 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_userdef 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_enginefrom sqlalchemy.orm import sessionmakerfrom 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.Basedef 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, HTTPExceptiondef 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 uvicornfrom fastapi import Dependsfrom fastapi import FastAPIfrom sc_app.routers import ro_userfrom sc_app.dependencies import verify_tokenapp = 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