FastAPI 小白教程：從入門級到實戰(源碼教程)

1. FastAPI 基本介紹

安裝 FastAPI

2. 簡單的 CRUD 示例

2.1 創建基本應用

2.2 添加 CRUD 操作???????

3. 處理跨域請求 (CORS)

4. 普通案例：待辦事項 API???????

5. 企業案例：認證和數據庫集成

5.1 使用 SQLAlchemy 和 JWT 認證???????

6. 常見問題

6.1 性能問題

6.2 異步支持

6.3 數據庫遷移

6.4 部署問題

6.5 測試

6.6 文件上傳

1. FastAPI 基本介紹

FastAPI 是一個現代、快速（高性能）的 Web 框架，用于基于 Python 構建 API。它具有以下特點：

快速：與 NodeJS 和 Go 相當的高性能
快速編碼：開發功能的速度提高約 200%-300%
更少的錯誤：減少約 40% 的人為錯誤
直觀：強大的編輯器支持
簡單：易于使用和學習
簡短：最小化代碼重復
穩健：生產就緒的代碼
基于標準：完全兼容 API 開放標準（如 OpenAPI 和 JSON Schema）

安裝 FastAPI

pip install fastapipip install uvicorn  # ASGI 服務器

2. 簡單的 CRUD 示例

讓我們創建一個簡單的用戶管理 API，包含創建、讀取、更新和刪除操作。

2.1 創建基本應用???????

from fastapi import FastAPIapp = FastAPI()# 模擬數據庫fake_db = []@app.get("/")def read_root():    return {"message": "Welcome to FastAPI!"}

運行應用：

uvicorn main:app --reload

2.2 添加 CRUD 操作???????

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import Listapp = FastAPI()# 用戶模型class User(BaseModel):    id: int    name: str    email: str# 模擬數據庫fake_db: List[User] = []# 創建用戶@app.post("/users/", response_model=User)def create_user(user: User):    fake_db.append(user)    return user# 獲取所有用戶@app.get("/users/", response_model=List[User])def read_users():    return fake_db# 獲取單個用戶@app.get("/users/{user_id}", response_model=User)def read_user(user_id: int):    for user in fake_db:        if user.id == user_id:            return user    raise HTTPException(status_code=404, detail="User not found")# 更新用戶@app.put("/users/{user_id}", response_model=User)def update_user(user_id: int, updated_user: User):    for index, user in enumerate(fake_db):        if user.id == user_id:            fake_db[index] = updated_user            return updated_user    raise HTTPException(status_code=404, detail="User not found")# 刪除用戶@app.delete("/users/{user_id}")def delete_user(user_id: int):    for index, user in enumerate(fake_db):        if user.id == user_id:            del fake_db[index]            return {"message": "User deleted"}    raise HTTPException(status_code=404, detail="User not found")

3. 處理跨域請求 (CORS)

當你的前端應用運行在不同的域名下時，需要處理跨域請求：???????

from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()# 設置允許的源origins = [    "http://localhost",    "http://localhost:8080",    "http://localhost:3000",]# 添加中間件app.add_middleware(    CORSMiddleware,    allow_origins=origins,    allow_credentials=True,    allow_methods=["*"],    allow_headers=["*"],)

4. 普通案例：待辦事項 API???????

from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelfrom typing import List, Optionalapp = FastAPI()class Todo(BaseModel):    id: int    title: str    description: Optional[str] = None    completed: bool = Falsetodos = []@app.post("/todos/", response_model=Todo)def create_todo(todo: Todo):    todos.append(todo)    return todo@app.get("/todos/", response_model=List[Todo])def read_todos(completed: Optional[bool] = None):    if completed is None:        return todos    return [todo for todo in todos if todo.completed == completed]@app.put("/todos/{todo_id}", response_model=Todo)def update_todo(todo_id: int, updated_todo: Todo):    for index, todo in enumerate(todos):        if todo.id == todo_id:            todos[index] = updated_todo            return updated_todo    raise HTTPException(status_code=404, detail="Todo not found")@app.delete("/todos/{todo_id}")def delete_todo(todo_id: int):    for index, todo in enumerate(todos):        if todo.id == todo_id:            del todos[index]            return {"message": "Todo deleted"}    raise HTTPException(status_code=404, detail="Todo not found")

5. 企業案例：認證和數據庫集成

5.1 使用 SQLAlchemy 和 JWT 認證???????

from fastapi import FastAPI, Depends, HTTPException, statusfrom fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormfrom pydantic import BaseModelfrom datetime import datetime, timedeltafrom jose import JWTError, jwtfrom passlib.context import CryptContextfrom typing import Optionalfrom sqlalchemy import create_engine, Column, Integer, String, Booleanfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmaker, Session# 數據庫配置SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"engine = create_engine(SQLALCHEMY_DATABASE_URL)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()# 數據庫模型class DBUser(Base):    __tablename__ = "users"    id = Column(Integer, primary_key=True, index=True)    username = Column(String, unique=True, index=True)    hashed_password = Column(String)    is_active = Column(Boolean, default=True)Base.metadata.create_all(bind=engine)# JWT 配置SECRET_KEY = "your-secret-key"ALGORITHM = "HS256"ACCESS_TOKEN_EXPIRE_MINUTES = 30class Token(BaseModel):    access_token: str    token_type: strclass TokenData(BaseModel):    username: Optional[str] = Noneclass User(BaseModel):    username: str    is_active: Optional[bool] = Noneclass UserInDB(User):    hashed_password: strpwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")app = FastAPI()# 數據庫依賴def get_db():    db = SessionLocal()    try:        yield db    finally:        db.close()# 密碼哈希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 authenticate_user(db: Session, username: str, password: str):    user = db.query(DBUser).filter(DBUser.username == username).first()    if not user:        return False    if not verify_password(password, user.hashed_password):        return False    return user# 創建 tokendef create_access_token(data: dict, expires_delta: Optional[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), db: Session = Depends(get_db)):    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        token_data = TokenData(username=username)    except JWTError:        raise credentials_exception    user = db.query(DBUser).filter(DBUser.username == token_data.username).first()    if user is None:        raise credentials_exception    return user# 路由@app.post("/token", response_model=Token)async def login_for_access_token(    form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):    user = authenticate_user(db, 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"}@app.get("/users/me/", response_model=User)async def read_users_me(current_user: User = Depends(get_current_user)):    return current_user@app.post("/users/", response_model=User)def create_user(user: User, db: Session = Depends(get_db)):    hashed_password = get_password_hash(user.password)    db_user = DBUser(username=user.username, hashed_password=hashed_password)    db.add(db_user)    db.commit()    db.refresh(db_user)    return db_user

6. 常見問題

6.1 性能問題

Q: FastAPI 真的比 Flask 快嗎？

A: 是的，FastAPI 基于 Starlette（用于 ASGI）和 Pydantic，性能接近 NodeJS 和 Go。根據基準測試，FastAPI 比 Flask 快得多，特別是在處理大量請求時。

6.2 異步支持

Q: 如何在 FastAPI 中正確使用 async/await？

A: FastAPI 完全支持異步。對于 I/O 密集型操作（如數據庫查詢、API 調用），使用 async def：???????

@app.get("/items/{item_id}")async def read_item(item_id: int):    item = await some_async_function(item_id)    return item

6.3 數據庫遷移

Q: 如何管理數據庫遷移？

A: 推薦使用 Alembic：???????

pip install alembicalembic init migrations

然后配置?alembic.ini?和?migrations/env.py?文件。

6.4 部署問題

Q: 如何部署 FastAPI 應用？

A: 常見部署方式：

使用 Uvicorn 或 Hypercorn 作為 ASGI 服務器
使用 Gunicorn 作為進程管理器（配合 Uvicorn worker）
使用 Docker 容器化
部署到云平臺（如 AWS, GCP, Azure）

6.5 測試

Q: 如何測試 FastAPI 應用？

A: 使用?TestClient：???????

from fastapi.testclient import TestClientclient = TestClient(app)def test_read_main():    response = client.get("/")    assert response.status_code == 200    assert response.json() == {"message": "Welcome to FastAPI!"}

6.6 文件上傳

Q: 如何處理文件上傳？

A: 使用?File?和?UploadFile：???????

from fastapi import FastAPI, File, UploadFile@app.post("/uploadfile/")async def create_upload_file(file: UploadFile = File(...)):    return {"filename": file.filename}

FastAPI 是一個強大而現代的 Python Web 框架，適合構建高性能的 API。它結合了 Python 類型提示、自動文檔生成和異步支持，使得開發體驗非常愉快。通過本教程，你應該已經掌握了 FastAPI 的基本用法、CRUD 操作、跨域處理、認證授權等核心功能。

* Thanks you *

如果覺得文章內容不錯，隨手幫忙點個贊、在看、轉發一下，如果想第一時間收到推送，也可以給我個星標?～謝謝你看我的文章。

*往期推薦?*

實現如何利用 Kafka 延時刪除用戶郵箱的驗證碼（如何發送郵箱+源碼) - 第一期

Docker小白入門教程一篇領你入門(CRUD全命令+無廢話版+問題集)-第三期

Docker小白入門教程一篇領你入門(CRUD全命令+無廢話版+問題集)

想要高效處理，那不妨看看 Python的異步 Asyncio 保證效率翻多倍

銀河麒麟 | ubuntu 安裝國產達夢DM8數據庫(安裝+外網通+IDEA連接)

網絡設備日志存儲到指定的Kiwi-log服務器(圖解+軟件)

銀河麒麟 | ubuntu 安裝運用 docker 容器，實現容器化部署項目

銀河麒麟 | ubuntu 安裝zabbix監控設備信息(親測包對)