Home » วิธีสร้าง REST API ด้วย FastAPI Python แบบครบจบ
ไม่มีหมวดหมู่

วิธีสร้าง REST API ด้วย FastAPI Python แบบครบจบ

March 16, 2026
0 Views

สวัสดีครับ! ในโลกของการพัฒนาซอฟต์แวร์ยุคใหม่ การสร้าง API (Application Programming Interface) ถือเป็นหัวใจสำคัญที่เชื่อมโยงระบบต่าง ๆ เข้าไว้ด้วยกัน และ FastAPI คือหนึ่งในสุดยอดเฟรมเวิร์ค Python ที่เข้ามาปฏิวัติวงการด้วยความเร็ว ประสิทธิภาพ และความง่ายในการใช้งาน ที่สำคัญคือมันถูกสร้างขึ้นมาเพื่อการพัฒนา REST API โดยเฉพาะเลยครับ บทความนี้จะพาทุกท่านดำดิ่งสู่โลกของ FastAPI ตั้งแต่เริ่มต้นติดตั้ง ไปจนถึงการสร้าง REST API ที่ครบวงจร พร้อมฟีเจอร์เด็ด ๆ อย่างการเชื่อมต่อฐานข้อมูล, การจัดการข้อมูลด้วย Pydantic, ระบบยืนยันตัวตน, และการนำไปใช้งานจริง เพื่อให้คุณพร้อมที่จะสร้างสรรค์ API คุณภาพสูงได้อย่างมืออาชีพครับ

สารบัญ

ทำไมต้อง FastAPI? (What is FastAPI and Why Use It?)

FastAPI คือเว็บเฟรมเวิร์คสำหรับสร้าง API ที่ทันสมัย รวดเร็ว และมีประสิทธิภาพสูง ซึ่งพัฒนาขึ้นบนพื้นฐานของ Python 3.7+ และมาตรฐาน Type Hints ความน่าสนใจของ FastAPI อยู่ที่การผสมผสานจุดเด่นของเฟรมเวิร์คอื่น ๆ เข้าไว้ด้วยกัน พร้อมทั้งเพิ่มฟีเจอร์ที่ช่วยให้นักพัฒนาทำงานได้ง่ายและเร็วขึ้นมากครับ

จุดเด่นของ FastAPI:

  • ความเร็วสูง (High Performance): สร้างขึ้นบน Starlette (สำหรับเว็บ) และ Pydantic (สำหรับการจัดการข้อมูล) ทำให้มีประสิทธิภาพเทียบเท่ากับ Node.js และ Go ซึ่งเป็นภาษาที่ขึ้นชื่อเรื่องความเร็วในการประมวลผล I/O Bound Operations ครับ
  • เขียนโค้ดได้รวดเร็ว (Fast to Code): ลดเวลาในการพัฒนาได้ถึง 20-40% เพราะมีการใช้ Type Hints ของ Python และ Pydantic ในการตรวจสอบข้อมูลอัตโนมัติ ทำให้โค้ดกระชับขึ้นและลดโอกาสเกิดข้อผิดพลาด
  • ลด Bug (Fewer Bugs): ระบบ Type Hints และ Pydantic ช่วยให้คุณจับข้อผิดพลาดที่เกี่ยวกับ Type ได้ตั้งแต่ตอนเขียนโค้ด ทำให้ Bug ลดลงถึง 50% ครับ
  • ใช้งานง่าย (Easy to Use): มีการเรียนรู้ที่ค่อนข้างต่ำ (low learning curve) สำหรับผู้ที่คุ้นเคยกับ Python อยู่แล้ว
  • เอกสาร API อัตโนมัติ (Automatic Interactive API Documentation): FastAPI สร้างเอกสาร API แบบอินเทอร์แอคทีฟให้โดยอัตโนมัติด้วย Swagger UI (OpenAPI) และ ReDoc ซึ่งช่วยให้การทดสอบและทำความเข้าใจ API ทำได้ง่ายขึ้นมาก ทั้งสำหรับนักพัฒนา API เองและผู้ใช้ API ครับ
  • มาตรฐาน (Standard-based): สนับสนุนมาตรฐาน OpenAPI (เดิมคือ Swagger) สำหรับการสร้าง API และ JSON Schema สำหรับการตรวจสอบข้อมูล ซึ่งเป็นมาตรฐานที่ได้รับการยอมรับในอุตสาหกรรม
  • Dependency Injection ที่ทรงพลัง: ช่วยให้การจัดการ Dependencies ทำได้ง่ายและยืดหยุ่น ทำให้โค้ดเป็นระเบียบและทดสอบได้ง่ายขึ้นครับ

ด้วยเหตุผลเหล่านี้ FastAPI จึงเป็นตัวเลือกที่ยอดเยี่ยมสำหรับการสร้าง REST API ที่ต้องการประสิทธิภาพ ความเร็วในการพัฒนา และความน่าเชื่อถือครับ

การเตรียมความพร้อมก่อนเริ่มต้น

ก่อนที่เราจะเริ่มลงมือสร้าง API กันจริงจัง เรามาเตรียมเครื่องมือและทำความเข้าใจพื้นฐานบางอย่างกันก่อนครับ

ติดตั้ง Python และ Virtual Environment

สิ่งแรกที่คุณต้องมีคือ Python ครับ แนะนำให้ใช้ Python 3.7 ขึ้นไป เพื่อให้เข้ากันได้กับ FastAPI

  1. ติดตั้ง Python:
    • สำหรับ Windows: ดาวน์โหลด Installer จาก python.org และอย่าลืมติ๊กช่อง “Add Python to PATH” ระหว่างติดตั้งครับ
    • สำหรับ macOS: ส่วนใหญ่มี Python ติดตั้งมาให้อยู่แล้ว แต่อาจเป็นเวอร์ชันเก่า แนะนำให้ติดตั้งผ่าน Homebrew (`brew install python3`)
    • สำหรับ Linux: ติดตั้งผ่าน Package Manager ของคุณ (`sudo apt install python3` สำหรับ Debian/Ubuntu หรือ `sudo yum install python3` สำหรับ Fedora/CentOS)
  2. สร้าง Virtual Environment: การใช้ Virtual Environment เป็นสิ่งสำคัญมากในการแยกโปรเจกต์ Python ออกจากกัน เพื่อป้องกันความขัดแย้งของ Dependency ครับ
    3. 
# สร้าง virtual environment ชื่อ .venv
python3 -m venv .venv

# เปิดใช้งาน virtual environment
# สำหรับ macOS/Linux
source .venv/bin/activate

# สำหรับ Windows
.venv\Scripts\activate

    เมื่อเปิดใช้งานแล้ว คุณจะเห็นชื่อของ environment (เช่น `(.venv)`) นำหน้า prompt ใน Terminal ของคุณครับ

ตั้งค่า IDE และเครื่องมือที่จำเป็น

เพื่อประสบการณ์การพัฒนาที่ดีเยี่ยม ผมขอแนะนำ Visual Studio Code (VS Code) ครับ เป็น IDE ที่ฟรี ทรงพลัง และมี Extension มากมาย

  • ติดตั้ง VS Code: ดาวน์โหลดและติดตั้งจากเว็บไซต์ทางการ
  • Extension ที่แนะนำ:
    • Python: จาก Microsoft สำหรับการรองรับ Python อย่างสมบูรณ์ (IntelliSense, Debugging, Formatting)
    • Pylance: สำหรับการตรวจสอบ Type และการแนะนำโค้ดที่ฉลาดขึ้น
    • Black Formatter: สำหรับจัดรูปแบบโค้ด Python ให้สวยงามตามมาตรฐาน PEP 8
    • isort: สำหรับจัดเรียง Import statements

พื้นฐาน REST API ที่ควรรู้

ก่อนที่เราจะเริ่มสร้าง API เรามาทบทวนแนวคิดพื้นฐานของ REST API กันเล็กน้อยครับ

  • REST (Representational State Transfer): เป็นสถาปัตยกรรมสำหรับออกแบบ API ที่เน้นการสื่อสารแบบ Stateless ระหว่าง Client กับ Server โดยใช้ทรัพยากร (Resources) เป็นศูนย์กลางครับ
  • Resources: ทุกอย่างที่เราต้องการจัดการจะถูกมองว่าเป็น Resource ตัวอย่างเช่น `/users`, `/products`, `/orders`
  • HTTP Methods (Verbs): ใช้เพื่อระบุการกระทำที่เราต้องการทำกับ Resource นั้น ๆ ครับ
    • GET: ดึงข้อมูลจาก Server (อ่าน)
    • POST: ส่งข้อมูลใหม่ไปยัง Server (สร้าง)
    • PUT: อัปเดตข้อมูลที่มีอยู่แล้วทั้งหมดบน Server (แก้ไขทั้งหมด)
    • PATCH: อัปเดตข้อมูลที่มีอยู่แล้วบางส่วนบน Server (แก้ไขบางส่วน)
    • DELETE: ลบข้อมูลออกจาก Server (ลบ)
  • Status Codes: ตัวเลข 3 หลักที่ Server ส่งกลับมาเพื่อบอกสถานะของการร้องขอ (เช่น 200 OK, 201 Created, 400 Bad Request, 404 Not Found, 500 Internal Server Error)
  • JSON (JavaScript Object Notation): รูปแบบข้อมูลที่นิยมใช้ในการสื่อสารระหว่าง Client และ Server เนื่องจากอ่านง่ายและเข้ากันได้กับหลายภาษาโปรแกรมครับ

การติดตั้ง FastAPI และ Uvicorn

เมื่อเตรียมพร้อมแล้ว เรามาติดตั้ง FastAPI และ Uvicorn กันครับ

  • FastAPI: ตัวเฟรมเวิร์คหลัก
  • Uvicorn: ASGI Server (Asynchronous Server Gateway Interface) ที่ใช้รันแอปพลิเคชัน FastAPI ในการพัฒนาและใน Production ครับ

ใน Terminal ที่เปิดใช้งาน Virtual Environment อยู่ ให้รันคำสั่งนี้ครับ:


pip install fastapi "uvicorn[standard]"

"uvicorn[standard]" จะติดตั้ง Uvicorn พร้อม Dependencies เพิ่มเติมที่จำเป็น เช่น `h11` และ `websockets` เพื่อการทำงานที่สมบูรณ์แบบครับ

สร้าง REST API แรกของเรา: Hello World

มาเริ่มต้นสร้าง API ง่าย ๆ กันครับ สร้างไฟล์ชื่อ `main.py` ในโปรเจกต์ของคุณ:


# main.py
from fastapi import FastAPI

# สร้าง instance ของ FastAPI
app = FastAPI()

# กำหนด Path Operation สำหรับ HTTP GET request ที่ root URL ("/")
@app.get("/")
async def read_root():
    return {"message": "Hello, SiamLancard.com! Welcome to FastAPI."}

# กำหนด Path Operation สำหรับ HTTP GET request ที่ "/items/{item_id}"
# โดยมี Path Parameter คือ item_id และ Query Parameter คือ q
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

รัน API และทดสอบ

เปิด Terminal ในโฟลเดอร์โปรเจกต์ของคุณ (ที่เปิดใช้งาน Virtual Environment แล้ว) และรันคำสั่ง Uvicorn ครับ:


uvicorn main:app --reload
  • main:app: บอก Uvicorn ว่าให้หา object `app` ภายในไฟล์ `main.py`
  • --reload: โหมด Development ที่จะทำให้ Server รีโหลดอัตโนมัติเมื่อคุณแก้ไขไฟล์โค้ด (สะดวกมาก ๆ ครับ)

เมื่อรันสำเร็จ คุณจะเห็นข้อความประมาณนี้:


INFO:     Will watch for changes in these directories: ['/path/to/your/project']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx]
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

ตอนนี้คุณสามารถเปิดเว็บเบราว์เซอร์ไปที่ http://127.0.0.1:8000 เพื่อดูผลลัพธ์ของ `read_root()` ซึ่งจะแสดง JSON ว่า {"message": "Hello, SiamLancard.com! Welcome to FastAPI."} ครับ

ลองเรียกอีก Endpoint: http://127.0.0.1:8000/items/5?q=somequery คุณจะเห็น {"item_id": 5, "q": "somequery"}

สำรวจ Automatic Interactive API Documentation (Swagger UI & ReDoc)

นี่คือหนึ่งในฟีเจอร์ที่โดดเด่นที่สุดของ FastAPI ครับ เมื่อ Server รันอยู่ คุณสามารถเข้าถึงเอกสาร API ได้โดยอัตโนมัติ:

  • Swagger UI: http://127.0.0.1:8000/docs (เป็น UI ที่นิยมใช้ในการทดสอบ API)
  • ReDoc: http://127.0.0.1:8000/redoc (เป็น UI ที่เน้นการอ่านและทำความเข้าใจเอกสาร)

คุณจะเห็น Endpoints ที่เราสร้างไว้ พร้อมรายละเอียดของ Path Parameters, Query Parameters และสามารถทดลองส่ง Request ได้จากหน้าเว็บนั้นเลยครับ สะดวกมาก ๆ

ทำความเข้าใจโครงสร้างหลักของ FastAPI

เมื่อเราได้สร้างและรัน API แรกไปแล้ว มาทำความเข้าใจโครงสร้างพื้นฐานกันให้ลึกซึ้งขึ้นครับ

Path Operations และ HTTP Methods

ใน FastAPI การสร้าง “Path Operation” คือการผูกฟังก์ชัน Python เข้ากับ Path (URL) และ HTTP Method (GET, POST, PUT, DELETE) ที่เฉพาะเจาะจงครับ


from fastapi import FastAPI

app = FastAPI()

# GET request: ดึงข้อมูล
@app.get("/users/")
async def get_users():
    return [{"username": "Alice"}, {"username": "Bob"}]

# POST request: สร้างข้อมูลใหม่
@app.post("/users/")
async def create_user(user_data: dict):
    return {"message": "User created", "data": user_data}

# PUT request: อัปเดตข้อมูลทั้งหมด
@app.put("/users/{user_id}")
async def update_user(user_id: int, user_data: dict):
    return {"message": f"User {user_id} updated", "data": user_data}

# DELETE request: ลบข้อมูล
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"message": f"User {user_id} deleted"}

# PATCH request: อัปเดตข้อมูลบางส่วน (มักใช้ Pydantic ในการจัดการ)
@app.patch("/users/{user_id}")
async def patch_user(user_id: int, user_data: dict):
    return {"message": f"User {user_id} partially updated", "data": user_data}

จะเห็นได้ว่าแต่ละ Decorator (`@app.get`, `@app.post`, ฯลฯ) จะรับ Path เป็นอาร์กิวเมนต์แรก และฟังก์ชันที่ตามมาคือ Path Operation Function ที่จะถูกเรียกใช้เมื่อมี Request เข้ามาตรงกับ Path และ Method นั้น ๆ ครับ

การจัดระเบียบ API ด้วย APIRouter

เมื่อแอปพลิเคชันของคุณใหญ่ขึ้น การเก็บ Path Operations ทั้งหมดไว้ในไฟล์ `main.py` เดียวจะไม่เป็นระเบียบครับ FastAPI มี `APIRouter` ที่ช่วยให้เราสามารถแยก Path Operations ออกเป็นโมดูลย่อย ๆ ได้

1. สร้างไฟล์ `routers/items.py` สำหรับจัดการ API ที่เกี่ยวกับ Item:


# routers/items.py
from fastapi import APIRouter

router = APIRouter(
    prefix="/items", # กำหนด prefix สำหรับทุก Path ใน Router นี้
    tags=["Items"],  # กำหนด Tag สำหรับเอกสาร API
    responses={404: {"description": "Not found"}},
)

fake_items_db = {"foo": {"name": "Foo", "price": 50.2}, "bar": {"name": "Bar", "price": 62.0}}

@router.get("/")
async def read_items():
    return fake_items_db

@router.get("/{item_id}")
async def read_item(item_id: str):
    if item_id not in fake_items_db:
        # สามารถเรียก HTTPException ได้แม้จะอยู่ใน Router
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_items_db[item_id]

@router.post("/")
async def create_item(item: dict):
    fake_items_db[item["name"]] = item
    return {"message": "Item created successfully", "item": item}

2. แก้ไข `main.py` เพื่อรวม Router เข้ามา:


# main.py
from fastapi import FastAPI
from routers import items # import router ที่เราสร้าง

app = FastAPI(
    title="SiamLancard FastAPI Demo", # กำหนดชื่อ API ในเอกสาร
    description="A comprehensive guide to building REST APIs with FastAPI.",
    version="1.0.0",
)

# รวม router เข้ามาในแอปพลิเคชันหลัก
app.include_router(items.router)

@app.get("/")
async def root():
    return {"message": "Welcome to the FastAPI Demo API!"}

ตอนนี้ Endpoint ของ Items จะเข้าถึงได้ที่ `/items/` และ `/items/{item_id}` โดยอัตโนมัติ และเอกสาร API จะถูกจัดกลุ่มภายใต้ Tag “Items” ครับ

การจัดการพารามิเตอร์ (Parameters) ใน API

FastAPI ทำให้การจัดการพารามิเตอร์ต่าง ๆ เป็นเรื่องง่ายด้วย Python Type Hints ครับ

Path Parameters: ระบุข้อมูลใน URL

ใช้เพื่อระบุส่วนหนึ่งของ URL ที่เป็นตัวแปร มักใช้ในการระบุ Resource ที่ต้องการเข้าถึง


# main.py (หรือใน router)
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int): # FastAPI จะแปลง user_id เป็น int อัตโนมัติ
    return {"user_id": user_id, "message": f"Fetching user with ID {user_id}"}

@app.get("/files/{file_path:path}") # ใช้ :path เพื่อให้ยอมรับ path ที่มี / ได้
async def get_file_path(file_path: str):
    return {"file_path": file_path}

ลองไปที่ `/users/123` หรือ `/files/home/user/document.txt` ครับ

Query Parameters: กรองและจัดเรียงข้อมูล

ใช้ในการส่งข้อมูลเพิ่มเติมที่ไม่ใช่ส่วนหนึ่งของ Path มักใช้สำหรับการกรอง การจัดเรียง หรือ Pagination


# main.py (หรือใน router)
from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

@app.get("/products/")
async def read_products(
    limit: int = 10, # default value 10
    offset: int = 0, # default value 0
    search: Optional[str] = None # Optional parameter
):
    results = {"limit": limit, "offset": offset}
    if search:
        results["search"] = search
    # ในความเป็นจริง เราจะใช้ limit, offset, search ไป query จาก database
    return results

@app.get("/items_with_validation/")
async def read_items_with_validation(
    # ใช้ Query เพื่อเพิ่มการตรวจสอบและ metadata
    q: Optional[str] = Query(
        None,
        min_length=3,
        max_length=50,
        title="Search query",
        description="Search term for items in the database"
    )
):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

ลองเรียก: `/products/?limit=5&offset=10&search=keyboard` หรือ `/items_with_validation/?q=apple`

FastAPI จะใช้ Type Hints ในการตรวจสอบ (validation) และแปลงข้อมูลโดยอัตโนมัติ ถ้าข้อมูลที่ส่งมาไม่ตรง Type (เช่น ส่ง `limit=abc`) FastAPI จะส่ง HTTP 422 Unprocessable Entity กลับไปพร้อมรายละเอียดข้อผิดพลาดทันทีครับ

Request Body และ Pydantic Models: ส่งข้อมูลที่ซับซ้อน

สำหรับ HTTP Methods อย่าง POST, PUT, PATCH เรามักจะส่งข้อมูลที่ซับซ้อน (เช่น JSON object) ไปยัง Server ซึ่ง FastAPI ใช้ Pydantic ในการจัดการ Request Body ได้อย่างยอดเยี่ยมครับ


# main.py (หรือใน router)
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# กำหนด Pydantic Model สำหรับ Item
class Item(BaseModel):
    name: str = Field(..., example="Laptop Pro") # ... คือ required
    description: Optional[str] = Field(None, example="Powerful laptop for professionals")
    price: float = Field(..., gt=0, example=1200.0) # gt=0 คือต้องมากกว่า 0
    tax: Optional[float] = Field(None, example=120.0)

# กำหนด Pydantic Model สำหรับ User
class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

@app.post("/items/")
async def create_item(item: Item): # FastAPI จะรับ Request Body มาสร้างเป็น object Item
    item_dict = item.dict() # แปลง Pydantic Model เป็น Python dict
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

@app.put("/users/{user_id}")
async def update_user(user_id: int, user: User):
    return {"user_id": user_id, **user.dict()}

เมื่อคุณเข้าดูเอกสาร Swagger UI (`/docs`) คุณจะเห็น Schema ของ `Item` และ `User` โดยอัตโนมัติ และสามารถทดลองส่ง JSON ที่ถูกต้องตาม Model ได้เลยครับ FastAPI จะตรวจสอบความถูกต้องของข้อมูล (Data Validation) ให้โดยอัตโนมัติ และแปลง JSON เป็น Python object ที่เป็น Type ของ Pydantic Model ให้เราใช้งานได้ทันที

เจาะลึก Pydantic: การตรวจสอบและจัดการข้อมูล

Pydantic คือไลบรารีที่ FastAPI ใช้ในการประกาศ Schema ของข้อมูล, ตรวจสอบความถูกต้อง, และแปลงข้อมูล (serialization/deserialization) ทำให้การจัดการข้อมูลเป็นเรื่องง่ายและปลอดภัยครับ

การสร้างและตรวจสอบข้อมูลด้วย BaseModel

ทุก Pydantic Model จะสืบทอดมาจาก `BaseModel` ครับ


# models.py (แยกไฟล์เพื่อความเป็นระเบียบ)
from pydantic import BaseModel
from typing import List, Optional

class Product(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tags: List[str] = [] # กำหนดค่า default เป็น list ว่าง

class Order(BaseModel):
    order_id: str
    products: List[Product]
    customer_email: str
    total_amount: float
    status: str = "pending" # กำหนดค่า default

# ใน Path Operation
@app.post("/products/")
async def create_product(product: Product):
    # product จะถูกตรวจสอบความถูกต้องตาม Product Model โดยอัตโนมัติ
    return {"message": "Product created", "product": product.dict()}

ถ้า Request Body ไม่ตรงตาม Schema ของ `Product` (เช่น `price` ไม่ใช่ float, `id` ไม่มี) FastAPI จะตอบกลับด้วย HTTP 422 Unprocessable Entity ทันทีครับ

การใช้ Field สำหรับการตรวจสอบขั้นสูง

เราสามารถใช้ `Field` จาก Pydantic เพื่อเพิ่มเงื่อนไขการตรวจสอบที่ซับซ้อนขึ้นได้ครับ


from pydantic import BaseModel, Field
from typing import Optional

class UserRegistration(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, regex="^[a-zA-Z0-9_]+$")
    email: str = Field(..., example="[email protected]") # Pydantic ตรวจสอบรูปแบบอีเมลอัตโนมัติ
    password: str = Field(..., min_length=8, max_length=64)
    age: Optional[int] = Field(None, gt=0, le=120) # ต้องมากกว่า 0 และน้อยกว่าหรือเท่ากับ 120
    is_active: bool = True # ค่าเริ่มต้นเป็น True

ตัวอย่างการใช้งาน `Field`:

  • ...: หมายถึง Required Field
  • min_length, max_length: สำหรับ String
  • gt (greater than), ge (greater than or equal), lt (less than), le (less than or equal): สำหรับตัวเลข
  • regex: สำหรับตรวจสอบรูปแบบ String ด้วย Regular Expression
  • example: เพิ่มตัวอย่างข้อมูลในเอกสาร Swagger UI

Nested Models และการสร้างโครงสร้างข้อมูลที่ซับซ้อน

Pydantic อนุญาตให้คุณสร้าง Model ที่ซ้อนกันได้ เพื่อจัดการข้อมูลที่มีความสัมพันธ์กัน


from pydantic import BaseModel, HttpUrl
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    zip_code: str
    country: str

class Company(BaseModel):
    name: str
    location: Address
    website: Optional[HttpUrl] = None # Pydantic ตรวจสอบ URL อัตโนมัติ

class Employee(BaseModel):
    id: int
    name: str
    email: str
    position: str
    company: Company # Nested Model
    skills: List[str] = []

@app.post("/employees/")
async def create_employee(employee: Employee):
    return {"message": "Employee data received", "employee": employee.dict()}

เมื่อคุณส่ง JSON สำหรับ `/employees/` คุณจะต้องส่ง JSON ที่มีโครงสร้างตรงตาม `Employee` ซึ่งรวมถึง `Company` และ `Address` ที่ซ้อนอยู่ข้างในด้วยครับ

การเชื่อมต่อฐานข้อมูล: FastAPI กับ SQLAlchemy และ SQLite

FastAPI ไม่ได้มี ORM (Object-Relational Mapping) ของตัวเอง แต่ทำงานร่วมกับ ORM ยอดนิยมอื่น ๆ ได้ดีเยี่ยมครับ ในตัวอย่างนี้เราจะใช้ SQLAlchemy ซึ่งเป็น ORM ที่ทรงพลังและยืดหยุ่น ร่วมกับฐานข้อมูล SQLite (สำหรับการพัฒนา) และไลบรารี `databases` เพื่อรองรับ Asynchronous Database Operations ครับ

ทำความเข้าใจ ORM (Object-Relational Mapping)

ORM เป็นเทคนิคที่ช่วยให้นักพัฒนาสามารถโต้ตอบกับฐานข้อมูลได้โดยใช้ Object-Oriented Programming Language แทนการเขียน SQL โดยตรงครับ

  • ข้อดี: ลด boilerplate code, ทำงานกับฐานข้อมูลได้หลากหลาย, ป้องกัน SQL Injection ได้ในระดับหนึ่ง, โค้ดอ่านง่ายขึ้น
  • ข้อเสีย: อาจมี Overhead เล็กน้อย, ไม่สามารถใช้ประโยชน์จาก SQL ขั้นสูงได้เต็มที่ถ้าไม่เข้าใจ ORM ดีพอ

ตั้งค่าและเตรียมฐานข้อมูล (SQLite)

1. ติดตั้งไลบรารีที่จำเป็น:


pip install sqlalchemy databases[sqlite]

2. สร้างไฟล์ `database.py` เพื่อตั้งค่าการเชื่อมต่อ:


# database.py
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from databases import Database

# สำหรับ SQLite ในไฟล์
DATABASE_URL = "sqlite:///./sql_app.db"

# สำหรับ SQLAlchemy
engine = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False} # จำเป็นสำหรับ SQLite ใน FastAPI
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# สำหรับ Databases (Async)
database = Database(DATABASE_URL)

3. สร้างไฟล์ `models.py` เพื่อกำหนด Schema ของตาราง (SQLAlchemy Models):


# models.py
from sqlalchemy import Column, Integer, String, Boolean
from .database import Base # import Base จาก database.py

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer) # Foreign Key (ไม่ได้กำหนดในตัวอย่างนี้เพื่อความง่าย)

4. แก้ไข `main.py` เพื่อสร้างตารางและเชื่อมต่อ/ตัดการเชื่อมต่อฐานข้อมูล:


# main.py
from fastapi import FastAPI
from .database import Base, engine, database # import จาก database.py
from . import models # import models.py เพื่อให้ SQLAlchemy รู้จักตาราง

# สร้างตารางในฐานข้อมูล (ถ้ายังไม่มี)
Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

# ... Path Operations อื่นๆ ...

สร้าง CRUD Operations (Create, Read, Update, Delete)

เราจะใช้ Dependency Injection เพื่อจัดการ Database Session ครับ

1. เพิ่ม Pydantic Models สำหรับ Request/Response ใน `schemas.py`:


# schemas.py
from pydantic import BaseModel
from typing import List, Optional

class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int # ในความเป็นจริงควรมีเจ้าของ
    class Config:
        orm_mode = True # สำคัญมากสำหรับ SQLAlchemy!

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = [] # User สามารถมีหลาย Items ได้
    class Config:
        orm_mode = True

2. เพิ่มฟังก์ชัน CRUD ใน `crud.py` (หรือ `dependencies.py`):


# crud.py
from sqlalchemy.orm import Session
from . import models, schemas

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

การใช้ Dependency Injection สำหรับ Database Session

FastAPI มีระบบ Dependency Injection ที่ทรงพลัง ทำให้เราสามารถ “ฉีด” Database Session เข้าไปใน Path Operation Function ได้อย่างง่ายดาย

3. แก้ไข `main.py` อีกครั้งเพื่อเพิ่ม Path Operations สำหรับ User และ Item:


# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

from .database import SessionLocal, engine, Base, database
from . import models, schemas, crud # import crud.py

# สร้างตารางในฐานข้อมูล (ถ้ายังไม่มี)
Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="SiamLancard FastAPI Database Demo",
    description="Building a REST API with FastAPI and SQLAlchemy.",
    version="1.0.0",
)

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

# Dependency สำหรับการรับ Database Session
def get_db():
    db = SessionLocal()
    try:
        yield db # ส่ง db session ให้กับ Path Operation Function
    finally:
        db.close() # ปิด db session เมื่อ Path Operation ทำงานเสร็จ

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

ด้วยวิธีนี้ คุณไม่ต้องกังวลเรื่องการเปิด/ปิด Database Session ด้วยตัวเอง FastAPI และ Dependency Injection จะจัดการให้ครับ และคุณสามารถนำโค้ดที่เกี่ยวข้องกับ Database CRUD ไปรวมไว้ในโมดูล `crud.py` เพื่อให้โค้ดเป็นระเบียบยิ่งขึ้นครับ

อ่านเพิ่มเติมเกี่ยวกับการจัดการฐานข้อมูลใน FastAPI

การจัดการข้อผิดพลาด (Error Handling)

การจัดการข้อผิดพลาดที่ดีเป็นสิ่งสำคัญสำหรับ API ที่แข็งแกร่ง FastAPI มีวิธีจัดการข้อผิดพลาดที่ยืดหยุ่นครับ

การใช้ HTTPException สำหรับข้อผิดพลาดมาตรฐาน

สำหรับข้อผิดพลาด HTTP ทั่วไป เช่น 404 Not Found, 401 Unauthorized, 403 Forbidden, 400 Bad Request คุณสามารถใช้ `HTTPException` ได้โดยตรง


from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id % 2 != 0: # สมมติว่า user ID ต้องเป็นเลขคู่
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User ID must be an even number",
            headers={"X-Error-Code": "INVALID_USER_ID"} # สามารถเพิ่ม custom header ได้
        )
    if user_id > 100: # สมมติว่าไม่มี user ID เกิน 100
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return {"user_id": user_id, "name": f"User {user_id}"}

FastAPI จะแปลง `HTTPException` ให้เป็น HTTP Response ที่เหมาะสมโดยอัตโนมัติ

การสร้าง Custom Exception Handlers

ในบางกรณี คุณอาจต้องการจัดการข้อผิดพลาดประเภทอื่น ๆ หรือปรับแต่ง Response ของ `HTTPException` เอง คุณสามารถใช้ `@app.exception_handler` ได้


from fastapi import FastAPI, Request, HTTPException, status
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# กำหนด custom exception class
class CustomError(Exception):
    def __init__(self, name: str, message: str):
        self.name = name
        self.message = message

# Handler สำหรับ CustomError
@app.exception_handler(CustomError)
async def custom_error_handler(request: Request, exc: CustomError):
    return JSONResponse(
        status_code=status.HTTP_418_IM_A_TEAPOT, # ใช้สถานะที่ตลกๆ สำหรับตัวอย่าง
        content={"error_name": exc.name, "error_message": exc.message},
    )

# Handler สำหรับ HTTPException (เพื่อปรับแต่ง Response)
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "custom_info": "This is a custom HTTP error response."},
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 42:
        raise CustomError(name="MeaningOfLifeError", message="The item ID 42 is reserved for philosophical discussions.")
    if item_id == 99:
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Access to item 99 is forbidden.")
    return {"item_id": item_id, "name": "Item Name"}

ตอนนี้เมื่อคุณเรียก `/items/42` คุณจะได้รับ Response ที่มาจาก `custom_error_handler` และเมื่อเรียก `/items/99` จะได้รับ Response ที่มาจาก `http_exception_handler` ที่เราปรับแต่งไว้ครับ

การตรวจสอบสิทธิ์ (Authentication) และการอนุญาต (Authorization)

การยืนยันตัวตนและการอนุญาตเป็นสิ่งสำคัญสำหรับ API เพื่อควบคุมว่าใครสามารถเข้าถึงข้อมูลหรือฟังก์ชันใดได้บ้าง

แนวคิดพื้นฐานด้านความปลอดภัย

  • Authentication (การยืนยันตัวตน): คือกระบวนการพิสูจน์ว่าผู้ใช้คือใคร (เช่น Login ด้วย Username/Password)
  • Authorization (การอนุญาต): คือกระบวนการกำหนดว่าผู้ใช้ที่ได้รับการยืนยันตัวตนแล้วมีสิทธิ์ทำอะไรได้บ้าง (เช่น ผู้ดูแลระบบสามารถลบข้อมูลได้, ผู้ใช้ทั่วไปทำได้แค่อ่าน)

FastAPI มีเครื่องมือช่วยในการสร้างระบบความปลอดภัยได้อย่างง่ายดาย โดยใช้ Dependency Injection และ Security Utilities ที่มีมาให้ครับ

การสร้าง API Key Authentication แบบง่าย

เราสามารถใช้ `Security` และ `APIKeyHeader` หรือ `APIKeyQuery` เพื่อสร้างระบบ API Key ได้


from fastapi import FastAPI, Security, HTTPException, status
from fastapi.security import APIKeyHeader

app = FastAPI()

API_KEY = "supersecretapikey"
API_KEY_NAME = "X-API-Key" # ชื่อ Header ที่จะส่ง API Key มา

# สร้าง Security Scheme
api_key_header_auth = APIKeyHeader(name=API_KEY_NAME, auto_error=True)

# Dependency สำหรับตรวจสอบ API Key
async def get_api_key(api_key_header: str = Security(api_key_header_auth)):
    if api_key_header == API_KEY:
        return api_key_header
    else:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Could not validate credentials"
        )

@app.get("/secure-data/", dependencies=[Depends(get_api_key)])
async def read_secure_data():
    return {"message": "This is highly confidential data!"}

@app.get("/public-data/")
async def read_public_data():
    return {"message": "This is public data, no authentication needed."}

เมื่อคุณเรียก `/secure-data/` คุณจะต้องส่ง Header `X-API-Key: supersecretapikey` มาด้วย มิฉะนั้นจะถูกปฏิเสธครับ

การใช้ JWT และ OAuth2PasswordBearer สำหรับระบบ Login

สำหรับระบบ Login ที่มี Username/Password และต้องการ Token (เช่น JWT) เพื่อยืนยันตัวตนใน Request ถัดไป FastAPI มี `OAuth2PasswordBearer` ซึ่งเป็นเครื่องมือที่ใช้ในการจัดการ OAuth2 protocol ครับ

1. ติดตั้งไลบรารีที่จำเป็น:


pip install python-jose[cryptography] passlib[bcrypt]

2. สร้างไฟล์ `auth_utils.py` สำหรับฟังก์ชันช่วยในการเข้ารหัส/ถอดรหัส Token:


# auth_utils.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext

# กำหนดค่า Secret Key และ Algorithm สำหรับ JWT
SECRET_KEY = "your-super-secret-key" # ควรเก็บใน Environment Variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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 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=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def decode_access_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        return None

3. แก้ไข `main.py` เพื่อเพิ่มระบบ Login และ Protected Endpoint:


# main.py (เพิ่มส่วนนี้เข้าไป)
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
from .auth_utils import verify_password, create_access_token, decode_access_token, ACCESS_TOKEN_EXPIRE_MINUTES
from datetime import timedelta

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Pydantic Model สำหรับ Token
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

# Mock Database สำหรับผู้ใช้ (ในความเป็นจริงจะดึงจาก DB)
fake_users_db = {
    "john.doe": {
        "username": "john.doe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": "$2b$12$R.S/Xf.A/O/k0N8b.e0O.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u.e.u", # password: "password123"
        "disabled": False,
    }
}

def get_user(db, username: str):
    if username in db:
        return db[username]
    return None

async def authenticate_user(username: str, password: str):
    user = get_user(fake_users_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    payload = decode_access_token(token)
    if payload is None:
        raise credentials_exception
    username: str = payload.get("sub")
    if username is None:
        raise credentials_exception
    token_data = TokenData(username=username)
    user = get_user(fake_users_db, token_data.username)
    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()):
    user = await authenticate_user(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=schemas.User) # ควรใช้ schemas.User จากตัวอย่าง DB
async def read_users_me(current_user: dict = Depends(get_current_user)):
    # แปลง current_user dict เป็น Pydantic Model ที่เหมาะสม
    return schemas.User(
        id=1, # สมมติ ID
        email=current_user["email"],
        is_active=current_user["disabled"], # หรือ is_active
        items=[] # หรือดึงจาก DB จริงๆ
    )

@app.get("/private-resource/")
async def read_private_resource(current_user: dict = Depends(get_current_user)):
    return {"message": f"Hello {current_user['username']}! This is a private resource."}

คุณสามารถใช้ `/token` Endpoint เพื่อ Login ด้วย Username/Password (ลองใช้ `john.doe` / `password123`) เพื่อรับ Access Token มาใช้ใน Request ถัดไปสำหรับ Endpoint ที่ protected โดย `get_current_user` ครับ

อ่านเพิ่มเติมเกี่ยวกับการสร้างระบบยืนยันตัวตนใน FastAPI

การจัดการไฟล์ (File Uploads)

FastAPI ทำให้การจัดการไฟล์ที่อัปโหลดง่ายมาก โดยใช้ `UploadFile` และ `File` ครับ


from fastapi import FastAPI, File, UploadFile
from typing import List
import shutil # สำหรับ save ไฟล์

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
    try:
        # บันทึกไฟล์ที่อัปโหลดลงในเครื่อง
        with open(f"uploaded_{file.filename}", "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
    finally:
        file.file.close() # ต้องปิดไฟล์ที่อัปโหลดมา
    return {"filename": file.filename, "content_type": file.content_type}

@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile]):
    uploaded_filenames = []
    for file in files:
        try:
            with open(f"uploaded_{file.filename}", "wb") as buffer:
                shutil.copyfileobj(file.file, buffer)
            uploaded_filenames.append(file.filename)
        finally:
            file.file.close()
    return {"message": f"Successfully uploaded {len(uploaded_filenames)} files", "filenames": uploaded_filenames}

@app.post("/upload_with_metadata/")
async def create_upload_file_with_metadata(
    description: str,
    file: UploadFile = File(...) # File(...) ทำให้เป็น Required Field
):
    try:
        with open(f"uploaded_{file.filename}", "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
    finally:
        file.file.close()
    return {"filename": file.filename, "description": description, "content_type": file.content_type}

คุณสามารถทดสอบ Endpoint เหล่านี้ได้จาก Swagger UI โดยเลือก “Try it out” แล้วเลือกไฟล์ที่ต้องการอัปโหลดครับ

การทำ Asynchronous Operations (Async/Await)

FastAPI ถูกออกแบบมาให้ทำงานร่วมกับ Asynchronous Python (async/await) ได้อย่างเต็มประสิทธิภาพ ซึ่งเหมาะมากสำหรับ I/O-bound operations เช่น การอ่าน/เขียนไฟล์, การเรียกใช้ Database, หรือการเรียก API ภายนอก

เมื่อคุณประกาศ Path Operation Function ด้วย `async def` FastAPI จะรันมันใน Event Loop หลัก ทำให้สามารถจัดการ Request อื่น ๆ ได้ในขณะที่รอ I/O Operation เสร็จสิ้น


from fastapi import FastAPI
import asyncio # สำหรับจำลองงานที่ใช้เวลา

app = FastAPI()

async def fetch_data_from_external_api():
    # จำลองการเรียก API ภายนอกที่ใช้เวลา
    await asyncio.sleep(2) # รอ 2 วินาที
    return {"source": "external_api", "data": "Some fetched data"}

@app.get("/async-data/")
async def get_async_data():
    data = await fetch_data_from_external_api()
    return {"message": "Data fetched asynchronously", "result": data}

@app.get("/sync-data/")
def get_sync_data(): # ฟังก์ชันนี้เป็น synchronous
    import time
    time.sleep(2) # blocking call
    return {"message": "Data fetched synchronously (blocking)"}

เมื่อคุณเรียก `/async-data/` หลาย ๆ ครั้งพร้อมกัน คุณจะสังเกตได้ว่า Server ยังคงตอบสนองต่อ Request อื่น ๆ ได้รวดเร็ว แต่ถ้าคุณเรียก `/sync-data/` หลาย ๆ ครั้งพร้อมกัน Server อาจจะดูเหมือนค้างไปชั่วขณะ เพราะ `time.sleep()` เป็น Blocking Call ครับ

ข้อควรระวัง: ถ้าคุณมีโค้ดที่เป็น Blocking Call (เช่น ไลบรารีที่ไม่ได้รองรับ `async/await`) และต้องการรันใน `async def` Path Operation, FastAPI จะย้ายมันไปรันใน Thread Pool แยกให้โดยอัตโนมัติ เพื่อไม่ให้ไป Block Event Loop หลัก แต่มันก็ยังดีกว่าการใช้ `def` โดยตรงกับ Blocking Call ในแง่ของ Concurrent Request ครับ

การทดสอบ (Testing) API ของคุณ

การเขียน Unit Test และ Integration Test เป็นสิ่งสำคัญเพื่อให้มั่นใจว่า API ของคุณทำงานได้ถูกต้องและไม่เกิด Regression Bug ครับ FastAPI มีเครื่องมือที่ช่วยให้การทดสอบเป็นเรื่องง่าย

1. ติดตั้งไลบรารีที่จำเป็น:


pip install pytest httpx

2. สร้างไฟล์ `test_main.py` ในโฟลเดอร์โปรเจกต์ของคุณ:



# test_main.py

from fastapi.testclient import TestClient

from main import app # import app instance จากไฟล์ main.py


client = TestClient(app)


def test_read_main():

    response = client.get("/")

    assert response.status_code == 200

    assert response.json() == {"message": "Welcome to the FastAPI Demo API!"}


def test_create_item():

    response = client.post(

        "/items/",

        json={"name": "Test Item", "description": "A test item", "price": 10.99},

    )

    assert response.status_code == 200

    assert response.json()["item"]["name"] == "Test Item"

    assert "Item created successfully" in response.json()["message"]


def test_read_non_existent_item():

    response = client.get("/items/nonexistent") # สมมติว่า item_id เป็น str ใน router

    assert response.status_code == 404

    assert response.json()["detail"] == "Item not found"


def test_secure_data_unauthorized():

    # ทดสอบ Endpoint ที่ต้องการ API Key โดย

    	

บทความที่เกี่ยวข้อง
iCafeForex.com — EA Forex และเครื่องมือเทรด · SiamCafe.net — ชุมชน IT ที่ใหญ่ที่สุด

Related Articles
จัดส่งรวดเร็วส่งด่วนทั่วประเทศ
รับประกันสินค้าเคลมง่าย มีใบรับประกัน
ผ่อนชำระได้บัตรเครดิต 0% สูงสุด 10 เดือน
สะสมแต้ม รับส่วนลดส่วนลดและคะแนนสะสม

ติดต่อเรา | LINE: @siamlancard

Facebook | YouTube | LINE OA

เว็บในเครือ: SiamCafe.net | iCafeForex.com | Siam2R.com | XMSignal.com

© 2026 SiamLancard — จำหน่ายการ์ดแลน อุปกรณ์ Server และเครื่องพิมพ์ใบเสร็จ

SiamLancard
Logo
Free Forex EA Download — XM Signal · EA Forex ฟรี
iCafeForex.com - สอนเทรด Forex | SiamCafe.net
Shopping cart