书接上文,上篇文章唠了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:[email protected]:3307/fastapi"
# _SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:[email protected]: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
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)