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

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

สารบัญ

• REST API คืออะไร? ทำไมถึงสำคัญ?
• ทำไมต้อง FastAPI? ข้อได้เปรียบเหนือเฟรมเวิร์กอื่น ๆ
• เตรียมความพร้อมก่อนเริ่มต้น
• การตั้งค่าสภาพแวดล้อมการพัฒนา
  • ติดตั้ง Python
  • สร้าง Virtual Environment
  • ติดตั้ง FastAPI และ Uvicorn
• สร้าง FastAPI Application ตัวแรก: “Hello, World!”
  • การรัน FastAPI Application
  • สำรวจเอกสาร API อัตโนมัติ (Swagger UI / ReDoc)
• การใช้งาน Path Parameters
• การใช้งาน Query Parameters
• การจัดการ Request Body ด้วย Pydantic Model
  • สร้าง Pydantic Model
  • ส่งข้อมูลด้วย HTTP POST
• ทำความเข้าใจ HTTP Methods และการสร้าง CRUD API
  • GET: ดึงข้อมูลทั้งหมด
  • GET: ดึงข้อมูลตาม ID
  • POST: สร้างข้อมูลใหม่
  • PUT: อัปเดตข้อมูลทั้งหมด
  • PATCH: อัปเดตข้อมูลบางส่วน
  • DELETE: ลบข้อมูล
• การตรวจสอบข้อมูลและการจัดการข้อผิดพลาด
  • การตรวจสอบข้อมูลด้วย Pydantic
  • การจัดการข้อผิดพลาดแบบกำหนดเอง
• Dependency Injection ใน FastAPI
  • ตัวอย่าง Dependency อย่างง่าย
  • Dependency ที่มีพารามิเตอร์
• การเชื่อมต่อกับฐานข้อมูล (SQLite และ SQLAlchemy)
  • ติดตั้ง SQLAlchemy และ Pydantic Settings
  • กำหนด Database Models
  • สร้าง CRUD Operations กับฐานข้อมูล
  • Dependency สำหรับ Database Session
• การยืนยันตัวตน (Authentication) และการอนุญาต (Authorization) ด้วย JWT
  • ภาพรวม JWT
  • ติดตั้งไลบรารีที่จำเป็น
  • สร้าง User Model
  • ฟังก์ชันสำหรับ JWT
  • สร้าง Endpoint สำหรับ Login และ Protected Route
• การนำ API ไปใช้งานจริง (Deployment Considerations)
  • Uvicorn และ Gunicorn
  • การใช้งาน Docker
  • แพลตฟอร์มคลาวด์
• คำถามที่พบบ่อย (FAQ)
• สรุปและ Call to Action

REST API คืออะไร? ทำไมถึงสำคัญ?

ก่อนที่เราจะเริ่มสร้าง API ด้วย FastAPI เรามาทำความเข้าใจพื้นฐานของ REST API กันก่อนครับ REST ย่อมาจาก Representational State Transfer ซึ่งเป็นสถาปัตยกรรม (architectural style) สำหรับการออกแบบระบบเครือข่าย โดยเฉพาะอย่างยิ่งสำหรับเว็บเซอร์วิส ที่เน้นการสื่อสารแบบไร้สถานะ (stateless) และใช้ HTTP Protocol เป็นหลักครับ

หลักการสำคัญของ REST API คือการมองทุกอย่างเป็น “ทรัพยากร” (Resource) ซึ่งสามารถเข้าถึงได้ด้วย URL (Uniform Resource Locator) และมีการดำเนินการกับทรัพยากรเหล่านั้นผ่าน HTTP Methods มาตรฐาน เช่น:

• GET: ใช้สำหรับดึงข้อมูลทรัพยากร
• POST: ใช้สำหรับสร้างทรัพยากรใหม่
• PUT: ใช้สำหรับอัปเดตทรัพยากรทั้งหมด
• PATCH: ใช้สำหรับอัปเดตทรัพยากรบางส่วน
• DELETE: ใช้สำหรับลบทรัพยากร

REST API มีความสำคัญอย่างยิ่งในโลกของการพัฒนาซอฟต์แวร์สมัยใหม่ เพราะช่วยให้แอปพลิเคชันต่าง ๆ ไม่ว่าจะเป็นเว็บเบราว์เซอร์, โมบายล์แอป, หรือแม้แต่บริการ Backend ด้วยกัน สามารถสื่อสารและแลกเปลี่ยนข้อมูลกันได้อย่างเป็นมาตรฐาน ทำให้เกิดความยืดหยุ่นในการพัฒนาและสามารถผสานรวมระบบต่าง ๆ เข้าด้วยกันได้อย่างง่ายดายครับ

ทำไมต้อง FastAPI? ข้อได้เปรียบเหนือเฟรมเวิร์กอื่น ๆ

ในบรรดาเฟรมเวิร์ก Python สำหรับการสร้าง API นั้นมีให้เลือกมากมาย เช่น Flask, Django REST Framework แต่ FastAPI ได้รับความนิยมอย่างรวดเร็วและโดดเด่นขึ้นมาด้วยเหตุผลหลายประการครับ

FastAPI สร้างขึ้นบน Starlette สำหรับส่วนของ Web และ Pydantic สำหรับส่วนของการจัดการข้อมูล ทำให้มันได้ประโยชน์จากทั้งสองไลบรารีนี้อย่างเต็มที่ครับ

• ประสิทธิภาพสูง (High Performance): FastAPI เป็นหนึ่งในเฟรมเวิร์ก Python ที่เร็วที่สุด สามารถรองรับ Request ได้จำนวนมากเทียบเท่ากับ Node.js และ Go ด้วยความสามารถในการทำงานแบบ Asynchronous (async/await) ครับ
• พัฒนาได้รวดเร็ว (Fast to Code): ช่วยลดเวลาในการพัฒนาลง 20% ถึง 40% เพราะคุณไม่ต้องเขียนโค้ดซ้ำซ้อนสำหรับ Data Validation และ Serialization ครับ
• ลดข้อผิดพลาด (Fewer Bugs): ด้วยระบบ Type Hint ของ Python และ Pydantic ทำให้มีการตรวจสอบข้อมูลในเวลาคอมไพล์ (compile-time) และรันไทม์ (run-time) ช่วยลดข้อผิดพลาดที่เกี่ยวกับชนิดของข้อมูลได้มากถึง 40% ครับ
• เอกสาร API อัตโนมัติ (Automatic Docs): FastAPI สร้างเอกสาร API แบบอินเทอร์แอคทีฟให้อัตโนมัติในรูปแบบ Swagger UI และ ReDoc ซึ่งช่วยให้นักพัฒนา Frontend และ Backend ทำงานร่วมกันได้ง่ายขึ้นมากครับ
• รองรับ Python Type Hints: ใช้ประโยชน์จาก Type Hints ของ Python อย่างเต็มที่ ทำให้โค้ดอ่านง่ายขึ้น มีการตรวจสอบชนิดข้อมูลที่ดีขึ้น และรองรับการทำงานร่วมกับ IDEs ได้ดียิ่งขึ้นครับ
• Dependency Injection ที่ทรงพลัง: ระบบ Dependency Injection ที่ใช้งานง่ายและมีประสิทธิภาพ ช่วยให้การจัดการ Dependency ของฟังก์ชันต่าง ๆ เป็นไปอย่างราบรื่นและทดสอบได้ง่ายครับ

เพื่อให้เห็นภาพชัดเจนยิ่งขึ้น ลองดูตารางเปรียบเทียบ FastAPI กับเฟรมเวิร์กยอดนิยมอื่น ๆ ครับ:

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

ก่อนที่เราจะดำดิ่งสู่การเขียนโค้ด คุณต้องแน่ใจว่าได้ติดตั้งเครื่องมือที่จำเป็นสำหรับการพัฒนา FastAPI API ครับ

• Python 3.7+: FastAPI ใช้ประโยชน์จากคุณสมบัติใหม่ ๆ ของ Python โดยเฉพาะ Type Hints และ Async/Await ดังนั้นต้องเป็น Python เวอร์ชัน 3.7 ขึ้นไปครับ
• Text Editor / IDE: เช่น VS Code, PyCharm, Sublime Text หรือ Atom
• Command Line Interface (CLI): สำหรับรันคำสั่งต่าง ๆ ครับ

การตั้งค่าสภาพแวดล้อมการพัฒนา

ติดตั้ง Python

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

เมื่อติดตั้งแล้ว ให้ลองตรวจสอบเวอร์ชันของ Python ใน Command Line:

python3 --version
# หรือ
python --version

ควรจะได้ผลลัพธ์ประมาณ Python 3.x.x ครับ

สร้าง Virtual Environment

การใช้ Virtual Environment เป็นแนวปฏิบัติที่ดีในการพัฒนา Python เพื่อแยก Dependency ของโปรเจกต์ออกจากกันครับ

สร้างโฟลเดอร์สำหรับโปรเจกต์ของคุณและเข้าไปในโฟลเดอร์นั้น:

mkdir my-fastapi-app
cd my-fastapi-app

สร้าง Virtual Environment:

python3 -m venv venv

เปิดใช้งาน Virtual Environment:

# บน macOS/Linux
source venv/bin/activate

# บน Windows (Command Prompt)
venv\Scripts\activate.bat

# บน Windows (PowerShell)
venv\Scripts\Activate.ps1

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

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

Uvicorn เป็น ASGI server (Asynchronous Server Gateway Interface) ที่ FastAPI ใช้ในการรันแอปพลิเคชันของเราครับ

pip install fastapi "uvicorn[standard]"

ตอนนี้คุณพร้อมที่จะสร้าง FastAPI API ตัวแรกแล้วครับ!

สร้าง FastAPI Application ตัวแรก: “Hello, World!”

มาสร้างไฟล์ main.py ในโฟลเดอร์โปรเจกต์ของคุณ และเพิ่มโค้ดต่อไปนี้ครับ

# main.py
from fastapi import FastAPI

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

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

# คุณสามารถเพิ่ม Path Operation อื่นๆ ได้ที่นี่
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    """
    Endpoint สำหรับดึงข้อมูล Item ตาม ID
    """
    return {"item_id": item_id, "q": q}

จากโค้ดด้านบน:

• เรานำเข้า FastAPI จากไลบรารี fastapi ครับ
• สร้าง instance ของ FastAPI ชื่อ app
• ใช้ decorator @app.get("/") เพื่อระบุว่าฟังก์ชัน read_root จะถูกเรียกเมื่อมี HTTP GET request เข้ามาที่ root path (/) ครับ
• ฟังก์ชัน read_root ถูกกำหนดให้เป็น async เพราะ FastAPI ถูกออกแบบมาเพื่อรองรับ Asynchronous Operation โดยธรรมชาติครับ
• ฟังก์ชันจะคืนค่าเป็น Python dictionary ซึ่ง FastAPI จะแปลงเป็น JSON response อัตโนมัติครับ

การรัน FastAPI Application

เปิด Command Line ของคุณ (ตรวจสอบให้แน่ใจว่า Virtual Environment ยังทำงานอยู่) และรันคำสั่ง Uvicorn:

uvicorn main:app --reload

คำอธิบาย:

• main: คือชื่อโมดูล Python (ไฟล์ main.py) ครับ
• app: คือชื่อของ instance FastAPI ที่เราสร้างขึ้นในไฟล์ main.py (app = FastAPI()) ครับ
• --reload: เป็น flag ที่ทำให้ Uvicorn รีโหลดเซิร์ฟเวอร์อัตโนมัติเมื่อมีการเปลี่ยนแปลงโค้ด ซึ่งมีประโยชน์มากสำหรับการพัฒนาครับ

คุณจะเห็นข้อความคล้าย ๆ กับ:

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 คุณจะเห็น JSON response: {"message": "Hello, World!"} ครับ

สำรวจเอกสาร API อัตโนมัติ (Swagger UI / ReDoc)

ความเจ๋งของ FastAPI คือมันสร้างเอกสาร API ให้อัตโนมัติครับ ลองไปที่ URL เหล่านี้ในเบราว์เซอร์ของคุณ:

• Swagger UI: http://127.0.0.1:8000/docs
• ReDoc: http://127.0.0.1:8000/redoc

คุณจะเห็นหน้าเอกสาร API ที่สวยงามและสามารถโต้ตอบได้ ซึ่งช่วยให้คุณทดสอบ Endpoint และทำความเข้าใจโครงสร้าง API ของคุณได้อย่างง่ายดายครับ

การใช้งาน Path Parameters

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

ตัวอย่างใน main.py:

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI

app = FastAPI()

# ... (โค้ด read_root และ read_item เดิม)

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

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    return {"user_id": user_id}

ในตัวอย่าง read_item เรากำหนด item_id: int ซึ่งหมายความว่า FastAPI จะตรวจสอบให้แน่ใจว่า item_id ที่ส่งมาเป็นตัวเลขจำนวนเต็ม ถ้าไม่ใช่ จะส่งข้อผิดพลาดกลับไปอัตโนมัติครับ นี่คือพลังของ Type Hints และ Pydantic ที่ FastAPI ใช้ครับ

ลองไปที่ http://127.0.0.1:8000/items/5 หรือ http://127.0.0.1:8000/users/john_doe ครับ

การใช้งาน Query Parameters

Query Parameters คือพารามิเตอร์ที่ส่งมากับ URL หลังเครื่องหมายคำถาม (?) มักใช้สำหรับการกรอง, การจัดเรียง หรือการแบ่งหน้า (pagination) ข้อมูลครับ

ตัวอย่างใน main.py:

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# ... (โค้ดเดิม)

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    """
    Endpoint สำหรับดึงรายการ Items พร้อม Query Parameters สำหรับ pagination
    """
    return {"message": f"Fetching items with skip={skip} and limit={limit}"}

@app.get("/products/{product_id}")
async def get_product(product_id: int, query: Optional[str] = None, price: float = 0.0):
    """
    Endpoint สำหรับดึงข้อมูลสินค้าพร้อม Query Parameters
    """
    return {"product_id": product_id, "query": query, "price": price}

ในฟังก์ชัน read_items:

• skip: int = 0 และ limit: int = 10 เป็น Query Parameters ที่มีค่าเริ่มต้น ถ้าไม่มีการระบุค่ามา FastAPI จะใช้ค่าเริ่มต้นเหล่านี้ครับ
• FastAPI ยังคงใช้ Type Hints เพื่อตรวจสอบชนิดข้อมูลของ Query Parameters อัตโนมัติครับ

ลองทดสอบ:

• http://127.0.0.1:8000/items/ (ใช้ค่าเริ่มต้น skip=0, limit=10)
• http://127.0.0.1:8000/items/?skip=20&limit=5
• http://127.0.0.1:8000/products/123?query=shoes&price=99.99

การจัดการ Request Body ด้วย Pydantic Model

เมื่อเราต้องการส่งข้อมูลที่ซับซ้อนไปยัง API เช่น การสร้างหรืออัปเดตข้อมูล เราจะใช้ Request Body ครับ FastAPI ใช้ไลบรารี Pydantic สำหรับการกำหนดโครงสร้างข้อมูล การตรวจสอบความถูกต้องของข้อมูล (validation) และการแปลงข้อมูล (serialization/deserialization) ซึ่งเป็นจุดเด่นที่ทำให้ FastAPI มีประสิทธิภาพสูงครับ

สร้าง Pydantic Model

เราจะสร้าง Pydantic Model สำหรับ Item ของเราครับ

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# ... (โค้ดเดิม)

# สร้าง Pydantic Model สำหรับ Item
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# สร้าง Pydantic Model สำหรับ User (ตัวอย่างเพิ่มเติม)
class User(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None
    disabled: Optional[bool] = False

ในคลาส Item:

• name: str: กำหนดให้ name เป็น string และเป็นฟิลด์ที่จำเป็น (required) ครับ
• description: Optional[str] = None: กำหนดให้ description เป็น string และเป็นฟิลด์ที่ไม่จำเป็น (optional) โดยมีค่าเริ่มต้นเป็น None ครับ
• price: float: กำหนดให้ price เป็น float และเป็นฟิลด์ที่จำเป็นครับ
• tax: Optional[float] = None: กำหนดให้ tax เป็น float และเป็นฟิลด์ที่ไม่จำเป็นครับ

Pydantic จะทำการตรวจสอบชนิดข้อมูลและข้อกำหนดอื่น ๆ ให้อัตโนมัติ และถ้าข้อมูลไม่ถูกต้อง จะส่งข้อผิดพลาดที่ชัดเจนกลับไปครับ

ส่งข้อมูลด้วย HTTP POST

ทีนี้เรามาสร้าง Endpoint สำหรับการสร้าง Item ใหม่โดยใช้ HTTP POST และรับ Request Body ที่เป็น Pydantic Model ของเราครับ

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# ... (Pydantic Models และโค้ดเดิม)

@app.post("/items/")
async def create_item(item: Item):
    """
    Endpoint สำหรับสร้าง 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.post("/users/")
async def create_user(user: User):
    """
    Endpoint สำหรับสร้าง User ใหม่
    """
    return user

เมื่อคุณส่ง HTTP POST request ไปที่ /items/ พร้อม JSON body ที่ตรงกับโครงสร้างของ Item model, FastAPI จะ:

• อ่าน Request Body
• แปลง JSON เป็น Python dictionary
• ตรวจสอบข้อมูลใน dictionary นั้นกับ Item model ด้วย Pydantic
• หากข้อมูลถูกต้อง จะสร้าง instance ของ Item และส่งผ่านไปยังฟังก์ชัน create_item ครับ
• หากข้อมูลไม่ถูกต้อง (เช่น name ไม่ใช่ string หรือ price ไม่มี) FastAPI จะส่ง HTTP 422 Unprocessable Entity พร้อมรายละเอียดข้อผิดพลาดกลับไปโดยอัตโนมัติครับ

คุณสามารถทดสอบ Endpoint นี้ได้จากหน้า Swagger UI (/docs) โดยเลือกเมนู POST /items/ แล้วคลิก “Try it out” และใส่ JSON body ลงไปครับ ตัวอย่าง Request Body:

{
  "name": "Foo",
  "description": "A very nice Item",
  "price": 35.4,
  "tax": 3.2
}

ทำความเข้าใจ HTTP Methods และการสร้าง CRUD API

การสร้าง REST API มักจะเกี่ยวข้องกับการดำเนินการพื้นฐาน 4 อย่าง หรือที่เรียกว่า CRUD: Create (สร้าง), Read (อ่าน), Update (อัปเดต), และ Delete (ลบ) ครับ เราจะใช้ HTTP Methods ต่าง ๆ เพื่อให้สอดคล้องกับการดำเนินการเหล่านี้

เพื่อความเรียบง่าย เราจะเก็บข้อมูลในหน่วยความจำ (in-memory) ก่อนครับ

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    id: Optional[int] = None # เพิ่ม id เข้ามาเพื่อใช้ในการอ้างอิง
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# เก็บข้อมูล Items ชั่วคราวในหน่วยความจำ
fake_items_db = []
next_item_id = 1

GET: ดึงข้อมูลทั้งหมด

# main.py (เพิ่มในไฟล์เดิม)
@app.get("/items/", response_model=List[Item])
async def read_all_items():
    """
    ดึงข้อมูล Item ทั้งหมด
    """
    return fake_items_db

เราใช้ response_model=List[Item] เพื่อบอก FastAPI ว่า Response ที่ส่งกลับไปควรมีโครงสร้างเป็น List ของ Item ครับ สิ่งนี้ช่วยให้ Swagger UI แสดงเอกสาร Response ที่ถูกต้อง และ FastAPI จะทำการตรวจสอบ Response ด้วย Pydantic ด้วยครับ

GET: ดึงข้อมูลตาม ID

# main.py (เพิ่มในไฟล์เดิม)
@app.get("/items/{item_id}", response_model=Item)
async def read_single_item(item_id: int):
    """
    ดึงข้อมูล Item ตาม ID
    """
    for item in fake_items_db:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

หากไม่พบ Item ด้วย ID ที่ระบุ เราจะใช้ raise HTTPException เพื่อส่ง HTTP 404 Not Found กลับไปครับ

POST: สร้างข้อมูลใหม่

# main.py (เพิ่มในไฟล์เดิม)
@app.post("/items/", response_model=Item, status_code=201)
async def create_new_item(item: Item):
    """
    สร้าง Item ใหม่
    """
    global next_item_id
    item.id = next_item_id
    next_item_id += 1
    fake_items_db.append(item)
    return item

เราใช้ status_code=201 เพื่อระบุว่าเมื่อสร้างทรัพยากรสำเร็จ ควรคืนค่า HTTP 201 Created ครับ

PUT: อัปเดตข้อมูลทั้งหมด

PUT ใช้สำหรับอัปเดตทรัพยากรทั้งหมด หรือสร้างใหม่หากไม่มีอยู่ครับ

# main.py (เพิ่มในไฟล์เดิม)
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, item: Item):
    """
    อัปเดตข้อมูล Item ทั้งหมดตาม ID
    """
    for i, existing_item in enumerate(fake_items_db):
        if existing_item.id == item_id:
            item.id = item_id # ให้แน่ใจว่า ID ไม่เปลี่ยน
            fake_items_db[i] = item
            return item
    raise HTTPException(status_code=404, detail="Item not found")

PATCH: อัปเดตข้อมูลบางส่วน

PATCH ใช้สำหรับอัปเดตข้อมูลบางส่วนของทรัพยากร โดยปกติจะใช้ร่วมกับ Pydantic model ที่มีฟิลด์เป็น Optional ทั้งหมดครับ

# main.py (เพิ่มในไฟล์เดิม)
from pydantic import BaseModel
from typing import List, Optional

# ... (โค้ดเดิม)

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

@app.patch("/items/{item_id}", response_model=Item)
async def partial_update_item(item_id: int, item_update: ItemUpdate):
    """
    อัปเดตข้อมูล Item บางส่วนตาม ID
    """
    for i, existing_item in enumerate(fake_items_db):
        if existing_item.id == item_id:
            update_data = item_update.dict(exclude_unset=True) # ไม่รวมฟิลด์ที่ไม่ได้ถูกตั้งค่า
            for key, value in update_data.items():
                setattr(existing_item, key, value)
            return existing_item
    raise HTTPException(status_code=404, detail="Item not found")

ใน partial_update_item เราสร้าง ItemUpdate model ที่ทุกฟิลด์เป็น Optional ครับ และใช้ item_update.dict(exclude_unset=True) เพื่อให้ได้เฉพาะฟิลด์ที่มีการส่งค่าเข้ามาใน request body เท่านั้น

DELETE: ลบข้อมูล

# main.py (เพิ่มในไฟล์เดิม)
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    """
    ลบ Item ตาม ID
    """
    global fake_items_db
    initial_len = len(fake_items_db)
    fake_items_db = [item for item in fake_items_db if item.id != item_id]
    if len(fake_items_db) == initial_len:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

เมื่อลบทรัพยากรสำเร็จ มักจะคืนค่า HTTP 204 No Content ครับ

การตรวจสอบข้อมูลและการจัดการข้อผิดพลาด

FastAPI มีระบบการตรวจสอบข้อมูลและการจัดการข้อผิดพลาดที่มีประสิทธิภาพในตัวครับ

การตรวจสอบข้อมูลด้วย Pydantic

อย่างที่เราได้เห็นไปแล้ว Pydantic ทำการตรวจสอบข้อมูลที่เข้ามาใน Request Body และ Query/Path Parameters โดยอัตโนมัติ ตาม Type Hints ที่เรากำหนดไว้ครับ

ตัวอย่างการเพิ่มเงื่อนไขการตรวจสอบเพิ่มเติม:

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI, HTTPException, Body
from pydantic import BaseModel, Field
from typing import List, Optional

app = FastAPI()

# ... (โค้ด Item Model เดิม)

class ItemWithValidation(BaseModel):
    name: str = Field(min_length=3, max_length=50) # ชื่อต้องมีความยาว 3-50 ตัวอักษร
    description: Optional[str] = Field(None, min_length=10, max_length=200)
    price: float = Field(..., gt=0, description="The price must be greater than zero") # ราคาต้องมากกว่า 0
    tax: Optional[float] = Field(None, le=0.2) # ภาษีต้องน้อยกว่าหรือเท่ากับ 0.2

@app.post("/items_with_validation/", response_model=ItemWithValidation)
async def create_item_with_validation(item: ItemWithValidation):
    """
    สร้าง Item พร้อมการตรวจสอบข้อมูลเพิ่มเติม
    """
    return item

ใน ItemWithValidation เราใช้ Field จาก Pydantic เพื่อเพิ่มเงื่อนไขการตรวจสอบที่ละเอียดยิ่งขึ้น เช่น min_length, max_length, gt (greater than), le (less than or equal) ครับ

การจัดการข้อผิดพลาดแบบกำหนดเอง

FastAPI ใช้ HTTPException เพื่อจัดการข้อผิดพลาด HTTP โดยเฉพาะครับ

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse

app = FastAPI()

# ... (โค้ดเดิม)

@app.get("/items/{item_id}/info")
async def get_item_info(item_id: int):
    """
    ดึงข้อมูล Item และแสดงข้อผิดพลาดแบบกำหนดเอง
    """
    if item_id == 0:
        # ส่ง HTTP 400 Bad Request
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Item ID cannot be zero")
    if item_id not in [1, 2, 3]: # สมมติว่ามีแค่ item id 1, 2, 3
        # ส่ง HTTP 404 Not Found
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found", headers={"X-Error": "There goes my error"})
    return {"item_id": item_id, "name": f"Item {item_id}"}

# คุณยังสามารถสร้าง Custom Exception Handlers ได้
class CustomException(Exception):
    def __init__(self, name: str):
        self.name = name

@app.exception_handler(CustomException)
async def custom_exception_handler(request, exc: CustomException):
    return JSONResponse(
        status_code=status.HTTP_418_IM_A_TEAPOT, # ตัวอย่างสถานะ HTTP ที่ไม่ปกติ
        content={"message": f"Oops! {exc.name} did something wrong."},
    )

@app.get("/teapot/{name}")
async def tea_time(name: str):
    if name == "coffee":
        raise CustomException(name=name)
    return {"message": f"Enjoy your {name}!"}

# main.py (เพิ่มในไฟล์เดิม)
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# ... (โค้ดเดิม)

# Dependency Function
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items_di/")
async def read_items_di(commons: dict = Depends(common_parameters)):
    """
    Endpoint ที่ใช้ Dependency Injection สำหรับ Query Parameters ทั่วไป
    """
    return commons

# main.py (เพิ่มในไฟล์เดิม)
async def verify_token(token: str):
    if token != "mysecrettoken":
        raise HTTPException(status_code=400, detail="Invalid token")
    return token

@app.get("/secure_data/", dependencies=[Depends(verify_token)])
async def read_secure_data():
    """
    Endpoint ที่ต้องมีการยืนยัน Token ก่อนเข้าถึง
    """
    return {"message": "This is secure data!"}

pip install sqlalchemy "python-dotenv[cli]" # python-dotenv สำหรับจัดการ environment variables

# database.py
from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.sql import text
from dotenv import load_dotenv
import os

load_dotenv() # โหลด environment variables จาก .env

DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./sql_app.db") # ใช้ SQLite เป็นค่าเริ่มต้น

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) # สำหรับ SQLite

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# Database Model
class DBItem(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    price = Column(Float)
    tax = Column(Float, nullable=True)

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

# สร้างไฟล์ .env
# .env
# DATABASE_URL="sqlite:///./sql_app.db"

# main.py (ปรับปรุง Pydantic Item Model)
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
from sqlalchemy.orm import Session
import database # นำเข้า database.py

# สร้างตารางในฐานข้อมูลเมื่อแอปเริ่มทำงาน
database.create_db_tables()

# Pydantic Model สำหรับ Item ที่รับเข้ามาจาก Request Body
class ItemCreate(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# Pydantic Model สำหรับ Item ที่ส่งกลับไป (รวม ID)
class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

    class Config: # กำหนดให้ Pydantic สามารถอ่านจาก ORM objects ได้
        orm_mode = True

app = FastAPI()

# main.py (เพิ่มในไฟล์เดิม)

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

# POST: สร้าง Item ใหม่ในฐานข้อมูล
@app.post("/db_items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_db_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = database.DBItem(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# GET: ดึง Item ทั้งหมดจากฐานข้อมูล
@app.get("/db_items/", response_model=List[Item])
async def read_db_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = db.query(database.DBItem).offset(skip).limit(limit).all()
    return items

# GET: ดึง Item ตาม ID จากฐานข้อมูล
@app.get("/db_items/{item_id}", response_model=Item)
async def read_db_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(database.DBItem).filter(database.DBItem.id == item_id).first()
    if item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    return item

# PUT: อัปเดต Item ในฐานข้อมูล
@app.put("/db_items/{item_id}", response_model=Item)
async def update_db_item(item_id: int, item: ItemCreate, db: Session = Depends(get_db)):
    db_item = db.query(database.DBItem).filter(database.DBItem.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")

    for key, value in item.dict(exclude_unset=True).items(): # ใช้ exclude_unset สำหรับ PATCH หรือ PUT
        setattr(db_item, key, value)
    
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

# DELETE: ลบ Item จากฐานข้อมูล
@app.delete("/db_items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_db_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(database.DBItem).filter(database.DBItem.id == item_id).first()
    if db_item is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
    
    db.delete(db_item)
    db.commit()
    return None

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

# main.py (เพิ่มในไฟล์เดิม)
from datetime import datetime, timedelta
from typing import Optional

from jose import JWTError, jwt
from passlib.context import CryptContext

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

app = FastAPI()

# ... (โค้ด Item และ DB Item Model เดิม)

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

class UserInDB(User):
    hashed_password: str

# JWT Configuration
SECRET_KEY = "your-secret-key" # ควรเก็บใน Environment Variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Password Hashing Context
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2PasswordBearer สำหรับดึง token จาก Authorization header
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ฟังก์ชันช่วยสำหรับ JWT
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 verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

# จำลองฐานข้อมูลผู้ใช้ (ในความเป็นจริงควรเชื่อมกับ DB)
fake_users_db = {
    "john_doe": {
        "username": "john_doe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": get_password_hash("secret"),
        "disabled": False,
    },
    "jane_smith": {
        "username": "jane_smith",
        "email": "[email protected]",
        "full_name": "Jane Smith",
        "hashed_password": get_password_hash("anothersecret"),
        "disabled": True,
    },
}

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

async def authenticate_user(username: str, password: str):
    user = get_user_from_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"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        user = get_user_from_db(username)
        if user is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return user

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

# main.py (เพิ่มในไฟล์เดิม)

# Endpoint สำหรับ Login และรับ Access Token
@app.post("/token", response_model=dict)
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"}

# Endpoint ที่ต้องการการยืนยันตัวตน
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    """
    ดึงข้อมูลผู้ใช้ปัจจุบันที่ล็อกอินอยู่
    """
    return current_user

# ตัวอย่าง Endpoint ที่ต้องการการยืนยันตัวตนและเป็นผู้ดูแลระบบ
@app.get("/admin/data")
async def read_admin_data(current_user: User = Depends(get_current_active_user)):
    # ในความเป็นจริงควรมี logic ตรวจสอบ role/permission ของ user
    if current_user.username != "john_doe": # ตัวอย่างง่ายๆ สำหรับ admin
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
    return {"message": f"Welcome, admin {current_user.username}! Here is your secret data."}