企业官网建设流程全解析

北京网站制作策划,现在一般做网站都是去哪家做的,百度关键词策划和seo的优化,建设网站的好公司FastAPI 完整 RESTful API 示例#xff1a;JWT 认证与限流 基于FastAPI的完整RESTful API示例 下面是一个可直接运行的、基于PythonFastAPI的RESTful API示例#xff0c;包含用户资源的CRUD、JWT认证、接口限流和自动文档化#xff0c;并提供完整的代码和运行说明#xff0…FastAPI 完整 RESTful API 示例JWT 认证与限流基于FastAPI的完整RESTful API示例下面是一个可直接运行的、基于PythonFastAPI的RESTful API示例包含用户资源的CRUD、JWT认证、接口限流和自动文档化并提供完整的代码和运行说明确保代码开箱即用。一、环境准备首先安装所需依赖执行以下命令pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart slowapi httpx依赖说明fastapi核心Web框架原生支持RESTful、自动文档、依赖注入uvicornASGI服务器用于运行FastAPI应用python-joseJWT认证相关passlib密码哈希bcrypt算法slowapi接口限流扩展核心限流依赖库httpx内部请求测试可选二、完整代码实现创建文件main.py复制以下完整代码from datetime import datetime, timedelta from typing import Optional, List from fastapi import FastAPI, Depends, HTTPException, status, Query from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from fastapi.middleware.cors import CORSMiddleware from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel, EmailStr, Field from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded # -------------------------- 配置项 -------------------------- # 密钥生产环境请改为随机生成的安全密钥可通过 openssl rand -hex 32 生成 SECRET_KEY 09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7 ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 # Token有效期30分钟 # 密码哈希上下文 pwd_context CryptContext(schemes[bcrypt], deprecatedauto) # OAuth2认证方案 oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # 限流配置基于IP # 1. 初始化限流器key_func指定限流维度此处为客户端IP limiter Limiter(key_funcget_remote_address) # FastAPI应用初始化 app FastAPI( title用户管理RESTful API, description基于FastAPI实现的用户CRUD、JWT认证、限流示例, version1.0.0 ) # 2. 注册限流中间件与异常处理器 # 绑定限流器到应用状态使全局可用 app.state.limiter limiter # 注册限流超额异常处理器返回标准化响应 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 跨域配置允许前端访问 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境请指定具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # -------------------------- 数据模型 -------------------------- # 模拟数据库生产环境替换为真实数据库如PostgreSQL/MySQL fake_users_db { admin: { id: 1, username: admin, email: adminexample.com, full_name: Admin User, hashed_password: pwd_context.hash(admin123), disabled: False, role: admin }, testuser: { id: 2, username: testuser, email: testexample.com, full_name: Test User, hashed_password: pwd_context.hash(test123), disabled: False, role: user } } # Pydantic模型请求/响应验证 class UserBase(BaseModel): username: str Field(..., min_length3, max_length50, description用户名) email: EmailStr Field(..., description用户邮箱) full_name: Optional[str] Field(None, description用户全名) disabled: Optional[bool] Field(False, description是否禁用) role: str Field(user, pattern^(admin|user)$, description角色admin/user) class UserCreate(UserBase): password: str Field(..., min_length6, description密码至少6位) class UserUpdate(BaseModel): email: Optional[EmailStr] Field(None, description用户邮箱) full_name: Optional[str] Field(None, description用户全名) disabled: Optional[bool] Field(None, description是否禁用) role: Optional[str] Field(None, pattern^(admin|user)$, description角色admin/user) class UserResponse(UserBase): id: int Field(..., description用户ID) class Config: orm_mode True # 支持从字典/ORM对象转换 class Token(BaseModel): access_token: str Field(..., description访问令牌) token_type: str Field(..., description令牌类型bearer) class TokenData(BaseModel): username: Optional[str] None # -------------------------- 工具函数 -------------------------- def verify_password(plain_password, hashed_password): 验证密码 return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): 生成密码哈希 return pwd_context.hash(password) def get_user(db, username: str): 从数据库获取用户 if username in db: user_dict db[username] return UserResponse(**user_dict) def authenticate_user(fake_db, username: str, password: str): 用户认证 user get_user(fake_db, username) if not user: return False if not verify_password(password, fake_db[username][hashed_password]): return False return user def create_access_token(data: dict, expires_delta: Optional[timedelta] None): 生成JWT令牌 to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt async def get_current_user(token: str Depends(oauth2_scheme)): 获取当前登录用户依赖项 credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无法验证凭据, 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 token_data TokenData(usernameusername) except JWTError: raise credentials_exception user get_user(fake_users_db, usernametoken_data.username) if user is None: raise credentials_exception return user async def get_current_active_user(current_user: UserResponse Depends(get_current_user)): 获取当前活跃用户依赖项 if current_user.disabled: raise HTTPException(status_code400, detail用户已被禁用) return current_user async def get_current_admin_user(current_user: UserResponse Depends(get_current_active_user)): 获取当前管理员用户依赖项 if current_user.role ! admin: raise HTTPException(status_code403, detail需要管理员权限) return current_user # -------------------------- API接口 -------------------------- # 3. 为接口添加限流装饰器指定限流规则 app.post(/token, response_modelToken, summary获取访问令牌, tags[认证]) limiter.limit(5/minute) # 限制每分钟最多5次请求格式次数/时间单位second/minute/hour/day async def login_for_access_token(form_data: OAuth2PasswordRequestForm Depends()): 用户登录获取JWT令牌 - 用户名/密码验证 - 返回有效期30分钟的Bearer令牌 user authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user.username}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer} app.get(/users/me/, response_modelUserResponse, summary获取当前用户信息, tags[用户管理]) limiter.limit(10/minute) # 普通接口可放宽限流每分钟10次 async def read_users_me(current_user: UserResponse Depends(get_current_active_user)): 获取当前登录用户的详细信息需要认证 return current_user app.get(/users/, response_modelList[UserResponse], summary获取用户列表, tags[用户管理]) limiter.limit(20/minute) # 列表查询接口限流可更高每分钟20次 async def read_users( skip: int Query(0, ge0, description跳过条数), limit: int Query(10, ge1, le100, description每页条数), current_user: UserResponse Depends(get_current_admin_user) ): 获取用户列表仅管理员可访问支持分页 user_list list(fake_users_db.values()) return [UserResponse(**user) for user in user_list[skip : skip limit]] app.get(/users/{user_id}, response_modelUserResponse, summary获取单个用户, tags[用户管理]) limiter.limit(10/minute) async def read_user( user_id: int, current_user: UserResponse Depends(get_current_active_user) ): 获取指定ID的用户信息 - 管理员可查看所有用户 - 普通用户仅可查看自己 # 查找用户 user next((u for u in fake_users_db.values() if u[id] user_id), None) if user is None: raise HTTPException(status_code404, detail用户不存在) # 权限检查 if current_user.role ! admin and current_user.id ! user_id: raise HTTPException(status_code403, detail无权限访问该用户信息) return UserResponse(**user) app.post(/users/, response_modelUserResponse, status_code201, summary创建用户, tags[用户管理]) limiter.limit(5/minute) # 写操作接口限流严格防止恶意创建 async def create_user( user: UserCreate, current_user: UserResponse Depends(get_current_admin_user) ): 创建新用户仅管理员可操作 # 检查用户名是否已存在 if user.username in fake_users_db: raise HTTPException(status_code409, detail用户名已存在) # 生成新用户ID new_id max(u[id] for u in fake_users_db.values()) 1 # 构建新用户数据 new_user { id: new_id, username: user.username, email: user.email, full_name: user.full_name, hashed_password: get_password_hash(user.password), disabled: user.disabled, role: user.role } # 存入模拟数据库 fake_users_db[user.username] new_user return UserResponse(**new_user) app.put(/users/{user_id}, response_modelUserResponse, summary更新用户, tags[用户管理]) limiter.limit(5/minute) # 写操作接口限流严格 async def update_user( user_id: int, user_update: UserUpdate, current_user: UserResponse Depends(get_current_active_user) ): 更新用户信息 - 管理员可更新所有用户 - 普通用户仅可更新自己 # 查找用户 user_dict next((u for u in fake_users_db.values() if u[id] user_id), None) if user_dict is None: raise HTTPException(status_code404, detail用户不存在) # 权限检查 if current_user.role ! admin and current_user.id ! user_id: raise HTTPException(status_code403, detail无权限更新该用户信息) # 更新字段仅更新非None的字段 if user_update.email: user_dict[email] user_update.email if user_update.full_name is not None: user_dict[full_name] user_update.full_name if user_update.disabled is not None: user_dict[disabled] user_update.disabled if user_update.role and current_user.role admin: # 只有管理员能改角色 user_dict[role] user_update.role return UserResponse(**user_dict) app.delete(/users/{user_id}, status_code204, summary删除用户, tags[用户管理]) limiter.limit(5/minute) # 高危操作限流最严格 async def delete_user( user_id: int, current_user: UserResponse Depends(get_current_admin_user) ): 删除用户仅管理员可操作 # 查找并删除用户 username_to_delete None for username, user in fake_users_db.items(): if user[id] user_id: username_to_delete username break if username_to_delete is None: raise HTTPException(status_code404, detail用户不存在) del fake_users_db[username_to_delete] return None # 204状态码无响应体 # -------------------------- 启动入口 -------------------------- if __name__ __main__: import uvicorn uvicorn.run(main:app, host0.0.0.0, port8000, reloadTrue)三、代码核心说明1. 核心功能模块模块功能说明配置项定义JWT密钥、算法、Token有效期、密码哈希规则等数据模型使用Pydantic做请求/响应数据验证定义用户相关模型工具函数密码加密/验证、JWT生成/解析、用户认证等核心逻辑依赖项get_current_user获取当前用户、get_current_admin_user管理员权限等实现权限控制API接口完整的用户CRUD 认证接口符合RESTful规范2. RESTful规范体现资源URI用名词/users表示用户资源/users/{user_id}表示单个用户HTTP方法GET查询用户/users/、/users/{user_id}、/users/me/POST创建用户/users/、登录获取Token/tokenPUT全量更新用户/users/{user_id}DELETE删除用户/users/{user_id}状态码200成功查询/更新201创建成功204删除成功401未认证/Token无效403权限不足404资源不存在409用户名冲突429请求过于频繁限流触发3. 关键特性实现JWT认证通过/token接口获取Token后续请求在Header中携带Authorization: Bearer {token}接口限流使用slowapi实现基于IP的限流如/token接口限制5次/分钟自动文档FastAPI原生支持OpenAPI文档无需额外配置权限控制通过依赖项区分普通用户/管理员权限数据验证Pydantic自动验证请求参数如邮箱格式、密码长度四、FastAPI接口限流详解1. 限流核心原理接口限流是通过控制单位时间内客户端对接口的请求次数防止恶意请求、爬虫攻击或流量峰值导致服务过载保障系统稳定性。本示例基于slowapi库实现核心依赖limits限流算法库支持多种限流策略和维度。2. 限流实现三步曲1初始化配置首先初始化限流器指定限流维度key_func常用维度包括get_remote_address基于客户端IP限流本示例用法适合公网接口自定义函数基于用户ID、Token、请求头信息限流适合已认证接口需结合JWT。示例代码# 基于IP限流 from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) # 基于用户ID限流自定义key_func async def get_user_id(token: str Depends(oauth2_scheme)): current_user await get_current_user(token) return current_user.id # 以用户ID作为限流键 limiter Limiter(key_funcget_user_id)2注册中间件与异常处理将限流器绑定到FastAPI应用并注册限流超额异常处理器确保触发限流时返回标准化的429响应符合HTTP规范。# 绑定限流器 app.state.limiter limiter # 注册异常处理器返回429状态码和提示信息 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)3为接口添加限流规则通过limiter.limit()装饰器为接口指定限流规则格式为「请求次数/时间单位」支持的时间单位second秒、minute分、hour时、day天。常见规则示例limiter.limit(10/second)每秒最多10次请求limiter.limit(100/minute)每分钟最多100次请求limiter.limit(1000/hour)每小时最多1000次请求。限流规则设计建议根据接口重要性和资源消耗调整写操作POST/PUT/DELETE限流严格读操作GET可适当放宽。3. 进阶限流配置1自定义限流响应默认限流响应为JSON格式可自定义响应内容、状态码和响应头如添加重试时间提示from fastapi.responses import JSONResponse async def custom_rate_limit_handler(request: Request, exc: RateLimitExceeded): return JSONResponse( status_code429, content{ code: TOO_MANY_REQUESTS, message: f请求过于频繁请{exc.retry_after}秒后重试, retry_after: exc.retry_after }, headers{Retry-After: str(exc.retry_after)} ) # 注册自定义异常处理器 app.add_exception_handler(RateLimitExceeded, custom_rate_limit_handler)2全局限流与局部限流结合可设置全局默认限流规则同时为特殊接口单独设置更严格的规则# 全局默认限流每分钟20次 limiter Limiter(key_funcget_remote_address, default_limits[20/minute]) # 局部接口覆盖全局规则每分钟5次 app.post(/token) limiter.limit(5/minute) async def login(): pass3支持分布式限流本示例为单机限流基于内存存储限流计数分布式部署时需使用共享存储如Redis实现跨节点限流# 安装依赖pip install limits[redis] from limits.storage import RedisStorage from limits.strategies import FixedWindowRateLimiter import redis # 连接Redis redis_client redis.Redis(hostlocalhost, port6379, db0) storage RedisStorage(redis_client) # 初始化限流器指定Redis存储 limiter Limiter( key_funcget_remote_address, storage_uriredis://localhost:6379/0, # 也可直接通过URI指定 strategyFixedWindowRateLimiter() # 限流算法 )4限流算法选择slowapi支持多种限流算法可根据需求选择固定窗口算法FixedWindowRateLimiter时间窗口固定实现简单默认算法滑动窗口算法SlidingWindowRateLimiter时间窗口滑动限流更精准避免窗口边界突发流量滑动窗口计数器算法SlidingWindowCounterRateLimiter结合固定窗口和滑动窗口优点性能均衡。4. 限流注意事项限流维度选择公网接口优先用IP限流内部接口可用用户ID/Token限流避免单一维度被绕过规则合理性需结合业务场景和服务器承载能力调整避免过松导致过载过严影响正常用户分布式兼容分布式部署必须使用共享存储Redis否则各节点独立计数限流失效异常监控需监控429状态码频率若突然激增可能是被攻击需及时调整规则或排查问题白名单配置可对可信IP如内部服务IP配置限流白名单避免正常业务被限流。五、运行与测试1. 启动服务python main.py服务会运行在http://localhost:80002. 访问自动文档Swagger UIhttp://localhost:8000/docs推荐可直接在线测试ReDochttp://localhost:8000/redoc3. 测试步骤通过Swagger UI登录获取Token打开http://localhost:8000/docs点击/token接口 →Try it out输入用户名admin密码admin123→Execute响应中会返回access_token复制该值认证授权点击页面顶部Authorize按钮输入Bearer {access_token}替换{access_token}为刚才复制的令牌→Authorize测试用户接口测试GET /users/获取所有用户仅管理员测试POST /users/创建新用户输入用户名、邮箱、密码等测试GET /users/{user_id}获取指定用户测试PUT /users/{user_id}更新用户信息测试DELETE /users/{user_id}删除用户限流测试快速重复调用/token接口超过5次/分钟会返回429状态码提示请求过于频繁等待1分钟后限流计数重置可再次正常调用。六、生产环境优化建议替换模拟数据库将fake_users_db替换为真实数据库如SQLAlchemy PostgreSQL安全配置生成随机安全的SECRET_KEYopenssl rand -hex 32启用HTTPS限制CORS的allow_origins为具体域名日志与监控添加日志记录、接口调用监控限流优化可基于用户ID而非IP限流适配分布式部署密码策略增加密码复杂度校验Token刷新实现Token刷新机制避免频繁登录总结本次实现的RESTful API具备以下核心特性符合RESTful设计规范资源导向、HTTP方法语义化、正确的状态码完整的用户CRUD功能包含精细的权限控制普通用户/管理员安全的JWT认证机制密码哈希存储接口限流保护防止恶意请求自动生成交互式API文档便于测试和对接完善的数据验证保证接口输入输出的规范性代码可直接运行你可以基于此示例扩展更多功能如用户头像、角色管理、数据持久化等。