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

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

March 16, 2026
0 Views

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

สารบัญ

ทำความรู้จักกับ API และ REST

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

API คืออะไร?

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

REST คืออะไร?

REST ย่อมาจาก Representational State Transfer เป็นชุดของหลักการทางสถาปัตยกรรม (architectural principles) สำหรับการออกแบบระบบเครือข่ายที่ไม่ต้องเก็บสถานะ (stateless) โดยใช้โปรโตคอล HTTP ในการสื่อสารเป็นหลักครับ API ที่ปฏิบัติตามหลักการ REST จะถูกเรียกว่า RESTful API ครับ หัวใจสำคัญของ REST คือการมองทุกอย่างเป็น “ทรัพยากร” (Resources) และการเข้าถึงทรัพยากรเหล่านั้นผ่าน URL (Uniform Resource Locator) ที่ไม่ซ้ำกันครับ

หลักการสำคัญของ RESTful API:

  • Client-Server: แยกส่วน Client (ผู้เรียกใช้) และ Server (ผู้ให้บริการ) ออกจากกันอย่างชัดเจน ทำให้แต่ละส่วนสามารถพัฒนาได้อย่างอิสระ
  • Stateless: Server จะไม่เก็บข้อมูลสถานะของ Client ระหว่างคำขอแต่ละครั้ง แต่ละคำขอจาก Client จะต้องมีข้อมูลที่จำเป็นทั้งหมดเพื่อให้ Server สามารถประมวลผลได้
  • Cacheable: Client หรือตัวกลางสามารถแคชการตอบกลับจาก Server ได้ เพื่อเพิ่มประสิทธิภาพ
  • Layered System: ระบบสามารถมีชั้น (layers) ระหว่าง Client และ Server ได้ เช่น Proxy, Load Balancer โดยที่ Client ไม่จำเป็นต้องรู้ว่ามีกี่ชั้น
  • Uniform Interface: เป็นหลักการสำคัญที่ทำให้ระบบ REST มีความสอดคล้องกัน โดยประกอบด้วย:
    • Identification of Resources: ทุกทรัพยากรต้องมีตัวระบุเฉพาะ (URI)
    • Manipulation of Resources Through Representations: Client สามารถแก้ไขทรัพยากรได้โดยการส่ง Representation ของทรัพยากรนั้นไป
    • Self-descriptive Messages: แต่ละข้อความที่ส่งไปมาต้องมีข้อมูลเพียงพอที่จะอธิบายวิธีการประมวลผลข้อความนั้น
    • Hypermedia as the Engine of Application State (HATEOAS): Server สามารถส่งลิงก์ไปยังทรัพยากรที่เกี่ยวข้องใน Response เพื่อให้ Client สามารถสำรวจ API ได้

โดยทั่วไป RESTful API จะใช้ HTTP Methods ในการระบุการกระทำกับทรัพยากรต่างๆ ดังนี้ครับ:

  • GET: ใช้สำหรับเรียกดูข้อมูลทรัพยากร (Read)
  • POST: ใช้สำหรับสร้างทรัพยากรใหม่ (Create)
  • PUT: ใช้สำหรับอัปเดตทรัพยากรทั้งหมด หรือแทนที่ทรัพยากรด้วยข้อมูลใหม่ (Update/Replace)
  • DELETE: ใช้สำหรับลบทรัพยากร (Delete)
  • PATCH: ใช้สำหรับอัปเดตบางส่วนของทรัพยากร (Partial Update)

ทำไมต้อง FastAPI?

จากเฟรมเวิร์ก Python ที่มีอยู่มากมาย เช่น Flask, Django, Pyramid ทำไม FastAPI ถึงโดดเด่นและกลายเป็นตัวเลือกที่น่าสนใจสำหรับนักพัฒนาในปัจจุบัน? มาดูเหตุผลกันครับ

  • ประสิทธิภาพสูง: FastAPI สร้างขึ้นบน Starlette สำหรับส่วนของเว็บ และ Pydantic สำหรับการจัดการข้อมูล ทำให้ได้ประสิทธิภาพที่เทียบเท่ากับ Node.js และ Go ในบางการทดสอบครับ ซึ่งหมายถึง API ของคุณจะตอบสนองได้รวดเร็วแม้มีการเรียกใช้งานจำนวนมาก
  • พัฒนาเร็วขึ้น: ด้วยคุณสมบัติที่ลดการเขียนโค้ดซ้ำซ้อน (boilerplate code) และการตรวจสอบข้อมูลอัตโนมัติ ทำให้คุณสามารถโฟกัสกับการพัฒนา Business Logic ได้มากขึ้น ช่วยลดเวลาในการพัฒนาลงได้อย่างมากครับ
  • ลดข้อผิดพลาด: การใช้ Python type hints ร่วมกับ Pydantic ช่วยให้ FastAPI สามารถตรวจสอบข้อมูลที่เข้ามา (request validation) และข้อมูลที่ส่งออกไป (response serialization) ได้โดยอัตโนมัติ ทำให้ลดข้อผิดพลาดที่เกิดจากประเภทข้อมูลไม่ตรงกัน
  • เอกสาร API อัตโนมัติ: หนึ่งในคุณสมบัติที่โดดเด่นที่สุดคือการสร้างเอกสาร API แบบอินเทอร์แอคทีฟ (Interactive API documentation) โดยอัตโนมัติ ด้วย Swagger UI (หรือ OpenAPI) และ ReDoc ครับ เพียงแค่คุณเขียนโค้ด FastAPI ระบบก็จะสร้างหน้าเอกสารที่สวยงามและใช้งานง่ายให้ทันที ทำให้การทดสอบและทำความเข้าใจ API ง่ายขึ้นมาก
  • รองรับ Asynchronous (async/await): FastAPI ถูกออกแบบมาให้รองรับ Asynchronous programming ตั้งแต่แกนกลาง ทำให้สามารถจัดการกับ I/O-bound operations (เช่น การอ่าน/เขียนฐานข้อมูล, การเรียกใช้ API ภายนอก) ได้อย่างมีประสิทธิภาพ โดยไม่บล็อกการทำงานของเซิร์ฟเวอร์
  • ใช้งานง่ายและ Learning Curve ต่ำ: ถึงแม้จะมีฟีเจอร์ที่ทรงพลัง แต่ FastAPI ก็มีโครงสร้างที่เข้าใจง่าย และใช้ประโยชน์จาก Python type hints ซึ่งเป็นมาตรฐานของ Python ทำให้ผู้ที่มีพื้นฐาน Python สามารถเรียนรู้และเริ่มต้นใช้งานได้ไม่ยากครับ
  • รองรับ Dependency Injection: ระบบ Dependency Injection ที่แข็งแกร่งช่วยให้คุณสามารถสร้างโค้ดที่สามารถนำกลับมาใช้ใหม่ได้ง่าย (reusable code) และจัดการกับการเชื่อมต่อฐานข้อมูล หรือการยืนยันตัวตนได้อย่างมีประสิทธิภาพ

เตรียมความพร้อมสำหรับ FastAPI

ก่อนที่เราจะเริ่มเขียนโค้ด FastAPI เราต้องติดตั้ง Python และตั้งค่าสภาพแวดล้อมการทำงานให้พร้อมก่อนครับ

1. ติดตั้ง Python

FastAPI รองรับ Python 3.7 ขึ้นไปครับ หากคุณยังไม่มี Python ติดตั้งอยู่บนเครื่อง สามารถดาวน์โหลดได้จากเว็บไซต์ทางการของ Python (python.org) ครับ แนะนำให้ติดตั้งเวอร์ชันล่าสุดเพื่อการใช้งานฟีเจอร์ที่ทันสมัยที่สุดครับ

2. สร้าง Virtual Environment

การใช้ Virtual Environment เป็นสิ่งสำคัญในการพัฒนา Python เพื่อแยกโปรเจกต์ออกจากกัน และป้องกันความขัดแย้งของแพ็กเกจครับ


# สร้าง Virtual Environment ชื่อ 'venv'
python3 -m venv venv

# เปิดใช้งาน Virtual Environment
# บน macOS/Linux:
source venv/bin/activate

# บน Windows:
# venv\Scripts\activate.bat

3. ติดตั้ง FastAPI และ Uvicorn

FastAPI ต้องการ Uvicorn ซึ่งเป็น ASGI server เพื่อรันแอปพลิเคชันของเราครับ


pip install fastapi uvicorn[standard]

uvicorn[standard] จะติดตั้ง Uvicorn พร้อมกับแพ็กเกจเสริมที่จำเป็น เช่น httptools และ watchfiles สำหรับการรีโหลดอัตโนมัติเมื่อโค้ดมีการเปลี่ยนแปลงครับ

FastAPI Hello World!

มาเริ่มต้นด้วยโค้ด FastAPI ง่ายๆ กันครับ สร้างไฟล์ชื่อ main.py ขึ้นมาครับ


# main.py
from fastapi import FastAPI

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

# กำหนด Path Operation สำหรับ HTTP GET request ที่ root path "/"
@app.get("/")
async def read_root():
    """
    Endpoint สำหรับทดสอบการทำงานพื้นฐานของ API
    """
    return {"message": "Hello, FastAPI! Welcome to SiamLancard.com"}

จากนั้นรันแอปพลิเคชันด้วย Uvicorn:


uvicorn main:app --reload

คำสั่งนี้หมายถึง:

  • main:app: บอก Uvicorn ให้หาตัวแปร app ในไฟล์ main.py
  • --reload: เปิดใช้งานโหมดรีโหลดอัตโนมัติ เมื่อโค้ดมีการเปลี่ยนแปลง Uvicorn จะรีสตาร์ทเซิร์ฟเวอร์ให้เองครับ เหมาะสำหรับการพัฒนา

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


INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx]
INFO:     Started server process [yyyyy]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

เปิดเบราว์เซอร์แล้วไปที่ http://127.0.0.1:8000 คุณจะเห็น {“message”: “Hello, FastAPI! Welcome to SiamLancard.com”} แสดงขึ้นมาครับ

นอกจากนี้ FastAPI ยังสร้างเอกสาร API ให้อัตโนมัติอีกด้วย ลองไปที่ http://127.0.0.1:8000/docs (Swagger UI) หรือ http://127.0.0.1:8000/redoc (ReDoc) เพื่อดูเอกสาร API ที่สวยงามและโต้ตอบได้ครับ

Path Operations และ HTTP Methods

ใน FastAPI การสร้าง API Endpoint เรียกว่า “Path Operations” ครับ แต่ละ Path Operation จะเชื่อมโยงกับ HTTP Method ที่ใช้ในการเรียกใช้งาน


from fastapi import FastAPI

app = FastAPI()

# GET request
@app.get("/items/")
async def read_items():
    return {"message": "รายการสินค้าทั้งหมด"}

# POST request
@app.post("/items/")
async def create_item():
    return {"message": "สร้างสินค้าใหม่"}

# PUT request
@app.put("/items/{item_id}")
async def update_item(item_id: int):
    return {"message": f"อัปเดตสินค้า ID: {item_id}"}

# DELETE request
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": f"ลบสินค้า ID: {item_id}"}

# PATCH request
@app.patch("/items/{item_id}")
async def partial_update_item(item_id: int):
    return {"message": f"อัปเดตบางส่วนของสินค้า ID: {item_id}"}

FastAPI มี decorator สำหรับ HTTP Methods หลัก ๆ ดังนี้ครับ: @app.get(), @app.post(), @app.put(), @app.delete(), @app.patch() และ @app.options(), @app.head(), @app.trace()

Path Parameters: ดึงข้อมูลจาก URL

Path Parameters คือพารามิเตอร์ที่ฝังอยู่ในส่วนหนึ่งของ URL เพื่อระบุทรัพยากรเฉพาะ เช่น /items/123 โดย 123 คือ ID ของสินค้าครับ


from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    """
    ดึงข้อมูลสินค้าตาม ID ที่ระบุ
    """
    return {"item_id": item_id}

@app.get("/users/{user_name}")
async def read_user(user_name: str):
    """
    ดึงข้อมูลผู้ใช้ตามชื่อที่ระบุ
    """
    return {"user_name": user_name}

ในตัวอย่างข้างต้น item_id: int เป็นการใช้ Python type hint เพื่อบอก FastAPI ว่า item_id ควรจะเป็นจำนวนเต็ม (integer) ครับ FastAPI จะทำการตรวจสอบประเภทข้อมูลให้อัตโนมัติ หากข้อมูลที่ส่งมาไม่ตรงกับ type hint จะเกิดข้อผิดพลาดและ FastAPI จะส่ง HTTP 422 Unprocessable Entity กลับไปให้ครับ

Query Parameters: ส่งข้อมูลเสริมใน URL

Query Parameters คือพารามิเตอร์ที่ส่งมาพร้อมกับ URL หลังจากเครื่องหมาย ? มักใช้สำหรับการกรอง, การจัดเรียง, หรือการแบ่งหน้า (pagination) ครับ เช่น /items/?skip=0&limit=10


from fastapi import FastAPI
from typing import Optional

app = FastAPI()

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    """
    ดึงรายการสินค้าพร้อมการแบ่งหน้า
    - skip: จำนวนรายการที่จะข้ามไป (เริ่มต้น)
    - limit: จำนวนรายการสูงสุดที่จะแสดง
    """
    return {"message": f"ดึงสินค้าตั้งแต่ {skip} ถึง {skip + limit}"}

@app.get("/users/")
async def read_users(name: Optional[str] = None, age: Optional[int] = None):
    """
    ค้นหาผู้ใช้ตามชื่อและอายุ (ไม่บังคับ)
    """
    if name and age:
        return {"message": f"ค้นหาผู้ใช้ชื่อ {name} อายุ {age}"}
    elif name:
        return {"message": f"ค้นหาผู้ใช้ชื่อ {name}"}
    elif age:
        return {"message": f"ค้นหาผู้ใช้ที่มีอายุ {age}"}
    return {"message": "ดึงข้อมูลผู้ใช้ทั้งหมด"}

ในตัวอย่าง skip: int = 0 และ limit: int = 10 เป็นการกำหนดค่าเริ่มต้นให้กับ Query Parameters หาก Client ไม่ได้ระบุมาครับ ส่วน Optional[str] จากโมดูล typing ใช้สำหรับการระบุว่าพารามิเตอร์นั้นเป็นค่าที่ไม่บังคับ (optional) ครับ

Request Body: การส่งข้อมูลแบบ POST, PUT, PATCH

เมื่อ Client ต้องการส่งข้อมูลจำนวนมากไปยัง Server มักจะใช้ Request Body โดยเฉพาะกับการทำ HTTP POST, PUT, และ PATCH ครับ FastAPI ใช้ Pydantic ในการจัดการและตรวจสอบความถูกต้องของ Request Body ได้อย่างยอดเยี่ยมครับ

1. สร้าง Pydantic Model

เราจะสร้างคลาสที่สืบทอดมาจาก pydantic.BaseModel เพื่อกำหนดโครงสร้างข้อมูลที่เราคาดหวังครับ


from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# Pydantic Model สำหรับสินค้า
class Item(BaseModel):
    name: str = Field(..., min_length=3, max_length=50, description="ชื่อสินค้า")
    description: Optional[str] = Field(None, max_length=300, description="รายละเอียดสินค้า")
    price: float = Field(..., gt=0, description="ราคาสินค้า ต้องมากกว่า 0")
    tax: Optional[float] = Field(None, ge=0, description="ภาษี (ถ้ามี) ต้องไม่น้อยกว่า 0")

# Pydantic Model สำหรับผู้ใช้
class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
    disabled: Optional[bool] = False

ใน Pydantic Model เราสามารถใช้ Field เพื่อกำหนดเงื่อนไขการตรวจสอบข้อมูลเพิ่มเติมได้ เช่น min_length, max_length, gt (greater than), ge (greater than or equal), lt (less than), le (less than or equal) ครับ

2. ใช้ Pydantic Model ใน Path Operation

เมื่อเรามี Pydantic Model แล้ว เราก็สามารถใช้มันเป็น Type Hint สำหรับพารามิเตอร์ใน Path Operation ได้เลยครับ


# เพิ่มโค้ด Item และ User class ด้านบน

@app.post("/items/")
async def create_item(item: Item):
    """
    สร้างสินค้าใหม่ด้วยข้อมูลที่ส่งมาใน Request Body
    """
    item_dict = item.dict() # แปลง Pydantic model เป็น Python dictionary
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return {"message": "สินค้าถูกสร้างแล้ว", "data": item_dict}

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    """
    อัปเดตข้อมูลสินค้าทั้งหมดตาม ID
    """
    return {"message": f"สินค้า ID {item_id} ถูกอัปเดตแล้ว", "data": item}

@app.post("/users/")
async def create_user(user: User):
    """
    สร้างผู้ใช้ใหม่
    """
    return {"message": "ผู้ใช้ถูกสร้างแล้ว", "data": user}

FastAPI จะรับผิดชอบในการอ่าน Request Body (ที่เป็น JSON), แปลงเป็น Pydantic Model, ตรวจสอบความถูกต้องของข้อมูล (validation), และหากมีข้อผิดพลาดจะส่ง HTTP 422 กลับไปโดยอัตโนมัติครับ

Response Model: กำหนดรูปแบบการตอบกลับ

การกำหนด Response Model ช่วยให้เราสามารถควบคุมรูปแบบข้อมูลที่ API ส่งกลับไปยัง Client ได้อย่างชัดเจน โดยเฉพาะอย่างยิ่งเมื่อเราต้องการซ่อนข้อมูลบางอย่าง หรือแปลงข้อมูลก่อนส่งออกไปครับ


from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

class ItemCreate(ItemBase):
    # ไม่ต้องมี ID ตอนสร้าง
    pass

class ItemInDB(ItemBase):
    id: int
    owner_id: int # สมมติว่ามีเจ้าของ

class ItemPublic(ItemBase):
    # อาจจะซ่อน owner_id หรือข้อมูลภายในอื่นๆ
    id: int

class UserBase(BaseModel):
    username: str
    email: str

class UserInDB(UserBase):
    id: int
    hashed_password: str # รหัสผ่านที่ถูกแฮช

class UserPublic(UserBase):
    id: int
    full_name: Optional[str] = None
    # ไม่ส่ง hashed_password ออกไป

# สมมติฐานฐานข้อมูล (ในความเป็นจริงจะเชื่อมต่อกับ DB จริงๆ)
fake_db_items = {}
fake_db_users = {}
next_item_id = 1
next_user_id = 1

@app.post("/items/", response_model=ItemPublic)
async def create_item_api(item: ItemCreate):
    global next_item_id
    db_item = ItemInDB(id=next_item_id, owner_id=1, **item.dict()) # owner_id สมมติ
    fake_db_items[next_item_id] = db_item
    next_item_id += 1
    # FastAPI จะแปลง db_item ไปเป็น ItemPublic ก่อนส่งกลับ
    return db_item

@app.get("/items/{item_id}", response_model=ItemPublic)
async def read_item_api(item_id: int):
    item = fake_db_items.get(item_id)
    if not item:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Item not found")
    return item # FastAPI จะแปลง item ไปเป็น ItemPublic ก่อนส่งกลับ

@app.get("/users/{user_id}", response_model=UserPublic)
async def read_user_api(user_id: int):
    user = fake_db_users.get(user_id)
    if not user:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="User not found")
    return user # FastAPI จะแปลง user ไปเป็น UserPublic ก่อนส่งกลับ

โดยการใช้พารามิเตอร์ response_model ใน decorator ของ Path Operation (เช่น @app.post("/items/", response_model=ItemPublic)) FastAPI จะทำการแปลงข้อมูลที่ฟังก์ชัน Path Operation ส่งคืนให้เป็นรูปแบบของ ItemPublic โดยอัตโนมัติ ทำให้เรามั่นใจได้ว่า Client จะได้รับข้อมูลในรูปแบบที่คาดหวังและปลอดภัยครับ

HTTP Status Codes: การตอบกลับสถานะที่เหมาะสม

HTTP Status Codes มีบทบาทสำคัญในการสื่อสารสถานะของคำขอ HTTP ระหว่าง Server และ Client ครับ การใช้ Status Code ที่ถูกต้องช่วยให้ Client เข้าใจผลลัพธ์ของการเรียก API ได้ชัดเจน และสามารถจัดการกับสถานการณ์ต่าง ๆ ได้อย่างเหมาะสม


from fastapi import FastAPI, Response, status, HTTPException
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Product(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

# ตัวอย่างการกำหนด status_code ใน decorator
@app.post("/products/", status_code=status.HTTP_201_CREATED)
async def create_product(product: Product):
    """
    สร้างสินค้าใหม่ พร้อมคืนค่า HTTP 201 Created
    """
    # ... logic สำหรับบันทึกสินค้าลงฐานข้อมูล ...
    return {"message": "สินค้าถูกสร้างเรียบร้อยแล้ว", "product": product}

# ตัวอย่างการคืนค่า status_code แบบไดนามิกด้วย Response object
@app.get("/items/{item_id}")
async def read_item_with_status_code(item_id: int, response: Response):
    """
    ดึงข้อมูลสินค้า และกำหนด Status Code ตามเงื่อนไข
    """
    if item_id == 1:
        return {"item_id": item_id, "name": "ตัวอย่างสินค้า 1"}
    elif item_id == 2:
        response.status_code = status.HTTP_203_NON_AUTHORITATIVE_INFORMATION
        return {"item_id": item_id, "name": "ข้อมูลจากแคช"}
    else:
        # หากไม่พบสินค้า จะคืนค่า 404 โดยอัตโนมัติหากใช้ HTTPException
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")

เราสามารถกำหนด status_code โดยตรงใน decorator ของ Path Operation (เช่น status_code=status.HTTP_201_CREATED) หรือใช้ Response object ที่ FastAPI จัดเตรียมให้เพื่อกำหนด Status Code แบบไดนามิกภายในฟังก์ชันก็ได้ครับ สำหรับกรณีที่เกิดข้อผิดพลาดและต้องการส่ง Status Code ที่บ่งบอกถึงข้อผิดพลาด (เช่น 404 Not Found, 400 Bad Request) เรามักใช้ HTTPException ครับ

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

การจัดการข้อผิดพลาดอย่างเหมาะสมเป็นสิ่งสำคัญสำหรับ API ที่ดีครับ FastAPI มีกลไกการจัดการข้อผิดพลาดในตัวที่แข็งแกร่ง โดยเฉพาะ HTTPException


from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import Dict, Any

app = FastAPI()

# สมมติฐานฐานข้อมูลสินค้า
fake_items_db = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "price": 62.0, "description": "The Bar Fighters"},
    "baz": {"name": "Baz", "description": "No price, just a description"}
}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    """
    ดึงข้อมูลสินค้า หากไม่พบจะคืนค่า 404
    """
    if item_id not in fake_items_db:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return fake_items_db[item_id]

@app.get("/user/{user_id}")
async def read_user(user_id: int):
    """
    ตัวอย่างการตรวจสอบเงื่อนไขที่ซับซ้อนขึ้น
    """
    if user_id == 3:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="User ID 3 is forbidden")
    if user_id == 4:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Operation forbidden",
            headers={"X-Error": "Cannot process user 4"} # เพิ่ม Header พิเศษได้
        )
    return {"user_id": user_id, "name": f"User {user_id}"}

# การจัดการข้อผิดพลาดแบบ Global (Custom Exception Handlers)
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    """
    Custom handler สำหรับ HTTPException ทั่วไป
    """
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": f"โอ้ ไม่! เกิดข้อผิดพลาด: {exc.detail}"}
    )

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request, exc: UnicornException):
    """
    Custom handler สำหรับ Exception ที่เราสร้างขึ้นเอง
    """
    return JSONResponse(
        status_code=status.HTTP_418_IM_A_TEAPOT,
        content={"message": f"ดูเหมือนว่า {exc.name} จะไม่ใช่ยูนิคอร์นที่แท้จริง :("}
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

โดยปกติแล้ว หากเกิด HTTPException FastAPI จะจัดการและคืนค่า JSON ที่มี detail ให้โดยอัตโนมัติครับ แต่เราสามารถสร้าง @app.exception_handler() เพื่อจัดการกับ Exception บางประเภทที่เราต้องการได้ เช่น การคืนค่าข้อความที่แตกต่างออกไป หรือการบันทึก Log เพิ่มเติมครับ

Dependencies: การใช้ Logic ซ้ำ ๆ

ระบบ Dependency Injection ของ FastAPI เป็นคุณสมบัติที่ทรงพลัง ช่วยให้คุณสามารถสร้างโค้ดที่นำกลับมาใช้ใหม่ได้ง่าย และจัดการกับ Logic ที่ซับซ้อน เช่น การยืนยันตัวตน, การเชื่อมต่อฐานข้อมูล หรือการดึงข้อมูลจาก Header ได้อย่างสง่างามครับ


from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional

app = FastAPI()

# Dependency 1: ตรวจสอบ Query Parameter
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    """
    Dependency สำหรับ Query Parameters ทั่วไป
    """
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    """
    ใช้งาน Dependency สำหรับการจัดการ Query Parameters
    """
    return {"message": "รายการสินค้า", "data": commons}

@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    """
    ใช้งาน Dependency เดียวกันกับ Path Operation อื่น
    """
    return {"message": "รายการผู้ใช้", "data": commons}

# Dependency 2: ตรวจสอบ API Key (สมมติ)
def get_api_key(api_key: str):
    """
    Dependency สำหรับตรวจสอบ API Key
    """
    if api_key != "supersecretkey":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Invalid API Key"
        )
    return api_key

@app.get("/protected-data/", dependencies=[Depends(get_api_key)])
async def get_protected_data():
    """
    Endpoint ที่ต้องการ API Key ในการเข้าถึง
    """
    return {"message": "นี่คือข้อมูลลับ!"}

# Dependency 3: Dependency ที่มีค่าคืนกลับ
async def get_current_user(token: str):
    """
    สมมติว่า Dependency นี้ดึงข้อมูลผู้ใช้จาก Token
    """
    if token == "fake-jwt-token":
        return {"username": "john.doe", "id": 123}
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

@app.get("/me/")
async def read_current_user(current_user: dict = Depends(get_current_user)):
    """
    แสดงข้อมูลผู้ใช้ที่กำลังเข้าสู่ระบบ
    """
    return current_user

ในการใช้งาน Depends() เราสามารถส่งฟังก์ชัน (async def หรือ def) ไปให้ได้ครับ FastAPI จะเรียกฟังก์ชัน Dependency นั้นก่อนที่ Path Operation จะทำงาน และจะส่งค่าที่ Dependency คืนกลับมาเป็นพารามิเตอร์ให้กับ Path Operation ครับ นอกจากนี้ยังสามารถใช้ dependencies=[Depends(dependency_function)] เพื่อเพิ่ม Dependency ให้กับ Path Operation ได้โดยตรง หากไม่ต้องการให้ Dependency คืนค่ากลับมาในฟังก์ชันหลักครับ

การรักษาความปลอดภัยและ Authentication

การยืนยันตัวตน (Authentication) และการอนุญาต (Authorization) เป็นส่วนสำคัญของ API ครับ FastAPI มีเครื่องมือที่ช่วยให้เราสามารถสร้างระบบเหล่านี้ได้อย่างง่ายดาย โดยเฉพาะการทำงานร่วมกับ OAuth2


from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
from jose import jwt, JWTError
from datetime import datetime, timedelta

app = FastAPI()

# --- Security Configuration ---
SECRET_KEY = "your-super-secret-key" # ควรเปลี่ยนเป็นค่าที่ซับซ้อนและเก็บเป็น Environment Variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# OAuth2PasswordBearer ใช้สำหรับรับ Token จาก Header "Authorization: Bearer "
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type: str

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

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

# --- User Database (Simulated) ---
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": "fakehashedpassword", # ในโลกจริงควรใช้ bcrypt หรือ securly hash
        "disabled": False,
    },
    "janedoe": {
        "username": "janedoe",
        "email": "[email protected]",
        "full_name": "Jane Doe",
        "hashed_password": "anotherhashedpassword",
        "disabled": True,
    },
}

def get_user(username: str):
    if username in fake_users_db:
        user_dict = fake_users_db[username]
        return UserInDB(**user_dict)
    return None

def verify_password(plain_password: str, hashed_password: str):
    # ในโลกจริงควรใช้ library เช่น `passlib`
    return plain_password + "fakehash" == hashed_password

def authenticate_user(username: str, password: str):
    user = get_user(username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

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

async def get_current_user_from_token(token: str = Depends(oauth2_scheme)):
    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 = get_user(token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: UserInDB = Depends(get_current_user_from_token)):
    if current_user.disabled:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user")
    return current_user

# --- Authentication Endpoints ---

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = 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=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return {"message": f"รายการสินค้าของ {current_user.username}"}

ในตัวอย่างข้างต้น เราใช้ OAuth2PasswordBearer เพื่อจัดการกับ Bearer Token และ jose library สำหรับการสร้างและถอดรหัส JWT (JSON Web Tokens) ครับ หลักการคือ:

  1. Client ส่ง username และ password ไปที่ /token
  2. Server ตรวจสอบข้อมูลและสร้าง JWT (Access Token) กลับไปให้ Client
  3. Client ใช้ Access Token นั้นใน Header Authorization: Bearer สำหรับทุกคำขอที่ต้องการการยืนยันตัวตน
  4. Dependency get_current_user_from_token จะรับ Token มาถอดรหัสและตรวจสอบความถูกต้อง
  5. Dependency get_current_active_user จะตรวจสอบเพิ่มเติมว่าผู้ใช้ไม่ได้ถูกปิดใช้งาน

สิ่งสำคัญคือในโลกจริง คุณควรใช้ library สำหรับการแฮชรหัสผ่านอย่าง passlib และเก็บ SECRET_KEY เป็น Environment Variable ครับ สำหรับรายละเอียดเพิ่มเติมเกี่ยวกับการรักษาความปลอดภัยใน FastAPI สามารถ อ่านเพิ่มเติม ได้ครับ

การเชื่อมต่อฐานข้อมูล

FastAPI ไม่ได้มาพร้อมกับ ORM (Object-Relational Mapping) ในตัว ทำให้คุณมีอิสระในการเลือก ORM ที่เหมาะสมกับโปรเจกต์ของคุณครับ ORM ยอดนิยมที่มักใช้กับ FastAPI ได้แก่ SQLAlchemy, Tortoise ORM และ SQLModel ครับ

ตารางเปรียบเทียบ ORM ยอดนิยมกับ FastAPI

คุณสมบัติ SQLAlchemy Tortoise ORM SQLModel
ประเภท Full-featured ORM (synchronous) Async ORM ORM + Pydantic
การรองรับ Async/Await ต้องใช้ร่วมกับไลบรารีอื่น (เช่น SQLAlchemy-Stubs, asyncpg, aiosqlite) สำหรับ Async DB Driver รองรับ Async/Await เต็มรูปแบบ รองรับ Async/Await เต็มรูปแบบ
การใช้งาน Pydantic ต้องสร้าง Pydantic Model แยกต่างหากเพื่อ Serialization/Validation ต้องสร้าง Pydantic Model แยกต่างหาก Model เป็น Pydantic Model ในตัว ทำให้การทำงานร่วมกันราบรื่น
ความซับซ้อน ค่อนข้างซับซ้อน แต่มีประสิทธิภาพสูงและยืดหยุ่นมาก ใช้งานง่ายกว่า SQLAlchemy สำหรับโปรเจกต์ขนาดเล็กถึงกลาง ออกแบบมาให้ใช้งานง่ายและลดโค้ดซ้ำซ้อน
ฐานข้อมูลที่รองรับ PostgreSQL, MySQL, SQLite, Oracle, MS SQL Server, etc. PostgreSQL, MySQL, SQLite, Oracle (ผ่าน aiomysql, asyncpg, aiosqlite) PostgreSQL, MySQL, SQLite (รองรับโดย SQLAlchemy)
การจัดการ Schema Alembic (Migration tool) มี Migration Tool ในตัว Alembic (ตาม SQLAlchemy)
เหมาะสำหรับ โปรเจกต์ขนาดใหญ่ที่ต้องการความยืดหยุ่นและการควบคุมสูง โปรเจกต์ที่ต้องการความเร็วในการพัฒนาและเป็น Async native โปรเจกต์ที่ต้องการความง่ายในการใช้งาน Pydantic และลดโค้ดซ้ำซ้อน

ตัวอย่างการใช้งาน FastAPI กับ SQLAlchemy

เราจะใช้ SQLAlchemy เป็นตัวอย่างในการเชื่อมต่อฐานข้อมูล SQLite แบบง่ายๆ ครับ (ในโปรดักชันควรใช้ PostgreSQL หรือ MySQL)

ติดตั้งแพ็กเกจที่จำเป็น:


pip install sqlalchemy databases aiosqlite # databases และ aiosqlite สำหรับ async DB connection

สร้างไฟล์ 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 # สำหรับ async database operations

# การตั้งค่าฐานข้อมูล
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # ใช้ SQLite สำหรับตัวอย่าง

# สำหรับ SQLAlchemy sync engine
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} # สำหรับ SQLite
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# สำหรับ async database operations
database = Database(SQLALCHEMY_DATABASE_URL)

# Dependency สำหรับการรับ Session จาก SQLAlchemy
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

สร้างไฟล์ models.py (SQLAlchemy Models):


# models.py
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from .database import Base

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)

    items = relationship("Item", back_populates="owner")

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, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

สร้างไฟล์ schemas.py (Pydantic Schemas):


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

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

# Properties to receive on item creation
class ItemCreate(ItemBase):
    pass

# Properties to return to client
class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True # สำคัญ: เพื่อให้ Pydantic อ่านข้อมูลจาก ORM model ได้

# Shared properties for User
class UserBase(BaseModel):
    email: str

# Properties to receive on user creation
class UserCreate(UserBase):
    password: str

# Properties to return to client
class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = [] # User สามารถมีหลาย Item

    class Config:
        orm_mode = True

แก้ไขไฟล์ main.py:


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

from . import models, schemas, database # ต้องเปลี่ยนเป็น import แบบ relative หรือแก้ไข PYTHONPATH

app = FastAPI()

# สร้างตารางในฐานข้อมูลเมื่อแอปพลิเคชันเริ่มต้น
@app.on_event("startup")
async def startup_event():
    models.Base.metadata.create_all(bind=database.engine)
    await database.database.connect() # เชื่อมต่อ async database

@app.on_event("shutdown")
async def shutdown_event():
    await database.database.disconnect() # ปิดการเชื่อมต่อ async database

# Dependency เพื่อรับ DB Session
def get_db():
    db = database.SessionLocal()
    try:
        yield db
    finally:
        db.close()

# --- User Endpoints ---
@app.post("/users/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered")
    
    # ในโลกจริงควรแฮชรหัสผ่าน
    hashed_password = user.password + "notreallyhashed" 
    db_user = models.User(email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

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

@app.get("/users/{user_id}", response_model=schemas.User)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user

# --- Item Endpoints ---
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
async def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

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

@app.get("/items/{item_id}", response_model=schemas.Item)
async def read_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return item

ในตัวอย่างนี้ เราใช้ Dependency get_db เพื่อสร้างและปิด SQLAlchemy Session สำหรับแต่ละคำขอครับ Pydantic Model (ใน schemas.py) มี Config.orm_mode = True เพื่อให้สามารถอ่านข้อมูลจาก SQLAlchemy ORM Model ได้โดยตรง ซึ่งช่วยให้การทำงานร่วมกันราบรื่นครับ สำหรับการใช้งานจริง ควรพิจารณาใช้ Alembic สำหรับการทำ Database Migration ด้วยนะครับ และสำหรับโปรเจกต์ที่ต้องการประสิทธิภาพสูงสุด ควรใช้ databases library หรือ ORM ที่เป็น async-native เช่น Tortoise ORM หรือ SQLModel ครับ

APIRouter: การจัดโครงสร้างโปรเจกต์ขนาดใหญ่

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

สร้างไฟล์ routers/items.py:


# routers/items.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from .. import schemas, models
from ..database import get_db

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

@router.post("/", response_model=schemas.Item)
async def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    db_item = models.Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@router.get("/", response_model=List[schemas.Item])
async def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(models.Item).offset(skip).limit(limit).all()
    return items

@router.get("/{item_id}", response_model=schemas.Item)
async def read_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(models.Item).filter(models.Item.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return item

สร้างไฟล์ routers/users.py:


# routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from .. import schemas, models
from ..database import get_db

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

@router.post("/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered")
    
    hashed_password = user.password + "notreallyhashed" # ควรแฮชจริง
    db_user = models.User(email=user.email, hashed_password=hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@router.get("/", response_model=List[schemas.User])
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = db.query(models.User).offset(skip).limit(limit).all()
    return users

@router.get("/{user_id}", response_model=schemas.User)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    return user

แก้ไขไฟล์ main.py เพื่อรวม Router:


# main.py
from fastapi import FastAPI
from . import models, database # ต้อง import แบบ relative
from .routers import items, users # Import routers

app = FastAPI(
    title="SiamLancard FastAPI API",
    description="API สำหรับเว็บไซต์ SiamLancard.com",
    version="1.0.0",
)

# สร้างตารางในฐานข้อมูลเมื่อแอปพลิเคชันเริ่มต้น
@app.on_event("startup")
async def startup_event():
    models.Base.metadata.create_all(bind=database.engine)
    await database.database.connect()

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

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

@app.get("/")
async def read_root():
    return {"message": "Welcome to SiamLancard FastAPI! Go to /docs for API documentation."}

ด้วย APIRouter โค้ดของเราจะมีความเป็นระเบียบมากขึ้น แต่ละโมดูลรับผิดชอบ Path Operations ของตัวเอง และ main.py ก็จะทำหน้าที่เป็นจุดรวมทั้งหมดครับ

Middleware: การประมวลผลคำขอก่อนถึง Endpoint

Middleware คือฟังก์ชันที่ทำงานระหว่างการรับคำขอจาก Client และการประมวลผลคำขอโดย Path Operation ของเราครับ มันมีประโยชน์สำหรับการทำงานที่ต้องการทำซ้ำๆ กับทุกคำขอ เช่น การบันทึก Log, การตรวจสอบ CORS, หรือการยืนยันตัวตนบางอย่าง


from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time

app = FastAPI()

# --- CORS Middleware ---
origins = [
    "http://localhost",
    "http://localhost:8080",
    "https://www.siamlancard.com", # เพิ่มโดเมนของเว็บไซต์คุณ
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"], # อนุญาตทุก HTTP method
    allow_headers=["*"], # อนุญาตทุก Header
)

# --- Custom Middleware ---
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        response = await call_next(request)
        process_time = time.time() - start_time
        response.headers["X-Process-Time"] = str(process_time)
        print(f"Request took {process_time:.4f} seconds") # ตัวอย่างการบันทึก Log
        return response

app.add_middleware(TimingMiddleware)

@app.get("/hello")
async def hello():
    return {"message": "Hello after middleware!"}

@app.get("/long-task")
async def long_task():
    time.sleep(2) # จำลองการทำงานที่ใช้เวลานาน
    return {"message": "Task completed after 2 seconds"}

ในตัวอย่างนี้ เราได้เพิ่ม Middleware สองตัว:

  1. CORSMiddleware: สำหรับจัดการ Cross-Origin Resource Sharing ซึ่งจำเป็นเมื่อ Client (เช่น เว็บไซต์ Frontend) อยู่คนละโดเมนกับ API Server ครับ
  2. TimingMiddleware: Middleware ที่เราสร้างขึ้นเองเพื่อวัดเวลาที่ใช้ในการประมวลผลคำขอแต่ละครั้ง และเพิ่ม Header X-Process-Time ใน Response ครับ

Middleware จะทำงานตามลำดับที่ถูกเพิ่มเข้ามาครับ ดังนั้น CORSMiddleware จะทำงานก่อน และ TimingMiddleware จะทำงานหลังจากนั้นครับ

การทดสอบ API ด้วย FastAPI TestClient

การทดสอบเป็นส่วนสำคัญของการพัฒนาซอฟต์แวร์ครับ FastAPI มี TestClient ที่สร้างขึ้นบน Starlette TestClient ซึ่งช่วยให้เราสามารถทดสอบ Path Operations ได้อย่างง่ายดายโดยไม่ต้องรันเซิร์ฟเวอร์จริงครับ

ติดตั้ง pytest:


pip install pytest

สร้างไฟล์ test_main.py:



# test_main.py

from fastapi.testclient import TestClient

from .main import app # import app จากไฟล์ main.py ของคุณ


# สร้าง TestClient สำหรับแอปพลิเคชัน FastAPI ของเรา

client = TestClient(app)


def test_read_root():

    response = client.get("/")

    assert response.status_code == 200

    assert response.json() == {"message": "Welcome to SiamLancard FastAPI! Go to /docs for API documentation."}


def test_create_user():

    response = client.post(

        "/users/",

        json={"email": "[email protected]", "password": "testpassword"},

    )

    assert response.status_code == 200

    assert response.json()["email"] == "[email protected]"

    assert "id" in response.json()

    assert response.json()["is_active"] is True


    # ลองสร้างผู้ใช้ซ้ำ ควรได้ข้อผิดพลาด 400

    response_duplicate = client.post(

        "/users/",

        json={"email": "[email protected]", "password": "testpassword"},

    )

    assert response_duplicate.status_code == 400

    assert response_duplicate.json()["detail"] == "Email already registered"


def test_read_users():

    response = client.get("/users/")

    assert response.status_code == 200

    assert isinstance(response.json(), list

    	

บทความที่เกี่ยวข้อง
Siam2R.com — Portfolio งาน IT · XM Signal — Free Forex EA Download

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