Skip to main content
fastapi-patterns FastAPI + SQLModel + Neon PostgreSQL 後端開發最佳實踐。適用於 Render 部署的 API 服務。涵蓋路由設計、資料庫操作、錯誤處理、認證授權。
Zur Installation springen Skills Marktplatz Entdecken und erkunden Sie KI-Skills, die von der Community erstellt wurden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Prompt-Details anzeigen Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/seikaikyo/dash-skills --skill fastapi-patternsDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
ZIP herunterladen Herunterladen... Mehr aus diesem Repository Use when reviewing code for security vulnerabilities, implementing authentication/authorization, handling user input, or discussing web application security. Covers OWASP Top 10:2025, ASVS 5.0, LLM Top 10 (2025), and Agentic AI security (2026).
Guides authoring of high-quality YARA-X detection rules for malware identification. Use when writing, reviewing, or optimizing YARA rules. Covers naming conventions, string selection, performance optimization, migration from legacy YARA, and false positive reduction. Triggers on: YARA, YARA-X, malware detection, threat hunting, IOC, signature, crx module, dex module.
Throwaway HTML mockups: 2-3 design variants to compare.
Basierend auf der SOC-Berufsklassifikation
name fastapi-patterns description FastAPI + SQLModel + Neon PostgreSQL 後端開發最佳實踐。適用於 Render 部署的 API 服務。涵蓋路由設計、資料庫操作、錯誤處理、認證授權。 source custom updated "2025-01-16T00:00:00.000Z"
FastAPI + SQLModel 後端開發規範
適用場景
Render 部署的 API 服務
搭配 Neon PostgreSQL
RESTful API 設計
MES/ERP 後端服務
專案結構
project/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 應用入口
│ ├── config.py # 環境設定
│ ├── database.py # 資料庫連線
│ ├── dependencies.py # 共用依賴
│ ├── models/ # SQLModel 資料模型
│ │ ├── __init__.py
│ │ ├── base.py
│ │ ├── work_order.py
│ │ └── user.py
│ ├── schemas/ # Pydantic 請求/回應 schema
│ │ ├── __init__.py
│ │ ├── base.py
│ │ └── work_order.py
│ ├── routers/ # API 路由
│ │ ├── __init__.py
│ │ ├── work_orders.py
│ │ └── users.py
│ ├── services/ # 業務邏輯
│ │ ├── __init__.py
│ │ └── work_order_service.py
│ └── utils/ # 工具函數
│ ├── __init__.py
│ └── audit.py
├── requirements.txt
├── render.yaml
└── .env.example
核心設定
config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings (BaseSettings
str
str
"/api"
bool
False
str
str
"HS256"
int
60
24
list
str
"*"
class
Config
".env"
@lru_cache
def
get_settings
return
database.py from sqlmodel import SQLModel, create_engine, Session
from app.config import get_settings
settings = get_settings()
engine = create_engine(
settings.database_url,
echo=settings.debug,
pool_pre_ping=True ,
pool_recycle=300 ,
)
def init_db ():
"""建立所有資料表"""
SQLModel.metadata.create_all(engine)
def get_session ():
"""取得資料庫 session（依賴注入用）"""
with Session(engine) as session:
yield session
main.py from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from app.config import get_settings
from app.database import init_db
from app.routers import work_orders, users
settings = get_settings()
@asynccontextmanager
async def lifespan (app: FastAPI ):
init_db()
yield
app = FastAPI(
title="MES API" ,
version="1.0.0" ,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True ,
allow_methods=["*" ],
allow_headers=["*" ],
)
app.include_router(work_orders.router, prefix=f"{settings.api_prefix} /work-orders" , tags=["工單" ])
app.include_router(users.router, prefix=f"{settings.api_prefix} /users" , tags=["使用者" ])
@app.get("/health"
def health_check ():
"""健康檢查（Render 用）"""
return {"status" : "ok" }
資料模型 (SQLModel)
models/base.py from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional
import uuid
class BaseModel (SQLModel ):
"""基礎模型"""
id : str = Field(default_factory=lambda : str (uuid.uuid4()), primary_key=True )
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: Optional [datetime] = Field(default=None )
class Config :
json_encoders = {
datetime: lambda v: v.isoformat()
}
models/work_order.py from sqlmodel import Field, Relationship
from typing import Optional , List
from datetime import datetime
from enum import Enum
from app.models.base import BaseModel
class WorkOrderStatus (str , Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
CANCELLED = "cancelled"
class WorkOrder (BaseModel, table=True ):
"""工單資料表"""
__tablename__ = "work_orders"
order_number: str = Field(unique=True , index=True )
customer_id: str = Field(foreign_key="customers.id" , index=True )
product_model: str
quantity: int = Field(ge=1 )
status: WorkOrderStatus = Field(default=WorkOrderStatus.PENDING, index=True )
priority: int = Field(default=5 , ge=1 , le=10 )
due_date: Optional [datetime] = None
note: Optional [str ] = None
dispatches: List ["Dispatch" ] = Relationship(back_populates="work_order" )
請求/回應 Schema
schemas/base.py from pydantic import BaseModel
from typing import TypeVar, Generic , Optional , List
T = TypeVar("T" )
class ApiResponse (BaseModel, Generic [T]):
"""統一回應格式"""
success: bool
data: Optional [T] = None
error: Optional [dict ] = None
message: Optional [str ] = None
class PaginatedResponse (ApiResponse[List [T]], Generic [T]):
"""分頁回應"""
pagination: Optional [dict ] = None
def success_response (data: T, message: str = "成功" ) -> ApiResponse[T]:
return ApiResponse(success=True , data=data, message=message)
def error_response (code: str , message: str ) -> ApiResponse:
return ApiResponse(success=False , error={"code" : code, "message" : message})
schemas/work_order.py from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from app.models.work_order import WorkOrderStatus
class WorkOrderCreate (BaseModel ):
"""新增工單請求"""
order_number: str = Field(..., min_length=1 , max_length=50 )
customer_id: str
product_model: str
quantity: int = Field(..., ge=1 )
priority: int = Field(default=5 , ge=1 , le=10 )
due_date: Optional [datetime] = None
note: Optional [str ] = None
class WorkOrderUpdate (BaseModel ):
"""更新工單請求"""
product_model: Optional [str ] = None
quantity: Optional [int ] = Field(default=None , ge=1 )
status: Optional [WorkOrderStatus] = None
priority: Optional [int ] = Field(default=None , ge=1 , le=10 )
due_date: Optional [datetime] = None
note: Optional [str ] = None
class WorkOrderResponse (BaseModel ):
"""工單回應"""
id : str
order_number: str
customer_id: str
product_model: str
quantity: int
status: WorkOrderStatus
priority: int
due_date: Optional [datetime]
note: Optional [str ]
created_at: datetime
updated_at: Optional [datetime]
class Config :
from_attributes = True
路由設計
routers/work_orders.py from fastapi import APIRouter, Depends, HTTPException, Query
from sqlmodel import Session, select
from typing import List , Optional
from datetime import datetime
from app.database import get_session
from app.models.work_order import WorkOrder, WorkOrderStatus
from app.schemas.base import ApiResponse, PaginatedResponse, success_response, error_response
from app.schemas.work_order import WorkOrderCreate, WorkOrderUpdate, WorkOrderResponse
from app.dependencies import get_current_user
from app.utils.audit import log_audit
router = APIRouter()
@router.get("" , response_model=PaginatedResponse[WorkOrderResponse]
def list_work_orders (
status: Optional [WorkOrderStatus] = None ,
customer_id: Optional [str ] = None ,
page: int = Query(1 , ge=1 int = Query(20 , ge=1 , le=100 get_session ),
):
"""取得工單列表"""
query = select(WorkOrder)
if status:
query = query.where(WorkOrder.status == status)
if customer_id:
query = query.where(WorkOrder.customer_id == customer_id)
total = session.exec (select(func.count()).select_from(query.subquery())).one()
offset = (page - 1 ) * limit
query = query.offset(offset).limit(limit).order_by(WorkOrder.created_at.desc())
items = session.exec (query).all ()
return PaginatedResponse(
success=True ,
data=[WorkOrderResponse.from_orm(item) for item in items],
pagination={"total" : total, "page" : page, "limit" : limit}
)
@router.get("/{work_order_id}" , response_model=ApiResponse[WorkOrderResponse]
def get_work_order (
work_order_id: str ,
session: Session = Depends(get_session ),
):
"""取得單一工單"""
work_order = session.get(WorkOrder, work_order_id)
if not work_order:
raise HTTPException(status_code=404 , detail="工單不存在" )
return success_response(WorkOrderResponse.from_orm(work_order))
@router.post("" , response_model=ApiResponse[WorkOrderResponse]
def create_work_order (
data: WorkOrderCreate,
session: Session = Depends(get_session ),
current_user: dict = Depends(get_current_user ),
):
"""新增工單"""
existing = session.exec (
select(WorkOrder).where(WorkOrder.order_number == data.order_number)
).first()
if existing:
raise HTTPException(status_code=400 , detail="工單編號已存在" )
work_order = WorkOrder(**data.model_dump())
session.add(work_order)
session.commit()
session.refresh(work_order)
log_audit(
session=session,
user_id=current_user["id" ],
action="CREATE" ,
resource_type="work_order" ,
resource_id=work_order.id ,
details=data.model_dump()
)
return success_response(WorkOrderResponse.from_orm(work_order), "工單建立成功" )
@router.put("/{work_order_id}" , response_model=ApiResponse[WorkOrderResponse]
def update_work_order (
work_order_id: str ,
data: WorkOrderUpdate,
session: Session = Depends(get_session ),
current_user: dict = Depends(get_current_user ),
):
"""更新工單"""
work_order = session.get(WorkOrder, work_order_id)
if not work_order:
raise HTTPException(status_code=404 , detail="工單不存在" )
update_data = data.model_dump(exclude_unset=True )
for key, value in update_data.items():
setattr (work_order, key, value)
work_order.updated_at = datetime.utcnow()
session.add(work_order)
session.commit()
session.refresh(work_order)
log_audit(
session=session,
user_id=current_user["id" ],
action="UPDATE" ,
resource_type="work_order" ,
resource_id=work_order.id ,
details=update_data
)
return success_response(WorkOrderResponse.from_orm(work_order), "工單更新成功" )
@router.delete("/{work_order_id}" , response_model=ApiResponse
def delete_work_order (
work_order_id: str ,
session: Session = Depends(get_session ),
current_user: dict = Depends(get_current_user ),
):
"""刪除工單"""
work_order = session.get(WorkOrder, work_order_id)
if not work_order:
raise HTTPException(status_code=404 , detail="工單不存在" )
session.delete(work_order)
session.commit()
log_audit(
session=session,
user_id=current_user["id" ],
action="DELETE" ,
resource_type="work_order" ,
resource_id=work_order_id,
)
return success_response(None , "工單刪除成功" )
認證授權
dependencies.py from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from datetime import datetime
from app.config import get_settings
settings = get_settings()
security = HTTPBearer()
def get_current_user (
credentials: HTTPAuthorizationCredentials = Depends(security )
) -> dict :
"""驗證 JWT Token 並取得當前使用者"""
token = credentials.credentials
try :
payload = jwt.decode(
token,
settings.jwt_secret,
algorithms=[settings.jwt_algorithm]
)
exp = payload.get("exp" )
if exp and datetime.utcnow().timestamp() > exp:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token 已過期"
)
return payload
except jwt.InvalidTokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="無效的 Token"
)
def require_role (allowed_roles: list [str ] ):
"""角色權限檢查"""
def role_checker (current_user: dict = Depends(get_current_user ) ):
user_role = current_user.get("role" , "" )
if user_role not in allowed_roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="權限不足"
)
return current_user
return role_checker
稽核紀錄 (ISO 27001)
utils/audit.py from sqlmodel import Session
from datetime import datetime
from typing import Optional
import json
from app.models.audit import AuditLog
def log_audit (
session: Session,
user_id: str ,
action: str ,
resource_type: str ,
resource_id: str ,
details: Optional [dict ] = None ,
ip_address: Optional [str ] = None ,
):
"""記錄稽核日誌"""
audit_log = AuditLog(
user_id=user_id,
action=action,
resource_type=resource_type,
resource_id=resource_id,
details=json.dumps(details) if details else None ,
ip_address=ip_address,
timestamp=datetime.utcnow(),
)
session.add(audit_log)
session.commit()
錯誤處理
全域例外處理 from fastapi import Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError
@app.exception_handler(HTTPException )
async def http_exception_handler (request: Request, exc: HTTPException ):
return JSONResponse(
status_code=exc.status_code,
content={
"success" : False ,
"error" : {
"code" : str (exc.status_code),
"message" : exc.detail
}
}
)
@app.exception_handler(IntegrityError )
async def integrity_error_handler (request: Request, exc: IntegrityError ):
return JSONResponse(
status_code=400 ,
content={
"success" : False ,
"error" : {
"code" : "INTEGRITY_ERROR" ,
"message" : "資料完整性錯誤，可能是重複的資料"
}
}
)
@app.exception_handler(Exception )
async def general_exception_handler (request: Request, exc: Exception ):
print (f"Unhandled error: {exc} " )
return JSONResponse(
status_code=500 ,
content={
"success" : False ,
"error" : {
"code" : "INTERNAL_ERROR" ,
"message" : "伺服器內部錯誤"
}
}
)
Render 部署
render.yaml services:
- type: web
name: mes-api
runtime: python
buildCommand: pip install -r requirements.txt
startCommand: uvicorn app.main:app --host 0.0 .0 .0 --port $PORT
envVars:
- key: DATABASE_URL
fromDatabase:
name: mes-db
property: connectionString
- key: JWT_SECRET
generateValue: true
healthCheckPath: /health
autoDeploy: true
databases:
- name: mes-db
plan: free
databaseName: mes
user: mes_user
requirements.txt fastapi>=0.109.0
uvicorn[standard]>=0.27.0
sqlmodel>=0.0.14
pydantic-settings>=2.1.0
python-jose[cryptography]>=3.3.0
psycopg2-binary>=2.9.9
禁止事項
禁止 Fallback Mock 資料 - 資料庫連線失敗應回傳 503禁止 Emoji - 程式碼、註解都不使用禁止簡體字 - 錯誤訊息使用正體中文禁止硬編碼機敏資料 - 使用環境變數
效能標準 指標 目標 API 回應時間 < 500ms 資料庫查詢 < 100ms 啟動時間 < 30s
參考資源 