---

一、pandas是什么？

示例：pandas 是基于NumPy 的一种工具，该工具是为了解决数据分析任务而创建的。

二、使用步骤

1.引入库

代码如下（示例）：

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import warnings
warnings.filterwarnings('ignore')
import  ssl
ssl._create_default_https_context = ssl._create_unverified_context

2.读入数据

代码如下（示例）：

data = pd.read_csv(
    'https://labfile.oss.aliyuncs.com/courses/1283/adult.data.csv')
print(data.head())

该处使用的url网络请求的数据。

---

总结

提示：这里对文章进行总结：
例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。例如：以上就是今天要讲的内容，本文仅仅简单介绍了pandas的使用，而pandas提供了大量能使我们快速便捷地处理数据的函数和方法。@TOC

使用fastapi

install

pip install fastapi
pip install uvicorn
pip install sqlalchemy

File Structure

sqlapp
		database.py	# 连接数据库文件
		crud.py			# 增删改查
		models.py		# 表文件
		schemas.py		
		
utils
		jwt.py			# 基于用户jwt
		
CustomMiddleware
		verification_user.py  # 自定义中间件
		
app
		main.py		#程序入口

sql

fastapi使用sql

1. database.py
   首先需要定义数据库部分，作为demo，使用sqlite作为我们的数据库。这部分都是常规配置操作。

   from sqlalchemy import create_engine
   from sqlalchemy.ext.declarative import declarative_base
   from sqlalchemy.orm import sessionmaker
   
   SQLALCHEMY_DATABASE_URL = "sqlite:///./fast_model.db"
   # SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
   # SQLALCHEMY_DATABASE_URL = "mysql://root:123456@localhost:3306/fast"
   
   engine = create_engine(
       SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
   )
   SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
   Base = declarative_base()
   

2. models.py
   第二步需要定义我们的表结构，这里定义了User表存储用户数据，item表用于存储物品数据，两者关系是一对多。

   from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
   from sqlalchemy.orm import relationship
   
   from .database import Base
   
   
   class User(Base):
       __tablename__ = "users"
   
       id = Column(Integer, primary_key=True, index=True)
       email = Column(String, unique=True, index=True)
       hashed_password = Column(String)
       is_active = Column(Boolean, default=True)
   
       # item表用于存储物品数据，两者关系是一对多
       items = relationship("Item", back_populates="owner")
   
       def to_dict(self):
           return {c.name: getattr(self, c.name) for c in self.__table__.columns}
   
   
   class Item(Base):
       __tablename__ = "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("User", back_populates="items")
   
       def to_dict(self):
           return {c.name: getattr(self, c.name) for c in self.__table__.columns}
   

3. schemas.py
   需要定义fastapi中schemas信息，后续路由、CRUD时都需要使用。由于增删改查需要不同的schema，所以官网的最佳实践一般都是通过继承解决。

   from typing import List,Optional
   
   from pydantic import BaseModel
   from app.main import *
   
   class ItemBase(BaseModel):
       title: str
       description: str = None
   
   
   class ItemCreate(ItemBase):
       pass
   
   
   class ItemUpdate(ItemBase):
       pass
   
   
   class Item(ItemBase):
       id: int
       owner_id: int
   
       class Config:
           orm_mode = True
   
   
   class UserBase(BaseModel):
       email: str
   
   
   class UserCreate(UserBase):
       password: str
   
   
   class UserUpdate(UserBase):
       is_active: bool
   
   
   class User(UserBase):
       id: int
       is_active: bool
       items: List[Item] = []
   
       class Config:
           orm_mode = True
   
   
   class Token(BaseModel):
       """定义token的数据模型"""
       access_token: str
       token_type: str
       
   
   class TokenData(BaseModel):
       username: Optional[str] = None
   

4. crud.py
   接下来就是实际完成CRUD，官网上没有对update、delete做实例展示，这里补充一下。

   # session 默认为数据库
   from sqlalchemy.orm import Session
   
   # 导入本地文件
   from . import models, schemas
   
   
   # 传入session userid
   def get_user(db: Session, user_id: int):
       # 根据传入的userid筛选出user
       return db.query(models.User).filter(models.User.id == user_id).first()
   
   
   def get_user_by_email(db: Session, email: str):
       # 根据email查询用户
       return db.query(models.User).filter(models.User.email == email).first()
   
   
   def get_users(db: Session, skip: int = 0, limit: int = 100):
       # skip可以理解为那个id之后的 limit显示多少条数据
       return db.query(models.User).offset(skip).limit(limit).all()
   
   
   def create_user(db: Session, user: schemas.UserCreate):
       # 新建user user根据 UserCreate 来验证传入的数据
   
       # 真实的password 原名为hashed_password 在用户写入的password 后面加入notreallyhashed
       fake_hashed_password = user.password + "notreallyhashed"
       db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
       # add
       db.add(db_user)
       # commit
       db.commit()
       # 刷新
       db.refresh(db_user)
       return db_user
   
   
   def update_user(db: Session, user_id: int, update_user: schemas.UserUpdate):
       db_user = db.query(models.User).filter(models.User.id == user_id).first()
       if db_user:
           update_dict = update_user.dict(exclude_unset=True)
           for k, v in update_dict.items():
               setattr(db_user, k, v)
           db.commit()
           # 相当于刷新
           db.flush()
           db.refresh(db_user)
           return db_user
   
   
   def delete_user(db: Session, user_id: int):
       # 删除user
       db_user = db.query(models.User).filter(models.User.id == user_id).first()
       if db_user:
           db.delete(db_user)
           db.commit()
           db.flush()
           return db_user
   
   
   def get_items(db: Session, skip: int = 0, limit: int = 100):
       # 获取item 参数同上
       return db.query(models.Item).offset(skip).limit(limit).all()
   
   
   def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
       # 
       db_item = models.Item(**item.dict(), owner_id=user_id)
       db.add(db_item)
       db.commit()
       db.refresh(db_item)
       return db_item
   
   
   def relate_user_item(db: Session, user_id: int, item_id: int):
       db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
       if db_item:
           db_item.owner_id = user_id
           db.commit()
           db.flush()
           return db.query(models.User).filter(models.User.id == user_id).first()
   
   
   def update_item(db: Session, item_id: int, update_item: schemas.ItemUpdate):
       # 修改item
       db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
       if db_item:
           update_dict = update_item.dict(exclude_unset=True)
           for k, v in update_dict.items():
               setattr(db_item, k, v)
           db.commit()
           db.flush()
           db.refresh(db_item)
           return db_item
   
   
   def delete_item(db: Session, item_id: int):
       db_item = db.query(models.Item).filter(models.Item.id == item_id).first()
       if db_item:
           db.delete(db_item)
           db.commit()
           db.flush()
           return db_item
   

5. main.py
   最后完成路由的配置，基本都是按照官网上来的，只是稍微补充了一些删和改的内容，为了让这个demo更加完善

       from typing import List
       import time
   
   	from fastapi import Depends, FastAPI, HTTPException,Request, status
   	from sqlalchemy.orm import Session
   	
   	from sqlapp import crud, models, schemas
   	from sqlapp.database import SessionLocal, engine
   	from utils.jwt import create_access_token,get_current_active_user
   	# from CustomMiddleware.verification_user import add_process_time_header
   	
   	
   	models.Base.metadata.create_all(bind=engine)
   	
   	app = FastAPI()
   	
   	async def get_app():
   	    return app
   	
   	# Dependency
   	def get_db():
   	    try:
   	        db = SessionLocal()
   	        yield db
   	    finally:
   	        db.close()
   	        
   	@app.post("/users/", response_model=schemas.User)
   	def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
   	    get_current_user()
   	    db_user = crud.get_user_by_email(db, email=user.email)
   	    if db_user:
   	        raise HTTPException(status_code=400, detail="Email already registered")
   	    return crud.create_user(db=db, user=user)
   	
   	
   	@app.get("/users/", response_model=List[schemas.User])
   	def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
   	    users = crud.get_users(db, skip=skip, limit=limit)
   	    return users
   	
   	
   	@app.get("/users/{user_id}", response_model=schemas.User)
   	def read_user(user_id: int, db: Session = Depends(get_db)):
   	    db_user = crud.get_user(db, user_id=user_id)
   	    if db_user is None:
   	        raise HTTPException(status_code=404, detail="User not found")
   	    return db_user
   	
   	
   	@app.delete('/users/{user_id}', response_model=schemas.User)
   	def delete_user(user_id: int, db: Session = Depends(get_db)):
   	    db_user = crud.delete_user(db, user_id=user_id)
   	    if db_user is None:
   	        raise HTTPException(status_code=404, detail="User not found")
   	    return db_user
   	
   	
   	@app.put("/users/{user_id}", response_model=schemas.User)
   	def update_user(user_id: int, update_user: schemas.UserUpdate, db: Session = Depends(get_db)):
   	    updated_user = crud.update_user(db, user_id, update_user)
   	    if updated_user is None:
   	        raise HTTPException(status_code=404, detail="User not found")
   	    return updated_user
   	
   	
   	@app.post("/users/{user_id}/items/", response_model=schemas.Item)
   	def create_item_for_user(
   	    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
   	):
   	    return crud.create_user_item(db=db, item=item, user_id=user_id)
   	
   	
   	@app.get("/items/", response_model=List[schemas.Item])
   	def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
   	    items = crud.get_items(db, skip=skip, limit=limit)
   	    return items
   	
   	
   	@app.put("/items/{user_id}/{item_id}/", response_model=schemas.User)
   	def relate_user_item(user_id: int, item_id: int, db: Session = Depends(get_db)):
   	    user = crud.relate_user_item(db=db, item_id=item_id, user_id=user_id)
   	    return user
   	
   	
   	@app.put("/items/{item_id}", response_model=schemas.Item)
   	def update_item(item_id: int, update_item: schemas.ItemUpdate, db: Session = Depends(get_db)):
   	    updated_item = crud.update_item(db, item_id, update_item)
   	    if updated_item is None:
   	        raise HTTPException(status_code=404, detail="Item not found")
   	    return updated_item
   	
   	
   	@app.delete('/items/{item_id}', response_model=schemas.Item)
   	def delete_item(item_id: int, db: Session = Depends(get_db)):
   	    db_item = crud.delete_item(db, item_id=item_id)
   	    if db_item is None:
   	        raise HTTPException(status_code=404, detail="User not found")
   	    return db_item
   
   

执行

在文件路径下执行，–reload当代码有修改时，可以自动加载。

uvicorn sql_app.main:app --reload

查看

可以通过postman发送请求做测试，这里贴图一张。

最后fastapi还可以自动生成文档，可以访问http://127.0.0.1:8000/docs查阅。

jwt

关于JWT

WT的意思是JSON Web Tokens，它是现阶段非常流行的认证系统。互联网服务总是离不开认证，在过去，开发者们通常使用 cookies 和 session来做认证系统，但随着互联网的发展，这种方式被淘汰。取而代之的，便是JWT。

JWT的原理

JWT的原理较为简单，为了方便理解，假设我们在登录场景。

1. 用户在网页上填写账号和密码后，通过JavaScript POST到服务端。
2. 服务端对接收的信息验证通过后，会生成一段json信息返回给客户端，信息大概是这样的：

   {"user_name": "lisi", "uid": 1, "id": 1}
   

3. 客户端收到该信息后，存储在浏览器的local Storage。
4. 客户端发送请求时，都携带这一段数据，那么服务端就知道客户端是谁在请求了。

但做到这些还不够，为了防止伪造和篡改，我们还需要一个只有服务器知道的密钥，并且使用HS256算法基于此密钥在签发的时候对信息加密，在接收信息时对信息解密。

安装相关模块

pip install python-jose[cryptography]		# 使用jose来签发和验证 JWT tokens
pip install passlib[bcrypt] 				# 使用passlib来处理密码hash

处理密码hash和验证

密码本身只是一段字符串，我们存储在数据库中的密码都是经过hash处理的，这样的好处在于，即便是数据库因为入侵或者其他原因泄露，那些不法分子也不能轻易破解出用户的密码，有利于保护用户隐私。

from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# 验证密码hash
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)
# 计算密码hash值
def get_password_hash(password):
    return pwd_context.hash(password)

一般而言，在用户设置密码时，我们将计算出密码的hash值并存储在数据库中，用户登录时，服务端计算用户输入的密码hash值与数据库中的hash值比对。如果一致，则密码正确。

完整jwt代码 utils/jwt.py

from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from sqlalchemy.orm import Session

from sqlapp.database import SessionLocal
from sqlapp.schemas import *
from sqlapp.crud import *



# 如果需要得到新的 SECRET_KEY 则需要运行  openssl rand -hex 32
SECRET_KEY = "bf3af7cd522a80657063b1eef7c6e54326656b286af90ab1eb79ea5eb68479d1"
# 定义加密解密所使用的算法
ALGORITHM = "HS256"
# token的有效期
ACCESS_TOKEN_EXPIRE_MINUTES = 30


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


# 创建一个加密解密上下文环境（甚至可以不用管这两句话啥意思）
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """创建tokens函数
    :param data: 对用JWT的Payload字段，这里是tokens的载荷，在这里就是用户的信息
    :param expires_delta: 缺省参数，截止时间
    :return:
    """
    # 深拷贝data
    to_encode = data.copy()
    # 如果携带了截至时间，就单独设置tokens的过期时间
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    # 编码,至此 JWT tokens诞生
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme),db: Session = Depends(get_db)):
    """获取当前用户信息，实际上是一个解密token的过程
    :param token: 携带的token
    :return:
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 解密tokens
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # 从tokens的载荷payload中获取用户名
        username: str = payload.get("sub")
        # 如果没有获取到，抛出异常
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # 从数据库查询用户信息
    user = get_user_by_email(db,email=username)
    if user is None:
        raise credentials_exception
    return user



async def get_current_active_user(current_user: User = Depends(get_current_user)):
    """获取当前用户信息，实际上是作为依赖，注入其他路由以使用。
    :param current_user:
    :return:
    """
    return current_user

app/main.py

from sqlapp import crud, models, schemas
from sqlapp.database import SessionLocal, engine
from utils.jwt import create_access_token,get_current_active_user

@app.post("/token", response_model=schemas.Token)
async def login_for_access_token(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """这里定义了一个接口，路径为 /token, 用于用户申请tokens
    :param form_data:
    :return:
    """
    # 首先对用户做出检查
    user = crud.get_user_by_email(db, email=user.email)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 创建token
    access_token = create_access_token(
        data={"sub": user.email}
    )
    # 返回token信息，JavaScript接收并存储，用于下次访问
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(current_user: models.User = Depends(get_current_active_user)):
    """获取当前用户信息
    :param current_user:
    :return:
    """
    return current_user

Tests

可以通过postman发送请求做测试，这里贴图两张。