後端攔截器、中間件、過濾器、裝飾器區別與使用指南

// middleware/logger.middleware.ts
import { Injectable, NestMiddleware } from '@nestjs/common';
import { Request, Response, NextFunction } from 'express';

@Injectable()
export class LoggerMiddleware implements NestMiddleware {
  use(req: Request, res: Response, next: NextFunction) {
    const startTime = Date.now();
    const { method, originalUrl } = req;
    
    res.on('finish', () => {
      const duration = Date.now() - startTime;
      console.log(`${method} ${originalUrl} ${res.statusCode} - ${duration}ms`);
    });
    
    next();
  }
}

// app.module.ts
import { Module, NestModule, MiddlewareConsumer } from '@nestjs/common';

@Module({})
export class AppModule implements NestModule {
  configure(consumer: MiddlewareConsumer) {
    consumer
      .apply(LoggerMiddleware)
      .forRoutes('*'); // 應用於所有路由
  }
}

import { Request, Response, NextFunction } from 'express';

export function logger(req: Request, res: Response, next: NextFunction) {
  const startTime = Date.now();
  const { method, originalUrl } = req;
  
  res.on('finish', () => {
    const duration = Date.now() - startTime;
    console.log(`${method} ${originalUrl} ${res.statusCode} - ${duration}ms`);
  });
  
  next();
}

// 使用
consumer.apply(logger).forRoutes('*');

// main.ts
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module';
import { logger } from './middleware/logger.middleware';

async function bootstrap() {
  const app = await NestFactory.create(AppModule);
  
  // 全局中間件
  app.use(logger);
  
  await app.listen(3000);
}
bootstrap();

// interceptors/logging.interceptor.ts
import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { Observable } from 'rxjs';
import { tap } from 'rxjs/operators';

@Injectable()
export class LoggingInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const request = context.switchToHttp().getRequest();
    const now = Date.now();
    
    return next.handle().pipe(
      tap(() => {
        const duration = Date.now() - now;
        console.log(`${request.method} ${request.url} - ${duration}ms`);
      })
    );
  }
}

// 使用方式 1：全局註冊
// main.ts
app.useGlobalInterceptors(new LoggingInterceptor());

// 使用方式 2：控制器級別
@Controller('users')
@UseInterceptors(LoggingInterceptor)
export class UsersController {}

// 使用方式 3：方法級別
@Get()
@UseInterceptors(LoggingInterceptor)
findAll() {}

@Injectable()
export class TransformInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(
      map(data => ({
        success: true,
        data,
        timestamp: new Date().toISOString()
      }))
    );
  }
}

import { Injectable, NestInterceptor, ExecutionContext, CallHandler, RequestTimeoutException } from '@nestjs/common';
import { Observable, throwError, TimeoutError } from 'rxjs';
import { timeout, catchError } from 'rxjs/operators';

@Injectable()
export class TimeoutInterceptor implements NestInterceptor {
  constructor(private readonly timeoutMs: number = 5000) {}

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    return next.handle().pipe(
      timeout(this.timeoutMs),
      catchError(err => {
        if (err instanceof TimeoutError) {
          return throwError(() => new RequestTimeoutException('Request timeout'));
        }
        return throwError(() => err);
      })
    );
  }
}

// 使用
app.useGlobalInterceptors(new TimeoutInterceptor(10000)); // 10 秒超時

import { Injectable, NestInterceptor, ExecutionContext, CallHandler, Inject } from '@nestjs/common';
import { Observable, of } from 'rxjs';
import { tap } from 'rxjs/operators';
import { CACHE_MANAGER } from '@nestjs/cache-manager';
import { Cache } from 'cache-manager';

@Injectable()
export class CacheInterceptor implements NestInterceptor {
  constructor(
    @Inject(CACHE_MANAGER) private cacheManager: Cache,
    private readonly ttl: number = 60000 // 60 秒
  ) {}

  async intercept(context: ExecutionContext, next: CallHandler): Promise<Observable<any>> {
    const request = context.switchToHttp().getRequest();
    const cacheKey = `cache:${request.method}:${request.url}`;
    
    // 檢查緩存
    const cached = await this.cacheManager.get(cacheKey);
    if (cached) {
      return of(cached);
    }

    // 執行請求並緩存結果
    return next.handle().pipe(
      tap(async (data) => {
        await this.cacheManager.set(cacheKey, data, this.ttl);
      })
    );
  }
}

import { Injectable, NestInterceptor, ExecutionContext, CallHandler, Logger } from '@nestjs/common';
import { Observable } from 'rxjs';
import { tap } from 'rxjs/operators';

@Injectable()
export class PerformanceInterceptor implements NestInterceptor {
  private readonly logger = new Logger(PerformanceInterceptor.name);

  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const request = context.switchToHttp().getRequest();
    const { method, url } = request;
    const startTime = Date.now();

    return next.handle().pipe(
      tap({
        next: () => {
          const duration = Date.now() - startTime;
          this.logger.log(
            `${method} ${url} - ${duration}ms`,
          );
          
          // 記錄慢請求
          if (duration > 1000) {
            this.logger.warn(`Slow request detected: ${method} ${url} took ${duration}ms`);
          }
        },
        error: (error) => {
          const duration = Date.now() - startTime;
          this.logger.error(
            `${method} ${url} - ${duration}ms - Error: ${error.message}`,
          );
        },
      })
    );
  }
}

// filters/http-exception.filter.ts
import { ExceptionFilter, Catch, ArgumentsHost, HttpException } from '@nestjs/common';
import { Request, Response } from 'express';

@Catch(HttpException)
export class HttpExceptionFilter implements ExceptionFilter {
  catch(exception: HttpException, host: ArgumentsHost) {
    const ctx = host.switchToHttp();
    const response = ctx.getResponse<Response>();
    const request = ctx.getRequest<Request>();
    const status = exception.getStatus();
    const exceptionResponse = exception.getResponse();

    response.status(status).json({
      statusCode: status,
      timestamp: new Date().toISOString(),
      path: request.url,
      message: exceptionResponse['message'] || exception.message
    });
  }
}

// 使用
@UseFilters(HttpExceptionFilter)
@Controller('users')
export class UsersController {}

// main.ts
app.useGlobalFilters(new HttpExceptionFilter());

// filters/all-exceptions.filter.ts
import {
  ExceptionFilter,
  Catch,
  ArgumentsHost,
  HttpException,
  HttpStatus,
  Logger,
} from '@nestjs/common';
import { Request, Response } from 'express';

@Catch()
export class AllExceptionsFilter implements ExceptionFilter {
  private readonly logger = new Logger(AllExceptionsFilter.name);

  catch(exception: unknown, host: ArgumentsHost) {
    const ctx = host.switchToHttp();
    const response = ctx.getResponse<Response>();
    const request = ctx.getRequest<Request>();

    const status =
      exception instanceof HttpException
        ? exception.getStatus()
        : HttpStatus.INTERNAL_SERVER_ERROR;

    const message =
      exception instanceof HttpException
        ? exception.getResponse()
        : 'Internal server error';

    const errorResponse = {
      statusCode: status,
      timestamp: new Date().toISOString(),
      path: request.url,
      method: request.method,
      message: typeof message === 'string' ? message : (message as any).message,
      requestId: request.headers['x-request-id'] || 'unknown',
    };

    this.logger.error(
      `${request.method} ${request.url} - ${status}`,
      exception instanceof Error ? exception.stack : exception,
    );

    response.status(status).json(errorResponse);
  }
}

// decorators/user.decorator.ts
import { createParamDecorator, ExecutionContext } from '@nestjs/common';

// 獲取當前用戶
export const CurrentUser = createParamDecorator(
  (data: unknown, ctx: ExecutionContext) => {
    const request = ctx.switchToHttp().getRequest();
    return request.user;
  }
);

// 使用
@Get('profile')
getProfile(@CurrentUser() user: User) {
  return user;
}

// decorators/roles.decorator.ts
import { SetMetadata } from '@nestjs/common';

export const Roles = (...roles: string[]) => SetMetadata('roles', roles);

// guards/roles.guard.ts
import { Injectable, CanActivate, ExecutionContext } from '@nestjs/common';
import { Reflector } from '@nestjs/core';

@Injectable()
export class RolesGuard implements CanActivate {
  constructor(private reflector: Reflector) {}

  canActivate(context: ExecutionContext): boolean {
    const requiredRoles = this.reflector.get<string[]>('roles', context.getHandler());
    if (!requiredRoles) {
      return true;
    }
    const request = context.switchToHttp().getRequest();
    const user = request.user;
    return requiredRoles.some(role => user.roles?.includes(role));
  }
}

// 使用
@Get('admin')
@Roles('admin')
@UseGuards(RolesGuard)
getAdminData() {
  return { message: 'Admin data' };
}

// middleware/request-id.middleware.ts
import { Injectable, NestMiddleware } from '@nestjs/common';
import { Request, Response, NextFunction } from 'express';
import { v4 as uuidv4 } from 'uuid';

@Injectable()
export class RequestIdMiddleware implements NestMiddleware {
  use(req: Request, res: Response, next: NextFunction) {
    const requestId = req.headers['x-request-id'] as string || uuidv4();
    req.headers['x-request-id'] = requestId;
    res.setHeader('x-request-id', requestId);
    next();
  }
}

// 獲取請求 ID 的裝飾器
export const RequestId = createParamDecorator(
  (data: unknown, ctx: ExecutionContext) => {
    const request = ctx.switchToHttp().getRequest();
    return request.headers['x-request-id'];
  }
);

// Program.cs (.NET 10+)
var builder = WebApplication.CreateBuilder(args);

// 添加服務
builder.Services.AddControllers();
builder.Services.AddEndpointsApiExplorer();
builder.Services.AddSwaggerGen();

var app = builder.Build();

// 內聯中間件（.NET 10 效能優化）
app.Use(async (context, next) =>
{
    var startTime = DateTime.UtcNow;
    
    await next();
    
    var duration = DateTime.UtcNow - startTime;
    Console.WriteLine($"{context.Request.Method} {context.Request.Path} - {context.Response.StatusCode} - {duration.TotalMilliseconds}ms");
});

// 使用擴展方法註冊中間件（.NET 10 推薦）
app.UseRequestLogging();
app.UseMiddleware<CustomMiddleware>();

app.MapControllers();
app.Run();

// Middleware/PerformanceMiddleware.cs (.NET 10)
public class PerformanceMiddleware
{
    private readonly RequestDelegate _next;
    private readonly ILogger<PerformanceMiddleware> _logger;

    public PerformanceMiddleware(RequestDelegate next, ILogger<PerformanceMiddleware> logger)
    {
        _next = next;
        _logger = logger;
    }

    public async Task InvokeAsync(HttpContext context)
    {
        var sw = System.Diagnostics.Stopwatch.StartNew();
        
        await _next(context);
        
        sw.Stop();
        
        // .NET 10 新增的結構化日誌改進
        _logger.LogInformation(
            "Request {Method} {Path} completed in {Duration}ms with status {StatusCode}",
            context.Request.Method,
            context.Request.Path,
            sw.ElapsedMilliseconds,
            context.Response.StatusCode
        );
    }
}

// 擴展方法
public static class PerformanceMiddlewareExtensions
{
    public static IApplicationBuilder UsePerformanceLogging(this IApplicationBuilder builder)
    {
        return builder.UseMiddleware<PerformanceMiddleware>();
    }
}

// Startup.cs
public void Configure(IApplicationBuilder app)
{
    app.Use(async (context, next) =>
    {
        // 請求前處理
        await next();
        // 響應後處理
    });
    
    app.UseMiddleware<CustomMiddleware>();
}

// Middleware/RequestLoggingMiddleware.cs
public class RequestLoggingMiddleware
{
    private readonly RequestDelegate _next;
    private readonly ILogger<RequestLoggingMiddleware> _logger;

    public RequestLoggingMiddleware(RequestDelegate next, ILogger<RequestLoggingMiddleware> logger)
    {
        _next = next;
        _logger = logger;
    }

    public async Task InvokeAsync(HttpContext context)
    {
        var startTime = DateTime.UtcNow;
        
        await _next(context);
        
        var duration = DateTime.UtcNow - startTime;
        _logger.LogInformation(
            "Request {Method} {Path} responded {StatusCode} in {Duration}ms",
            context.Request.Method,
            context.Request.Path,
            context.Response.StatusCode,
            duration.TotalMilliseconds
        );
    }
}

// 擴展方法
public static class RequestLoggingMiddlewareExtensions
{
    public static IApplicationBuilder UseRequestLogging(this IApplicationBuilder builder)
    {
        return builder.UseMiddleware<RequestLoggingMiddleware>();
    }
}

// 使用
app.UseRequestLogging();

// Filters/LoggingActionFilter.cs
using Microsoft.AspNetCore.Mvc.Filters;
using Microsoft.Extensions.Logging;

public class LoggingActionFilter : IActionFilter
{
    private readonly ILogger<LoggingActionFilter> _logger;

    public LoggingActionFilter(ILogger<LoggingActionFilter> logger)
    {
        _logger = logger;
    }

    public void OnActionExecuting(ActionExecutingContext context)
    {
        var actionName = context.ActionDescriptor.DisplayName;
        var startTime = DateTime.UtcNow;
        context.HttpContext.Items["ActionStartTime"] = startTime;
        
        _logger.LogInformation("Action {Action} is executing", actionName);
    }

    public void OnActionExecuted(ActionExecutedContext context)
    {
        var actionName = context.ActionDescriptor.DisplayName;
        var startTime = (DateTime)context.HttpContext.Items["ActionStartTime"]!;
        var duration = DateTime.UtcNow - startTime;
        
        _logger.LogInformation(
            "Action {Action} executed in {Duration}ms",
            actionName,
            duration.TotalMilliseconds
        );
    }
}

// 使用方式 1：全局註冊（.NET 10）
var builder = WebApplication.CreateBuilder(args);

// .NET 10 改進的服務註冊方式
builder.Services.AddControllers(options =>
{
    options.Filters.Add<LoggingActionFilter>();
});

// 或使用服務過濾器（支援依賴注入）
builder.Services.AddScoped<LoggingActionFilter>();
builder.Services.AddControllers(options =>
{
    options.Filters.AddService<LoggingActionFilter>();
});

var app = builder.Build();
app.MapControllers();
app.Run();

// 使用方式 2：控制器級別
[TypeFilter(typeof(LoggingActionFilter))]
public class UsersController : ControllerBase {}

// 使用方式 3：方法級別
[ServiceFilter(typeof(LoggingActionFilter))]
[HttpGet]
public IActionResult GetUsers() {}

// Filters/AsyncLoggingActionFilter.cs
public class AsyncLoggingActionFilter : IAsyncActionFilter
{
    private readonly ILogger<AsyncLoggingActionFilter> _logger;

    public AsyncLoggingActionFilter(ILogger<AsyncLoggingActionFilter> logger)
    {
        _logger = logger;
    }

    public async Task OnActionExecutionAsync(
        ActionExecutingContext context,
        ActionExecutionDelegate next)
    {
        var startTime = DateTime.UtcNow;
        var actionName = context.ActionDescriptor.DisplayName;
        
        _logger.LogInformation("Action {Action} starting", actionName);
        
        var executedContext = await next();
        
        var duration = DateTime.UtcNow - startTime;
        _logger.LogInformation(
            "Action {Action} completed in {Duration}ms",
            actionName,
            duration.TotalMilliseconds
        );
    }
}

// Filters/GlobalExceptionFilter.cs
using Microsoft.AspNetCore.Mvc.Filters;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Hosting;

public class GlobalExceptionFilter : IExceptionFilter
{
    private readonly ILogger<GlobalExceptionFilter> _logger;
    private readonly IHostEnvironment _environment;

    public GlobalExceptionFilter(
        ILogger<GlobalExceptionFilter> logger,
        IHostEnvironment environment)
    {
        _logger = logger;
        _environment = environment;
    }

    public void OnException(ExceptionContext context)
    {
        _logger.LogError(
            context.Exception,
            "Unhandled exception: {Message}",
            context.Exception.Message
        );

        var response = new
        {
            error = new
            {
                message = context.Exception.Message,
                type = context.Exception.GetType().Name,
                stackTrace = _environment.IsDevelopment()
                    ? context.Exception.StackTrace
                    : null,
                path = context.HttpContext.Request.Path,
                timestamp = DateTime.UtcNow
            }
        };

        context.Result = new ObjectResult(response)
        {
            StatusCode = context.Exception switch
            {
                ArgumentException => 400,
                UnauthorizedAccessException => 401,
                _ => 500
            }
        };

        context.ExceptionHandled = true;
    }
}

// 異步異常過濾器（推薦）
public class AsyncGlobalExceptionFilter : IAsyncExceptionFilter
{
    private readonly ILogger<AsyncGlobalExceptionFilter> _logger;
    private readonly IHostEnvironment _environment;

    public AsyncGlobalExceptionFilter(
        ILogger<AsyncGlobalExceptionFilter> logger,
        IHostEnvironment environment)
    {
        _logger = logger;
        _environment = environment;
    }

    public Task OnExceptionAsync(ExceptionContext context)
    {
        OnException(context);
        return Task.CompletedTask;
    }

    private void OnException(ExceptionContext context)
    {
        // 同上實作
    }
}

// Filters/CustomAuthorizationFilter.cs
public class CustomAuthorizationFilter : IAuthorizationFilter
{
    public void OnAuthorization(AuthorizationFilterContext context)
    {
        var user = context.HttpContext.User;
        if (!user.Identity.IsAuthenticated)
        {
            context.Result = new UnauthorizedResult();
        }
    }
}

// Attributes/AuthorizeRoleAttribute.cs
[AttributeUsage(AttributeTargets.Method | AttributeTargets.Class)]
public class AuthorizeRoleAttribute : Attribute, IAuthorizationFilter
{
    private readonly string[] _roles;

    public AuthorizeRoleAttribute(params string[] roles)
    {
        _roles = roles;
    }

    public void OnAuthorization(AuthorizationFilterContext context)
    {
        var user = context.HttpContext.User;
        var userRoles = user.Claims
            .Where(c => c.Type == ClaimTypes.Role)
            .Select(c => c.Value);

        if (!_roles.Any(role => userRoles.Contains(role)))
        {
            context.Result = new ForbidResult();
        }
    }
}

// 使用
[AuthorizeRole("Admin", "Manager")]
[HttpGet("admin")]
public IActionResult GetAdminData()
{
    return Ok(new { message = "Admin data" });
}

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# 類別實作中間件
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        response = await call_next(request)
        
        duration = time.time() - start_time
        print(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
        
        return response

# 註冊中間件
app.add_middleware(LoggingMiddleware)

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    start_time = time.time()
    
    response = await call_next(request)
    
    duration = time.time() - start_time
    print(f"{request.method} {request.url.path} - {response.status_code} - {duration:.3f}s")
    
    return response

from fastapi import HTTPException, status, Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Optional

class AuthMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, public_paths: Optional[list] = None):
        super().__init__(app)
        self.public_paths = public_paths or ["/docs", "/openapi.json", "/public"]
    
    async def dispatch(self, request: Request, call_next):
        # 跳過公開路由
        if any(request.url.path.startswith(path) for path in self.public_paths):
            return await call_next(request)
        
        # 驗證 token
        authorization = request.headers.get("Authorization")
        if not authorization or not authorization.startswith("Bearer "):
            return JSONResponse(
                status_code=status.HTTP_401_UNAUTHORIZED,
                content={"detail": "Not authenticated"}
            )
        
        # 驗證邏輯
        token = authorization.replace("Bearer ", "")
        if not self.verify_token(token):
            return JSONResponse(
                status_code=status.HTTP_401_UNAUTHORIZED,
                content={"detail": "Invalid token"}
            )
        
        # 將用戶信息添加到請求狀態
        request.state.user = self.get_user_from_token(token)
        
        return await call_next(request)
    
    def verify_token(self, token: str) -> bool:
        # Token 驗證邏輯（實際應使用 JWT 等）
        return len(token) > 0
    
    def get_user_from_token(self, token: str) -> dict:
        # 從 token 獲取用戶信息（實際應解析 JWT）
        return {"id": 1, "username": "user"}

# 使用
app.add_middleware(AuthMiddleware, public_paths=["/docs", "/public"])

from fastapi.middleware.cors import CORSMiddleware

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

from fastapi import Depends, FastAPI, Header, HTTPException, status, Request
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# 用戶模型
class User(BaseModel):
    id: int
    username: str
    email: Optional[str] = None

# 依賴函數（同步）
def get_current_user(authorization: str = Header(None)) -> User:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated"
        )
    
    token = authorization.replace("Bearer ", "")
    # 驗證 token 並返回用戶
    return User(id=1, username="user", email="user@example.com")

# 異步依賴函數（推薦）
async def get_current_user_async(
    authorization: str = Header(None)
) -> User:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated"
        )
    
    token = authorization.replace("Bearer ", "")
    # 異步驗證 token（例如從資料庫查詢）
    # await verify_token(token)
    return User(id=1, username="user", email="user@example.com")

# 使用依賴
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user_async)):
    return current_user

from typing import Generator
import time
import logging

logger = logging.getLogger(__name__)

def log_request_duration() -> Generator[None, None, None]:
    start_time = time.time()
    logger.info("Request started")
    
    try:
        yield
    finally:
        duration = time.time() - start_time
        logger.info(f"Request completed in {duration:.3f}s")

@app.get("/items")
async def get_items(_: None = Depends(log_request_duration)):
    return {"items": []}

from typing import Generator
from sqlalchemy.orm import Session
from database import SessionLocal

def get_db() -> Generator[Session, None, None]:
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users")
async def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

from fastapi import HTTPException, status, Depends
from typing import List, Callable
from functools import wraps

def require_roles(required_roles: List[str]):
    def role_checker(current_user: User = Depends(get_current_user_async)) -> User:
        user_roles = getattr(current_user, "roles", [])
        if not any(role in user_roles for role in required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user
    return role_checker

# 使用
@app.get("/admin")
async def get_admin_data(
    user: User = Depends(require_roles(["admin", "manager"]))
):
    return {"message": "Admin data", "user": user}

# 可重用的權限依賴類
class RoleChecker:
    def __init__(self, required_roles: List[str]):
        self.required_roles = required_roles
    
    def __call__(self, current_user: User = Depends(get_current_user_async)) -> User:
        user_roles = getattr(current_user, "roles", [])
        if not any(role in user_roles for role in self.required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return current_user

# 使用類別方式
admin_checker = RoleChecker(["admin"])
manager_checker = RoleChecker(["admin", "manager"])

@app.get("/admin-only")
async def admin_only(user: User = Depends(admin_checker)):
    return {"message": "Admin only"}

from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

# 全局異常處理器
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "status_code": exc.status_code,
                "message": exc.detail,
                "path": request.url.path
            }
        }
    )

# 驗證異常處理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": {
                "status_code": 422,
                "message": "Validation error",
                "details": exc.errors()
            }
        }
    )

# 通用異常處理器
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": {
                "status_code": 500,
                "message": "Internal server error",
                "detail": str(exc) if app.debug else None
            }
        }
    )

from fastapi import HTTPException

class CustomException(HTTPException):
    def __init__(self, detail: str, status_code: int = 400):
        super().__init__(status_code=status_code, detail=detail)

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail}
    )

# 使用
@app.get("/test")
async def test():
    raise CustomException("Something went wrong", status_code=400)

from fastapi import APIRouter, Depends
from functools import wraps

router = APIRouter()

# 自定義裝飾器
def require_auth(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # 認證邏輯
        if not kwargs.get("current_user"):
            raise HTTPException(status_code=401, detail="Not authenticated")
        return await func(*args, **kwargs)
    return wrapper

# 使用
@router.get("/protected")
@require_auth
async def protected_route(current_user: dict = Depends(get_current_user)):
    return {"message": "Protected data"}

from functools import wraps
from datetime import datetime, timedelta
from fastapi import HTTPException, status, Request
from typing import Optional
import redis.asyncio as redis

# 使用 Redis 進行速率限制（生產環境推薦）
redis_client: Optional[redis.Redis] = None

async def get_redis():
    global redis_client
    if redis_client is None:
        redis_client = await redis.from_url("redis://localhost:6379")
    return redis_client

def rate_limit(max_requests: int = 10, window_seconds: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            client_ip = request.client.host
            redis_conn = await get_redis()
            key = f"rate_limit:{func.__name__}:{client_ip}"
            
            # 獲取當前請求數
            current = await redis_conn.get(key)
            
            if current and int(current) >= max_requests:
                raise HTTPException(
                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                    detail=f"Rate limit exceeded: {max_requests} requests per {window_seconds} seconds"
                )
            
            # 增加計數器
            pipe = redis_conn.pipeline()
            pipe.incr(key)
            pipe.expire(key, window_seconds)
            await pipe.execute()
            
            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

# 簡單的記憶體版本（開發環境）
from collections import defaultdict

rate_limit_store = defaultdict(list)

def rate_limit_simple(max_requests: int = 10, window_seconds: int = 60):
    def decorator(func):
        @wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            client_ip = request.client.host
            now = datetime.now()
            
            # 清理過期記錄
            rate_limit_store[client_ip] = [
                req_time for req_time in rate_limit_store[client_ip]
                if now - req_time < timedelta(seconds=window_seconds)
            ]
            
            # 檢查速率限制
            if len(rate_limit_store[client_ip]) >= max_requests:
                raise HTTPException(
                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                    detail="Rate limit exceeded"
                )
            
            # 記錄請求
            rate_limit_store[client_ip].append(now)
            
            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

# 使用
@router.get("/api/data")
@rate_limit(max_requests=5, window_seconds=60)
async def get_data(request: Request):
    return {"data": "some data"}

from functools import wraps
from typing import Dict, Any, Optional
import json
import hashlib
from datetime import datetime, timedelta
from fastapi import Request
from pydantic import BaseModel

# 使用 Pydantic v2 的改進效能
cache_store: Dict[str, tuple[Any, datetime]] = {}

def cache_response(ttl: int = 300, key_prefix: str = ""):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 生成緩存鍵（包含請求路徑和參數）
            request = kwargs.get("request") if "request" in kwargs else None
            cache_data = {
                "func": func.__name__,
                "args": str(args),
                "kwargs": {k: v for k, v in kwargs.items() if k != "request"},
                "path": request.url.path if request else ""
            }
            
            cache_key = hashlib.md5(
                json.dumps(cache_data, default=str, sort_keys=True).encode()
            ).hexdigest()
            
            if key_prefix:
                cache_key = f"{key_prefix}:{cache_key}"
            
            # 檢查緩存
            if cache_key in cache_store:
                cached_data, cached_time = cache_store[cache_key]
                if datetime.now() - cached_time < timedelta(seconds=ttl):
                    return cached_data
                else:
                    # 清理過期緩存
                    del cache_store[cache_key]
            
            # 執行函數並緩存結果
            result = await func(*args, **kwargs)
            cache_store[cache_key] = (result, datetime.now())
            
            return result
        return wrapper
    return decorator

# 使用
@router.get("/expensive-operation")
@cache_response(ttl=600, key_prefix="expensive")
async def expensive_operation(request: Request):
    # 耗時操作
    return {"result": "expensive data"}

import logging
from functools import wraps

logger = logging.getLogger(__name__)

def log_execution(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        logger.info(f"Executing {func.__name__} with args: {args}, kwargs: {kwargs}")
        try:
            result = await func(*args, **kwargs)
            logger.info(f"{func.__name__} completed successfully")
            return result
        except Exception as e:
            logger.error(f"{func.__name__} failed with error: {e}")
            raise
    return wrapper

# 使用
@router.post("/users")
@log_execution
async def create_user(user_data: dict):
    return {"user": user_data}

// Program.cs (.NET 10)
var builder = WebApplication.CreateBuilder(args);

// .NET 10 新增的簡化配置
builder.Services.AddProblemDetails(); // 統一的錯誤響應格式

var app = builder.Build();

// .NET 10 改進的端點定義
app.MapGet("/api/users/{id}", async (int id, IUserService service) =>
{
    var user = await service.GetUserByIdAsync(id);
    return user is null ? Results.NotFound() : Results.Ok(user);
})
.WithName("GetUser")
.WithTags("Users")
.Produces<User>(StatusCodes.Status200OK)
.Produces(StatusCodes.Status404NotFound);

app.Run();

// app.module.ts (NestJS v11+)
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { CacheModule } from '@nestjs/cache-manager';

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true, // 全局配置模組
      cache: true,
    }),
    CacheModule.register({
      isGlobal: true,
      ttl: 60000, // 60 秒
    }),
  ],
})
export class AppModule {}

from fastapi import FastAPI, Depends
from pydantic import BaseModel, Field, field_validator
from typing import Optional
from datetime import datetime

app = FastAPI()

# Pydantic v2 模型（效能提升）
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+
使用場景建議
何時使用中間件（Middleware）
✅ 適合場景：
請求日誌記錄
跨域處理（CORS）
請求解析（Body Parser）
靜態文件服務
全局錯誤處理
請求驗證（在路由匹配前）
❌ 不適合場景：
需要訪問控制器特定信息
需要依賴注入特定服務
需要處理業務邏輯
何時使用攔截器（Interceptor）
✅ 適合場景：
日誌記錄（請求/響應）
響應數據轉換
性能監控
超時處理
緩存處理
❌ 不適合場景：
認證授權（應使用 Guard/Filter）
異常處理（應使用 Exception Filter）
何時使用過濾器（Filter）
✅ 適合場景：
異常處理
認證授權
請求驗證
響應格式化
安全檢查
❌ 不適合場景：
簡單的日誌記錄（使用中間件或攔截器）
數據轉換（使用攔截器）
何時使用裝飾器（Decorator）
✅ 適合場景：
路由定義
參數驗證
元數據標記
權限標記
文檔生成（Swagger）
❌ 不適合場景：
複雜的業務邏輯處理
需要異步操作的場景
執行順序
NestJS 執行順序
plain複製
1
2
3
4
5
6
7
8
1. 中間件（Middleware）
2. 守衛（Guard）
3. 攔截器（Interceptor - Before）
4. 管道（Pipe）
5. 控制器方法
6. 攔截器（Interceptor - After）
7. 異常過濾器（Exception Filter）
ASP.NET Core 執行順序
plain複製
1
2
3
4
5
6
7
8
9
10
1. 中間件（Middleware）
2. 授權過濾器（Authorization Filter）
3. 資源過濾器（Resource Filter - OnResourceExecuting）
4. 動作過濾器（Action Filter - OnActionExecuting）
5. 異常過濾器（Exception Filter）
6. 動作過濾器（Action Filter - OnActionExecuted）
7. 結果過濾器（Result Filter - OnResultExecuting）
8. 結果過濾器（Result Filter - OnResultExecuted）
9. 資源過濾器（Resource Filter - OnResourceExecuted）
FastAPI 執行順序
plain複製
1
2
3
4
5
6
7
1. 中間件（Middleware - 請求前）
2. 依賴注入（Dependency - 按順序執行）
3. 路由處理函數
4. 依賴注入（Dependency - yield 後的清理）
5. 中間件（Middleware - 響應後）
6. 異常處理器（Exception Handler - 如果發生異常）
最佳實踐
1. 性能考量
中間件：避免在中間件中執行耗時操作
攔截器：使用異步操作時注意性能影響
過濾器：異常過濾器應快速處理，避免阻塞
2. 錯誤處理
使用專門的異常過濾器處理全局異常
在中間件中捕獲未處理的異常
提供友好的錯誤響應格式
3. 日誌記錄
使用結構化日誌
記錄關鍵信息（請求 ID、用戶 ID、執行時間）
避免記錄敏感信息（密碼、Token）
4. 安全性
在早期階段進行認證（中間件或守衛）
使用 HTTPS
驗證和清理輸入數據
實施速率限制
5. 可測試性
保持邏輯簡單和獨立
使用依賴注入
編寫單元測試
模擬外部依賴
6. 現代化實踐（2025）
NestJS (v11+)
✅ 使用函數式中間件（更簡潔）
✅ 使用 app.use() 全局註冊中間件
✅ 使用 @Injectable() 裝飾器確保依賴注入
✅ 使用 RxJS 操作符處理異步流
✅ 實現請求 ID 追蹤以便調試
✅ 整合 Fastify 以提升效能（可選）
✅ 使用模組化設計提升可維護性
✅ 充分利用 TypeScript 5.x+ 新特性
ASP.NET Core (.NET 10 LTS)
✅ 使用最小 API 簡化配置（.NET 10 效能優化）
✅ 優先使用 IAsyncActionFilter 而非 IActionFilter
✅ 使用結構化日誌記錄（ILogger，.NET 10 改進）
✅ 使用 IHostEnvironment 檢查環境
✅ 實現統一的異常處理格式
✅ 利用 .NET 10 的效能提升（TechEmpower 基準測試表現優異）
✅ 跨平台部署（Windows、macOS、Linux、Docker）
✅ 使用 .NET 10 的新 API 和改進
FastAPI (v0.115+)
✅ 使用 Pydantic v2+ 模型進行類型驗證（效能提升）
✅ 優先使用異步依賴函數
✅ 使用 Generator 模式管理資源（資料庫連接等）
✅ 使用 Redis 進行速率限制（生產環境）
✅ 實現統一的錯誤響應格式
✅ 使用 Python 3.12+ 類型提示改進
✅ 充分利用 FastAPI 的高效能（接近 Node.js 和 Go 的速度）
✅ 自動生成互動式 API 文檔（OpenAPI/ReDoc）
✅ 固定生產環境版本以確保穩定性
常見問題
Q1: 中間件和攔截器有什麼區別？
A:
中間件：在路由匹配前執行，可以終止請求，通常用於全局處理
攔截器：在路由匹配後、控制器方法執行前後執行，可以訪問控制器上下文
Q2: 攔截器和過濾器有什麼區別？
A:
攔截器：主要用於日誌、數據轉換、性能監控等橫切關注點
過濾器：主要用於異常處理、認證授權、請求驗證等
Q3: 應該在哪裡處理認證？
A:
NestJS：使用 Guard（守衛）
ASP.NET Core：使用 Authorization Filter 或 Middleware
FastAPI：使用 Dependency Injection 或 Middleware
Q4: 如何選擇使用哪個？
A:
全局處理：使用中間件
控制器/方法級別：使用攔截器或過濾器
元數據標記：使用裝飾器
異常處理：使用異常過濾器
Q5: 可以同時使用多個嗎？
A: 可以，它們通常按特定順序執行，可以組合使用以實現複雜的需求。
Q6: 如何調試執行順序問題？
A:
添加詳細的日誌記錄
使用請求 ID 追蹤整個請求流程
檢查框架文檔中的執行順序說明
使用調試工具（如斷點）
總結
關鍵要點
中間件：請求處理管道的最外層，適合全局處理
攔截器：在控制器方法執行前後，適合日誌和數據轉換
過濾器：在特定階段執行，適合異常處理和認證
裝飾器：提供元數據，適合標記和配置
選擇建議
新專案：根據框架選擇對應的機制
現有專案：遵循框架的最佳實踐
複雜需求：組合使用多種機制
2025 年重點建議
.NET 10 LTS：新專案建議使用 .NET 10，享受長期支援和效能提升
NestJS v11+：充分利用 TypeScript 5.x+ 和模組化設計
FastAPI 最新版：使用 Pydantic v2+ 獲得更好的效能和開發體驗
效能優先：所有框架都強調效能優化，建議定期更新到最新版本
類型安全：充分利用各框架的類型系統提升代碼品質
)
    age: Optional[int] = Field(None, ge=0, le=150)
    created_at: datetime = Field(default_factory=datetime.now)
    
    @field_validator('username')
    @classmethod
    def validate_username(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError('Username must be alphanumeric')
        return v

@app.post("/users")
async def create_user(user: UserCreate):
    # Pydantic v2 自動驗證，效能更佳
    return {"message": "User created", "user": user}