当前位置： 首页 > article >正文

FastAPI学习最后一天： Cors跨域和token鉴权

article 2024/11/28 3:07:20

在这里插入图片描述

在FastAPI中配置CORS时，我们可以通过CORSMiddleware中间件来设置不同的限制。以下是几个具体的限制示例：
在这里插入图片描述

示例1：限制特定的源

在这个例子中，我们只允许来自http://example.com的跨域请求。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = ["http://example.com"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

示例2：使用正则表达式限制源

在这里插入图片描述

这里我们使用正则表达式来允许所有以http://example.com开头的域名进行跨域请求。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"https?://example\.com",
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

示例3：限制特定的HTTP方法

在这里插入图片描述

在这个例子中，我们只允许GET和POST方法的跨域请求。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = ["http://example.com"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

@app.post("/post")
async def post():
    return {"message": "POST request received"}

示例4：限制特定的请求头

在这里插入图片描述

这里我们只允许Content-Type和X-Custom-Header两个请求头。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = ["http://example.com"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["Content-Type", "X-Custom-Header"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

示例5：不允许凭据

在这个例子中，我们不允许跨域请求携带凭据（如Cookies）。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = ["http://example.com"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=False,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.get("/")
async def main():
    return {"message": "Hello World"}

这些示例展示了如何在FastAPI中使用CORSMiddleware来设置不同的CORS策略限制。根据你的应用需求，你可以选择适合的配置来确保应用的安全性和功能性。

在FastAPI中，你可以通过请求头中的Token来实现对所有接口的鉴权，并且可以设置某些接口不需要鉴权。以下是具体的实现方法和示例代码：

1. 使用依赖项实现Token鉴权

首先，你需要一个依赖项函数来解析和验证Token。这个函数将从请求头中获取Token，并进行验证。
在这里插入图片描述

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

# 配置项
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

# 验证Token的依赖项函数
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # 这里可以添加从数据库获取用户信息的逻辑
    return username

2. 保护路由

使用Depends依赖项来保护需要鉴权的路由。

@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_user)):
    return {"username": current_user}

在这里插入图片描述

3. 设置不鉴权的接口

对于不需要鉴权的接口，你可以直接定义它们而不使用Depends依赖项。

@app.get("/public")
async def read_public():
    return {"message": "这是公开信息，不需要鉴权"}

4. 登录接口

创建一个登录接口来生成Token。
在这里插入图片描述

from datetime import datetime, timedelta

# 生成Token的函数
def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

5. 综合示例

from fastapi import FastAPI
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

app = FastAPI()

SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

def authenticate_user(username: str, password: str):
    # 这里应该添加从数据库验证用户的逻辑
    if username == "admin" and password == "password":
        return username
    return False

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # 这里可以添加从数据库获取用户信息的逻辑
    return username

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=30)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: str = Depends(get_current_user)):
    return {"username": current_user}

@app.get("/public")
async def read_public():
    return {"message": "这是公开信息，不需要鉴权"}

在这个示例中，/users/me接口需要鉴权，而/public接口不需要鉴权。通过这种方式，你可以灵活地控制FastAPI应用中的鉴权逻辑。
在这里插入图片描述

学习Python是一个循序渐进的过程，以下是一些关键点，可以帮助你总结和巩固Python的学习：
在这里插入图片描述

  1. 基础语法：

    • 掌握变量、数据类型（如整数、浮点数、字符串、列表、元组、字典等）。
    • 理解控制流（if语句、for循环、while循环）。
    • 学习函数的定义和调用，以及如何传递参数。

  2. 面向对象编程：

    • 理解类和对象的概念，以及如何定义类和实例化对象。
    • 学习类的属性和方法，以及如何实现继承、封装和多态。

  3. 模块和包：

    • 了解如何导入和使用Python标准库中的模块。
    • 学习如何创建自定义模块和包，以及如何管理项目依赖。

  4. 异常处理：

    • 掌握try-except语句的使用，以及如何自定义异常。

  5. 文件操作：

    • 学习如何打开、读取、写入和关闭文件。
    • 理解文件路径和目录操作。

  6. 数据处理和分析：

    • 掌握列表推导式、生成器表达式和迭代器的使用。
    • 学习使用pandas库进行数据分析和处理。

  7. 网络编程：

    • 学习如何使用requests库进行HTTP请求。
    • 理解网络协议的基础知识，如HTTP和TCP/IP。

  8. Web开发：

    • 学习使用Flask或Django框架进行Web应用开发。
    • 理解RESTful API的概念和实现。

  9. 数据库操作：

    • 学习如何使用sqlite3或SQLAlchemy进行数据库操作。
    • 理解SQL语言的基础知识。

  10. 测试和调试：

    • 学习使用unittest框架进行单元测试。
    • 掌握使用调试工具，如pdb。

  11. 代码风格和最佳实践：

    • 遵循PEP 8代码风格指南。
    • 学习编写可读性高、可维护的代码。

  12. 高级特性：

    • 学习装饰器的使用。
    • 理解上下文管理器和with语句。
    • 学习异步编程和多线程/多进程。

  13. 项目实践：

    • 通过实际项目应用所学知识。
    • 学习版本控制工具，如Git。

  14. 持续学习：

    • 随着Python和相关技术的不断发展，持续学习新的库和框架。
    • 参与开源项目，与其他开发者交流。

每个学习阶段都有其重点，但最重要的是实践。通过编写代码、解决实际问题和构建项目，你可以更好地理解和掌握Python。此外，阅读他人的代码、参与编程社区讨论和学习最佳实践也是提高编程技能的有效途径。
在这里插入图片描述


http://www.kler.cn/a/412207.html

相关文章：

  • 关于SpringBoot集成Kafka
  • HTTP 管道传输与多路复用
  • 如何设计和实现通用唯一 Code 生成方法
  • nodepad配置c/c++ cmd快速打开创建项目文件
  • 【漏洞复现】CVE-2020-13925
  • Flume和kafka的整合：使用Flume将日志数据抽取到Kafka中
  • MySQL 存储引擎切换场景与示例
  • 泷羽Sec学习笔记：shell（2）永久环境变量和字符串显位
  • 【Vue】计算属性
  • Leetcode 每日一题 3.无重复字符的最长子串
  • 基于springboot的雪具销售系统
  • “华为杯”研究生数学建模比赛历年赛题汇总(2004-2024)
  • localStorage缓存 接口 配置
  • python写共享内存，格式json
  • 实践篇：青果IP助理跨境电商的高效采集
  • JQuery -- 第九课
  • AWS IAM 及其功能
  • 『VUE』33. 组件保持存活,切换组件时的生命周期（详细图文注释）
  • 标记matlab曲线的x坐标
  • pyhton+yaml+pytest+allure框架封装-全局变量接口关联
  • C#调用C++ DLL方法之C++/CLI(托管C++)
  • web分页查询
  • Linux 生成/proc/config.gz
  • Java项目实战II基于微信小程序的校运会管理系统（开发文档+数据库+源码）
  • 【redis】zset有序集合详解
  • 计算机网络习题解答--个人笔记（未完）