企业官网建设流程全解析

安徽网站建设方案优化,大数据时代的智能营销,云服务器搭建网站教程,wordpress 视频教程本文详解FastAPI的核心特性、项目实战和部署方案#xff0c;打造现代化的Python后端服务。前言Python Web框架那么多#xff0c;为什么选FastAPI#xff1f;性能最强#xff1a;基于Starlette#xff0c;性能媲美Node.js和Go类型提示#xff1a;Python类型注解#xff0…本文详解FastAPI的核心特性、项目实战和部署方案打造现代化的Python后端服务。前言Python Web框架那么多为什么选FastAPI性能最强基于Starlette性能媲美Node.js和Go类型提示Python类型注解IDE智能提示自动文档Swagger/ReDoc自动生成异步支持原生async/await今天来全面学习FastAPI。一、FastAPI简介1.1 对比其他框架框架性能异步类型提示自动文档FastAPI⭐⭐⭐⭐⭐✅✅✅Flask⭐⭐⭐❌❌❌Django⭐⭐⚠️❌❌Tornado⭐⭐⭐⭐✅❌❌1.2 核心特性markdown体验AI代码助手代码解读复制代码1. 高性能 - 基于StarletteASGI - 性能接近Node.js/Go 2. 开发效率 - 自动数据验证Pydantic - 自动API文档 - IDE智能补全 3. 标准化 - OpenAPI规范 - JSON Schema - OAuth2/JWT支持二、环境搭建2.1 安装bash体验AI代码助手代码解读复制代码# 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装FastAPI pip install fastapi # 安装ASGI服务器 pip install uvicorn[standard] # 常用依赖 pip install python-multipart # 表单 pip install python-jose[cryptography] # JWT pip install passlib[bcrypt] # 密码加密 pip install sqlalchemy # ORM pip install aiosqlite # 异步SQLite2.2 第一个应用python体验AI代码助手代码解读复制代码# main.py from fastapi import FastAPI app FastAPI( title我的API, descriptionFastAPI学习项目, version1.0.0 ) app.get(/) async def root(): return {message: Hello FastAPI!} app.get(/items/{item_id}) async def read_item(item_id: int, q: str None): return {item_id: item_id, q: q}2.3 运行bash体验AI代码助手代码解读复制代码# 开发模式自动重载 uvicorn main:app --reload --host 0.0.0.0 --port 8000 # 访问 # API: http://localhost:8000 # Swagger文档: http://localhost:8000/docs # ReDoc文档: http://localhost:8000/redoc三、核心功能3.1 路径参数python体验AI代码助手代码解读复制代码from fastapi import FastAPI, Path app FastAPI() app.get(/users/{user_id}) async def get_user( user_id: int Path(..., title用户ID, ge1) # 1 ): return {user_id: user_id} app.get(/files/{file_path:path}) # 匹配路径 async def read_file(file_path: str): return {file_path: file_path}3.2 查询参数python体验AI代码助手代码解读复制代码from fastapi import FastAPI, Query from typing import Optional, List app FastAPI() app.get(/items/) async def list_items( skip: int 0, limit: int Query(default10, le100), # 最大100 q: Optional[str] Query(None, min_length3, max_length50), tags: List[str] Query(default[]) ): return { skip: skip, limit: limit, q: q, tags: tags }3.3 请求体Pydantic模型python体验AI代码助手代码解读复制代码from fastapi import FastAPI from pydantic import BaseModel, Field, EmailStr from typing import Optional from datetime import datetime app FastAPI() class UserCreate(BaseModel): username: str Field(..., min_length3, max_length20) email: EmailStr password: str Field(..., min_length6) age: Optional[int] Field(None, ge0, le150) class Config: json_schema_extra { example: { username: 张三, email: zhangsanexample.com, password: 123456, age: 25 } } class UserResponse(BaseModel): id: int username: str email: str created_at: datetime app.post(/users/, response_modelUserResponse) async def create_user(user: UserCreate): # 创建用户逻辑 return { id: 1, username: user.username, email: user.email, created_at: datetime.now() }3.4 表单和文件上传python体验AI代码助手代码解读复制代码from fastapi import FastAPI, File, UploadFile, Form from typing import List app FastAPI() # 表单数据 app.post(/login/) async def login( username: str Form(...), password: str Form(...) ): return {username: username} # 文件上传 app.post(/upload/) async def upload_file(file: UploadFile File(...)): contents await file.read() return { filename: file.filename, size: len(contents), content_type: file.content_type } # 多文件上传 app.post(/uploads/) async def upload_files(files: List[UploadFile] File(...)): return {filenames: [f.filename for f in files]}四、依赖注入4.1 基本依赖python体验AI代码助手代码解读复制代码from fastapi import FastAPI, Depends, HTTPException, status app FastAPI() # 简单依赖 async def common_parameters(q: str None, skip: int 0, limit: int 10): return {q: q, skip: skip, limit: limit} app.get(/items/) async def list_items(commons: dict Depends(common_parameters)): return commons # 类作为依赖 class Pagination: def __init__(self, page: int 1, size: int 10): self.page page self.size size self.skip (page - 1) * size app.get(/users/) async def list_users(pagination: Pagination Depends()): return { page: pagination.page, size: pagination.size, skip: pagination.skip }4.2 数据库依赖python体验AI代码助手代码解读复制代码from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker # 数据库配置 DATABASE_URL sqliteaiosqlite:///./app.db engine create_async_engine(DATABASE_URL, echoTrue) AsyncSessionLocal sessionmaker(engine, class_AsyncSession, expire_on_commitFalse) # 数据库依赖 async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close() # 使用 app.get(/users/{user_id}) async def get_user(user_id: int, db: AsyncSession Depends(get_db)): # 使用db进行数据库操作 pass五、认证与授权5.1 JWT认证python体验AI代码助手代码解读复制代码from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from datetime import datetime, timedelta from pydantic import BaseModel app FastAPI() # 配置 SECRET_KEY your-secret-key-keep-it-secret ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 pwd_context CryptContext(schemes[bcrypt], deprecatedauto) oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # 模型 class Token(BaseModel): access_token: str token_type: str class User(BaseModel): username: str email: str disabled: bool False # 模拟用户数据库 fake_users_db { admin: { username: admin, email: adminexample.com, hashed_password: pwd_context.hash(admin123), disabled: False } } # 工具函数 def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def create_access_token(data: dict, expires_delta: timedelta None): to_encode data.copy() expire datetime.utcnow() (expires_delta or timedelta(minutes15)) to_encode.update({exp: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) 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 except JWTError: raise credentials_exception user fake_users_db.get(username) if user is None: raise credentials_exception return User(**user) # 登录接口 app.post(/token, response_modelToken) async def login(form_data: OAuth2PasswordRequestForm Depends()): user fake_users_db.get(form_data.username) if not user or not verify_password(form_data.password, user[hashed_password]): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误 ) access_token create_access_token( data{sub: user[username]}, expires_deltatimedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) ) return {access_token: access_token, token_type: bearer} # 受保护的接口 app.get(/users/me, response_modelUser) async def read_users_me(current_user: User Depends(get_current_user)): return current_user六、中间件与异常处理6.1 中间件python体验AI代码助手代码解读复制代码from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware import time app FastAPI() # CORS中间件 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境指定域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 自定义中间件 app.middleware(http) async def add_process_time_header(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response # 请求日志中间件 app.middleware(http) async def log_requests(request: Request, call_next): print(f请求: {request.method} {request.url}) response await call_next(request) print(f响应: {response.status_code}) return response6.2 异常处理python体验AI代码助手代码解读复制代码from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse app FastAPI() # 自定义异常 class CustomException(Exception): def __init__(self, code: int, message: str): self.code code self.message message # 异常处理器 app.exception_handler(CustomException) async def custom_exception_handler(request: Request, exc: CustomException): return JSONResponse( status_codeexc.code, content{code: exc.code, message: exc.message} ) # 全局异常处理 app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): return JSONResponse( status_code500, content{code: 500, message: 服务器内部错误} ) # 使用 app.get(/items/{item_id}) async def read_item(item_id: int): if item_id 0: raise CustomException(code400, messageID不能为负数) return {item_id: item_id}七、项目结构7.1 推荐结构bash体验AI代码助手代码解读复制代码project/ ├── app/ │ ├── __init__.py │ ├── main.py # 入口 │ ├── config.py # 配置 │ ├── database.py # 数据库 │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ └── user.py │ ├── schemas/ # Pydantic模型 │ │ ├── __init__.py │ │ └── user.py │ ├── routers/ # 路由 │ │ ├── __init__.py │ │ ├── users.py │ │ └── items.py │ ├── services/ # 业务逻辑 │ │ ├── __init__.py │ │ └── user_service.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── auth.py ├── tests/ # 测试 ├── requirements.txt └── README.md7.2 路由组织python体验AI代码助手代码解读复制代码# app/routers/users.py from fastapi import APIRouter, Depends router APIRouter( prefix/users, tags[用户管理] ) router.get(/) async def list_users(): return [] router.get(/{user_id}) async def get_user(user_id: int): return {user_id: user_id} # app/main.py from fastapi import FastAPI from app.routers import users, items app FastAPI() app.include_router(users.router) app.include_router(items.router)八、部署方案8.1 Docker部署dockerfile体验AI代码助手代码解读复制代码# Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, app.main:app, --host, 0.0.0.0, --port, 8000]yaml体验AI代码助手代码解读复制代码# docker-compose.yml version: 3.8 services: api: build: . ports: - 8000:8000 environment: - DATABASE_URLpostgresql://user:passdb:5432/mydb depends_on: - db restart: unless-stopped db: image: postgres:15 environment: - POSTGRES_USERuser - POSTGRES_PASSWORDpass - POSTGRES_DBmydb volumes: - postgres_data:/var/lib/postgresql/data volumes: postgres_data:8.2 Nginx反向代理nginx体验AI代码助手代码解读复制代码server { listen 80; server_name api.example.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }8.3 远程访问部署在内网服务器的API如何让外网访问markdown体验AI代码助手代码解读复制代码方案对比 1. 公网暴露 HTTPS → 需要公网IP安全风险 2. Cloudflare Tunnel → 配置复杂 3. 组网软件 → 最简单安全 使用组网软件如星空组网 - 服务器安装组网客户端 - 开发机安装组网客户端 - 通过虚拟IP访问API 适用场景 - 开发测试阶段 - 内部系统 - 不想暴露公网的API九、性能优化9.1 异步数据库python体验AI代码助手代码解读复制代码from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.future import select # 异步查询 async def get_users(db: AsyncSession): result await db.execute(select(User)) return result.scalars().all()9.2 后台任务python体验AI代码助手代码解读复制代码from fastapi import FastAPI, BackgroundTasks app FastAPI() def send_email(email: str, message: str): # 发送邮件的耗时操作 pass app.post(/notify/) async def send_notification( email: str, background_tasks: BackgroundTasks ): background_tasks.add_task(send_email, email, 通知消息) return {message: 通知已发送}9.3 缓存python体验AI代码助手代码解读复制代码from functools import lru_cache from fastapi import FastAPI app FastAPI() lru_cache() def get_settings(): # 读取配置只执行一次 return {app_name: MyAPI} app.get(/settings/) async def read_settings(): return get_settings()十、总结FastAPI核心要点功能方法路径参数app.get(/items/{id})查询参数def read(q: str None)请求体def create(item: Item)依赖注入Depends()认证OAuth2PasswordBearer中间件app.middleware(http)FastAPI优势Python最快的Web框架自动生成API文档类型安全IDE友好异步原生支持