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

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

สารบัญ

• ทำความเข้าใจ REST API และ FastAPI
  • REST API คืออะไร? หลักการสำคัญที่คุณควรรู้
  • ทำไมต้อง FastAPI? เจาะลึกจุดเด่นที่ทำให้เป็นตัวเลือกที่เหนือกว่า
  • FastAPI กับ Web Frameworks อื่นๆ: ตารางเปรียบเทียบ
• เตรียมความพร้อม: ติดตั้งและตั้งค่าโปรเจกต์ FastAPI
  • ข้อกำหนดเบื้องต้นและ Virtual Environment
  • ติดตั้ง FastAPI และ Uvicorn
  • โครงสร้างโปรเจกต์เบื้องต้น
• สร้าง REST API แรกของคุณ: Hello World และ Path Operations
  • Hello World API: เริ่มต้นง่ายๆ
  • Path Parameters: การรับค่าผ่าน URL
  • Query Parameters: การกรองและจำกัดข้อมูล
  • Request Body และ Pydantic: การส่งข้อมูลและการตรวจสอบ
  • การจัดการ CRUD Operations (GET, POST, PUT, DELETE, PATCH)
• คุณสมบัติขั้นสูงของ FastAPI ที่จะยกระดับ API ของคุณ
  • Dependencies: การจัดการการพึ่งพาและการฉีดค่า
  • Security และ Authentication: การรักษาความปลอดภัย API ของคุณ
  • Database Integration: เชื่อมต่อและจัดการข้อมูล (SQLAlchemy)
  • Error Handling: การจัดการข้อผิดพลาดอย่างมืออาชีพ
  • Background Tasks: การทำงานเบื้องหลังที่ไม่บล็อก API
  • Middleware: การประมวลผลคำขอก่อนถึง Path Operations
  • Testing: การทดสอบ API ด้วย TestClient
• เอกสาร API อัตโนมัติ: Swagger UI และ ReDoc
• การ Deploy API ของคุณสู่ Production
• การจัดการเวอร์ชัน API (API Versioning)
• คำถามที่พบบ่อย (FAQ)
• สรุปและก้าวต่อไป

ทำความเข้าใจ REST API และ FastAPI

REST API คืออะไร? หลักการสำคัญที่คุณควรรู้

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

หลักการสำคัญของ REST API ได้แก่:

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

การสื่อสารใน REST มักจะใช้ HTTP Methods (หรือ Verbs) เพื่อระบุการดำเนินการที่เราต้องการทำกับทรัพยากร:

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

และใช้ HTTP Status Codes ในการระบุผลลัพธ์ของการดำเนินการ (เช่น 200 OK, 201 Created, 404 Not Found, 500 Internal Server Error) ครับ

ทำไมต้อง FastAPI? เจาะลึกจุดเด่นที่ทำให้เป็นตัวเลือกที่เหนือกว่า

FastAPI คือเว็บเฟรมเวิร์ก Python ที่ทันสมัย รวดเร็ว (ประสิทธิภาพสูง) และใช้งานง่ายสำหรับการสร้าง RESTful APIs มันสร้างขึ้นบนพื้นฐานของ Starlette สำหรับส่วนของเว็บ และ Pydantic สำหรับการจัดการข้อมูลและการตรวจสอบครับ จุดเด่นที่ทำให้ FastAPI เป็นตัวเลือกที่น่าสนใจมีดังนี้:

• ความเร็วและประสิทธิภาพสูง: FastAPI สร้างขึ้นบน ASGI (Asynchronous Server Gateway Interface) ทำให้สามารถจัดการคำขอพร้อมกันจำนวนมากได้ดีมาก โดยเฉพาะเมื่อใช้ร่วมกับ Uvicorn ซึ่งเป็น ASGI Server ที่เร็วที่สุดตัวหนึ่ง ด้วยการรองรับ async/await ของ Python ทำให้ FastAPI มีประสิทธิภาพเทียบเท่ากับ Node.js และ Go ในบางกรณีครับ
• ความเร็วในการพัฒนา: ด้วยการใช้ Python type hints ร่วมกับ Pydantic ทำให้ FastAPI สามารถตรวจสอบข้อมูลขาเข้า (Request Body, Query Parameters, Path Parameters) ได้โดยอัตโนมัติ และสร้างเอกสาร API (Swagger UI, ReDoc) ให้คุณทันที ลดเวลาในการเขียนโค้ดซ้ำซ้อนและลดข้อผิดพลาดลงได้อย่างมาก
• การตรวจสอบข้อมูล (Data Validation) อัตโนมัติ: Pydantic ทำงานเบื้องหลังเพื่อตรวจสอบโครงสร้างข้อมูล ประเภทข้อมูล และค่าต่างๆ ให้เป็นไปตามที่คุณกำหนดไว้ หากข้อมูลไม่ถูกต้อง ระบบจะส่งข้อผิดพลาดที่เข้าใจง่ายกลับไปให้ไคลเอนต์ทันทีครับ
• เอกสาร API อัตโนมัติ: FastAPI สร้างเอกสาร API ที่เป็นมาตรฐาน OpenAPI (Swagger UI) และ ReDoc ให้คุณโดยอัตโนมัติจากโค้ดของคุณทันทีที่คุณสร้าง API ขึ้นมา ทำให้การทดสอบและทำความเข้าใจ API ทำได้ง่ายขึ้นมากสำหรับทั้งนักพัฒนา Back-end และ Front-end
• รองรับ Asynchronous (async/await): FastAPI ได้รับการออกแบบมาให้รองรับ Asynchronous programming ได้อย่างเต็มที่ ทำให้แอปพลิเคชันของคุณสามารถทำงานแบบ Non-blocking ได้ เหมาะสำหรับ I/O-bound tasks เช่น การเรียกฐานข้อมูล หรือการเรียก External API ครับ
• Type Hints: FastAPI ใช้ Python type hints ในการประกาศประเภทข้อมูล ซึ่งช่วยให้โค้ดมีความชัดเจน อ่านง่าย และสามารถใช้ประโยชน์จาก IDE ในการตรวจสอบข้อผิดพลาด (Linting) และการเติมโค้ดอัตโนมัติ (Autocompletion) ได้ดียิ่งขึ้น
• Dependencies Injection System: ระบบการพึ่งพาที่ทรงพลังของ FastAPI ช่วยให้คุณสามารถจัดการการเชื่อมต่อฐานข้อมูล การตรวจสอบสิทธิ์ หรือการดึงค่าจากส่วนอื่น ๆ ได้อย่างสะอาดและเป็นระเบียบ ทำให้โค้ดสามารถนำกลับมาใช้ใหม่ได้ง่ายและทดสอบได้สะดวกครับ

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

อ่านเพิ่มเติมเกี่ยวกับหลักการออกแบบ API ที่ดี

FastAPI กับ Web Frameworks อื่นๆ: ตารางเปรียบเทียบ

เพื่อให้เห็นภาพชัดเจนว่า FastAPI โดดเด่นกว่า Frameworks อื่นๆ อย่างไร เรามาดูตารางเปรียบเทียบกับ Frameworks Python ยอดนิยมอื่นๆ เช่น Flask และ Django กันครับ

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

เตรียมความพร้อม: ติดตั้งและตั้งค่าโปรเจกต์ FastAPI

ข้อกำหนดเบื้องต้นและ Virtual Environment

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

สิ่งที่สำคัญอีกอย่างคือการใช้งาน Virtual Environment ซึ่งเป็นวิธีที่ดีที่สุดในการจัดการ dependency ของโปรเจกต์ Python มันช่วยแยกแพ็กเกจที่คุณติดตั้งสำหรับโปรเจกต์หนึ่ง ออกจากแพ็กเกจของโปรเจกต์อื่นๆ และจาก Global Python Environment ของคุณ ทำให้เกิดความสะอาดและไม่ชนกันครับ

สร้าง Virtual Environment:

python3 -m venv venv_fastapi

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

• สำหรับ macOS / Linux:

  
  source venv_fastapi/bin/activate
          

• สำหรับ Windows (Command Prompt):

  
  venv_fastapi\Scripts\activate.bat
          

• สำหรับ Windows (PowerShell):

  
  venv_fastapi\Scripts\Activate.ps1
          

เมื่อเปิดใช้งานแล้ว คุณจะเห็นชื่อ (venv_fastapi) นำหน้า prompt ของคุณ ซึ่งหมายความว่าคุณกำลังอยู่ใน Virtual Environment ครับ

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

เมื่อ Virtual Environment ของคุณพร้อมแล้ว ก็ถึงเวลาติดตั้ง FastAPI และ Uvicorn ครับ

• FastAPI: ตัวเฟรมเวิร์กหลัก
• Uvicorn: ASGI server ที่ใช้รันแอปพลิเคชัน FastAPI ของเรา (FastAPI ไม่ได้มี server ในตัว)

ติดตั้งทั้งสองอย่างด้วย pip:

pip install fastapi uvicorn

หากคุณต้องการฟังก์ชันเสริมของ Uvicorn เช่น การรองรับ HTTP/2 หรือการติดตั้ง Gunicorn (สำหรับ Production) คุณสามารถติดตั้งแบบ standard ได้ครับ:

pip install "uvicorn[standard]"

ตอนนี้คุณก็พร้อมที่จะเริ่มเขียนโค้ดแล้วครับ!

โครงสร้างโปรเจกต์เบื้องต้น

สำหรับโปรเจกต์ขนาดเล็ก คุณสามารถเก็บโค้ดทั้งหมดไว้ในไฟล์เดียวได้ แต่สำหรับโปรเจกต์ที่ซับซ้อนขึ้น การจัดโครงสร้างไฟล์ที่ดีเป็นสิ่งสำคัญครับ นี่คือโครงสร้างพื้นฐานที่เราจะใช้สำหรับบทความนี้:

.
├── main.py             # ไฟล์หลักของ FastAPI application
├── requirements.txt    # รายชื่อ packages ที่โปรเจกต์นี้ใช้
└── venv_fastapi/       # Virtual environment ของเรา

เมื่อโปรเจกต์ใหญ่ขึ้น คุณอาจจะแยกไฟล์เป็น models/, routers/, schemas/, database/ เป็นต้น เพื่อความเป็นระเบียบเรียบร้อยครับ

อย่าลืมสร้างไฟล์ requirements.txt เพื่อบันทึก dependency ของโปรเจกต์ของคุณ เพื่อให้ผู้อื่นสามารถติดตั้งได้อย่างง่ายดาย:

pip freeze > requirements.txt

เนื้อหาใน requirements.txt อาจมีลักษณะดังนี้ (เวอร์ชันอาจแตกต่างกันไป):

fastapi==0.103.2
pydantic==2.5.2
pydantic_core==2.14.5
starlette==0.27.0
uvicorn==0.24.0.post1

สร้าง REST API แรกของคุณ: Hello World และ Path Operations

Hello World API: เริ่มต้นง่ายๆ

มาเริ่มต้นด้วย API ที่ง่ายที่สุด นั่นคือการส่งข้อความ “Hello World” กลับไปครับ สร้างไฟล์ main.py ในโฟลเดอร์โปรเจกต์ของคุณ แล้วเพิ่มโค้ดต่อไปนี้:

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    """
    Endpoint สำหรับหน้าหลัก ส่งข้อความ "Hello World" กลับไป
    """
    return {"message": "Hello World, from FastAPI!"}

คำอธิบายโค้ด:

• from fastapi import FastAPI: นำเข้าคลาส FastAPI ที่เป็นหัวใจของแอปพลิเคชันของเรา
• app = FastAPI(): สร้างอินสแตนซ์ของแอปพลิเคชัน FastAPI
• @app.get("/"): นี่คือ decorator ที่บอก FastAPI ว่าฟังก์ชัน read_root ด้านล่างควรถูกเรียกใช้เมื่อมีคำขอ HTTP GET ไปยังเส้นทาง / (root path)
• async def read_root():: ฟังก์ชันนี้ถูกกำหนดให้เป็น async ซึ่งหมายความว่ามันสามารถทำงานแบบ Asynchronous ได้ การใช้ async เป็นแนวทางปฏิบัติที่ดีใน FastAPI แม้ว่าฟังก์ชันนี้จะไม่ได้ทำอะไรที่ต้องรอ I/O ก็ตาม
• return {"message": "Hello World, from FastAPI!"}: FastAPI จะแปลง Python dictionary นี้ให้เป็น JSON response โดยอัตโนมัติ

วิธีการรันแอปพลิเคชัน:

เปิด Terminal (ที่เปิดใช้งาน Virtual Environment อยู่) แล้วรันคำสั่ง:

uvicorn main:app --reload

คำอธิบายคำสั่ง:

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

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

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] using WatchFiles
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

เปิดเว็บเบราว์เซอร์ของคุณแล้วไปที่ http://127.0.0.1:8000 คุณควรจะเห็น:

{"message": "Hello World, from FastAPI!"}

เยี่ยมมากครับ! คุณได้สร้าง REST API ตัวแรกของคุณสำเร็จแล้ว

Path Parameters: การรับค่าผ่าน URL

Path Parameters ช่วยให้เราสามารถส่งข้อมูลเป็นส่วนหนึ่งของ URL ได้ครับ เช่น /items/1 หรือ /users/john_doe

เพิ่มโค้ดต่อไปนี้ใน main.py:

# main.py (ต่อจากโค้ดเดิม)
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
    """
    Endpoint สำหรับดึงข้อมูล item ด้วย ID
    รับ item_id เป็น Path Parameter และ q เป็น Optional Query Parameter
    """
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    return item

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    """
    Endpoint สำหรับดึงข้อมูล user ด้วย ID
    """
    return {"user_id": user_id, "message": f"Hello User {user_id}"}

คำอธิบาย:

• /items/{item_id}: ส่วน {item_id} ใน path คือ Path Parameter ครับ
• item_id: int: FastAPI จะตรวจจับว่า item_id เป็น Path Parameter และใช้ type hint int เพื่อตรวจสอบว่าค่าที่รับเข้ามาเป็นตัวเลขจำนวนเต็ม ถ้าไม่เป็น มันจะส่งข้อผิดพลาด 422 (Unprocessable Entity) กลับไปโดยอัตโนมัติ
• q: str | None = None: นี่คือ Optional Query Parameter ครับ หมายความว่า q อาจมีค่าเป็น string หรือ None ก็ได้ และมีค่าเริ่มต้นเป็น None

ทดสอบโดยไปที่:

• http://127.0.0.1:8000/items/5
• http://127.0.0.1:8000/items/10?q=somequery
• http://127.0.0.1:8000/users/alice

จะเห็นว่า FastAPI จัดการการแปลงประเภทข้อมูลและให้ค่าเริ่มต้นได้อย่างชาญฉลาดครับ

Query Parameters: การกรองและจำกัดข้อมูล

Query Parameters มักใช้สำหรับการกรอง การจัดเรียง หรือการแบ่งหน้า (pagination) ข้อมูลครับ พวกมันจะปรากฏหลังเครื่องหมาย ? ใน URL เช่น /items?skip=0&limit=10

ในตัวอย่าง read_item ด้านบน q: str | None = None ก็เป็น Query Parameter ครับ

มาดูตัวอย่างเพิ่มเติมสำหรับการแบ่งหน้า:

# main.py (ต่อจากโค้ดเดิม)
from typing import List

# สมมติฐานข้อมูล
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    """
    Endpoint สำหรับดึงรายการ item ทั้งหมด
    รองรับ Query Parameters สำหรับ Pagination (skip, limit)
    """
    return fake_items_db[skip : skip + limit]

คำอธิบาย:

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

ทดสอบโดยไปที่:

• http://127.0.0.1:8000/items/ (จะเห็น 3 รายการแรก)
• http://127.0.0.1:8000/items/?skip=1&limit=2 (จะเห็นรายการที่ 2 และ 3)

Request Body และ Pydantic: การส่งข้อมูลและการตรวจสอบ

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

ก่อนอื่น มาสร้าง Pydantic Model กันครับ:

# main.py (เพิ่มส่วนนี้ที่ด้านบนของไฟล์, ใต้ import FastAPI)
from typing import Optional
from pydantic import BaseModel, Field

class Item(BaseModel):
    """
    Pydantic Model สำหรับ Item
    ใช้ในการกำหนดโครงสร้างข้อมูลและตรวจสอบความถูกต้อง
    """
    name: str = Field(..., example="Latest Smartphone")
    description: Optional[str] = Field(None, example="A brand new smartphone with advanced features.")
    price: float = Field(..., example=799.99, gt=0, description="Price must be greater than zero.")
    tax: Optional[float] = Field(None, example=0.10)
    tags: List[str] = Field(default_factory=list, example=["electronics", "mobile"])

class User(BaseModel):
    """
    Pydantic Model สำหรับ User
    """
    username: str = Field(..., example="john_doe")
    email: str = Field(..., example="[email protected]")
    full_name: Optional[str] = Field(None, example="John Doe")
    disabled: Optional[bool] = Field(False)

คำอธิบาย Pydantic Model:

• class Item(BaseModel):: เราสร้างคลาส Item ที่สืบทอดมาจาก pydantic.BaseModel
• name: str: กำหนดว่า name ต้องเป็น string และเป็น Field ที่บังคับ (required)
• description: Optional[str] = None: กำหนดว่า description เป็น string ก็ได้ หรือเป็น None ก็ได้ (optional)
• price: float: กำหนดว่า price ต้องเป็น float
• Field(...): เป็นฟังก์ชันจาก Pydantic ที่ช่วยให้เราสามารถกำหนด metadata เพิ่มเติมสำหรับ Field ได้
  • ... (Ellipsis): ใช้เพื่อระบุว่า Field นั้น จำเป็น (required)
  • example="...": ใช้สำหรับสร้างตัวอย่างในเอกสาร OpenAPI
  • gt=0: กำหนดเงื่อนไขว่า price ต้องมากกว่า 0 (Greater Than)
  • description="...": เพิ่มคำอธิบายสำหรับ Field นั้นๆ
  • default_factory=list: ใช้สำหรับตั้งค่าเริ่มต้นเป็น list ว่างเปล่า เพื่อหลีกเลี่ยงปัญหา mutable default arguments

เมื่อมี Model แล้ว เราสามารถใช้มันใน Path Operation ได้เลยครับ:

# main.py (ต่อจากโค้ดเดิม)
@app.post("/items/")
async def create_item(item: Item):
    """
    Endpoint สำหรับสร้าง item ใหม่
    รับ Request Body เป็น Item Pydantic Model
    """
    item_dict = item.model_dump() # ใช้ .model_dump() หรือ .dict() ใน Pydantic v1
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
    """
    Endpoint สำหรับอัปเดต item ด้วย ID
    รับ item_id เป็น Path Parameter และ Request Body เป็น Item Pydantic Model
    """
    results = {"item_id": item_id, **item.model_dump()}
    if q:
        results.update({"q": q})
    return results

คำอธิบาย:

• item: Item: FastAPI ตรวจพบว่า item เป็นพารามิเตอร์ของประเภท Pydantic Model และจะพยายามอ่าน Request Body เป็น JSON แล้วแปลงเป็นอินสแตนซ์ของคลาส Item ให้โดยอัตโนมัติ พร้อมทั้งตรวจสอบความถูกต้องของข้อมูลตามที่เรากำหนดไว้ใน Model ด้วยครับ
• หากข้อมูลใน Request Body ไม่ตรงตาม Model (เช่น ขาด Field ที่จำเป็น หรือประเภทข้อมูลไม่ถูกต้อง) FastAPI จะส่ง HTTP 422 (Unprocessable Entity) กลับไปโดยอัตโนมัติพร้อมรายละเอียดข้อผิดพลาด

ทดสอบโดยเข้า http://127.0.0.1:8000/docs (Swagger UI) คุณจะเห็นเอกสาร API ที่สร้างขึ้นโดยอัตโนมัติ และสามารถทดสอบ API ได้จากหน้านั้นเลยครับ ลองส่ง JSON body ที่ถูกต้องและไม่ถูกต้องดูนะครับ

ตัวอย่าง Request Body สำหรับ POST /items/:

{
  "name": "My new item",
  "description": "This is a very cool item.",
  "price": 12.5,
  "tax": 1.2
}

อ่านเพิ่มเติมเกี่ยวกับ Pydantic และการตรวจสอบข้อมูล

การจัดการ CRUD Operations (GET, POST, PUT, DELETE, PATCH)

เราได้เห็น GET, POST, PUT ไปแล้ว มาดู DELETE และ PATCH เพื่อให้ครบวงจรของ CRUD (Create, Read, Update, Delete) กันครับ

# main.py (ต่อจากโค้ดเดิม)
from typing import Dict, Any

# สมมติฐานข้อมูลในหน่วยความจำ
# ในโปรเจกต์จริงจะเชื่อมต่อกับ Database
fake_db: Dict[int, Item] = {}
next_item_id = 1

@app.post("/items/", response_model=Item, status_code=201)
async def create_item_in_db(item: Item):
    """
    สร้าง item ใหม่ในฐานข้อมูลจำลอง
    """
    global next_item_id
    item_id = next_item_id
    fake_db[item_id] = item
    next_item_id += 1
    return item # FastAPI จะ serialise Item object กลับเป็น JSON

@app.get("/items/{item_id}", response_model=Item)
async def read_item_from_db(item_id: int):
    """
    ดึงข้อมูล item จากฐานข้อมูลจำลองด้วย ID
    """
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return fake_db[item_id]

@app.put("/items/{item_id}", response_model=Item)
async def update_item_in_db(item_id: int, item: Item):
    """
    อัปเดตข้อมูล item ในฐานข้อมูลจำลองด้วย ID (อัปเดตทั้งหมด)
    """
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    fake_db[item_id] = item
    return item

@app.delete("/items/{item_id}", status_code=204)
async def delete_item_from_db(item_id: int):
    """
    ลบ item จากฐานข้อมูลจำลองด้วย ID
    """
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")
    del fake_db[item_id]
    return {"message": "Item deleted successfully"}

# สำหรับ PATCH เราอาจจะสร้าง Pydantic Model แยกต่างหากสำหรับ Partial Update
class ItemUpdate(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: Optional[float] = None
    tags: Optional[List[str]] = None

@app.patch("/items/{item_id}", response_model=Item)
async def partial_update_item_in_db(item_id: int, item_update: ItemUpdate):
    """
    อัปเดตข้อมูล item บางส่วนในฐานข้อมูลจำลองด้วย ID
    """
    if item_id not in fake_db:
        raise HTTPException(status_code=404, detail="Item not found")

    existing_item = fake_db[item_id]
    update_data = item_update.model_dump(exclude_unset=True) # ไม่รวม fields ที่ไม่ได้ถูกตั้งค่า
    updated_item_data = existing_item.model_dump()
    updated_item_data.update(update_data) # อัปเดตเฉพาะ fields ที่ส่งมา
    
    updated_item = Item(**updated_item_data)
    fake_db[item_id] = updated_item
    return updated_item

คำอธิบาย:

• response_model=Item: นี่คือพารามิเตอร์ใน decorator ที่บอก FastAPI ว่า response ที่ส่งกลับไปควรมีโครงสร้างตาม Pydantic Model Item FastAPI จะใช้ Model นี้ในการสร้างเอกสาร API และ serialise object ที่ return กลับมาให้เป็น JSON ที่ถูกต้อง
• status_code=201: สำหรับ POST เรามักจะส่ง status code 201 Created กลับไปเมื่อสร้างทรัพยากรสำเร็จ
• status_code=204: สำหรับ DELETE ที่สำเร็จ มักจะส่ง 204 No Content ซึ่งหมายถึงไม่มี content ใน response body (แต่เราก็ส่งข้อความกลับไปได้ ซึ่ง FastAPI จะจัดการให้)
• HTTPException: ใช้สำหรับส่ง HTTP Error พร้อม status code และรายละเอียดข้อผิดพลาด
• สำหรับ PATCH เราสร้าง ItemUpdate Model ที่ทุก Field เป็น Optional เพื่อให้ผู้ใช้สามารถส่งมาเฉพาะ Field ที่ต้องการอัปเดตได้
• item_update.model_dump(exclude_unset=True): Pydantic function ที่จะสร้าง dictionary จาก Model โดยไม่รวม Field ที่ไม่ได้ถูกตั้งค่า ซึ่งเหมาะสำหรับ partial updates ครับ

ตอนนี้คุณมีโครงสร้างพื้นฐานของ REST API ที่สมบูรณ์แบบพร้อม CRUD operations แล้วครับ!

คุณสมบัติขั้นสูงของ FastAPI ที่จะยกระดับ API ของคุณ

Dependencies: การจัดการการพึ่งพาและการฉีดค่า

ระบบ Dependencies Injection ของ FastAPI เป็นคุณสมบัติที่ทรงพลังมาก ช่วยให้คุณสามารถ “ฉีด” ฟังก์ชันหรือคลาสเข้าไปใน Path Operation functions ได้โดยอัตโนมัติ มีประโยชน์อย่างมากสำหรับการจัดการสิ่งต่างๆ เช่น:

• การตรวจสอบสิทธิ์ (Authentication)
• การเชื่อมต่อฐานข้อมูล
• การดึงค่าจาก Header หรือ Cookie
• การใช้ Logic ที่ซ้ำกันในหลายๆ Path Operation

มาดูตัวอย่างการใช้ Dependencies เพื่อดึงค่า API Key จาก Header ครับ:

# main.py (เพิ่มส่วนนี้)
from fastapi import Header, HTTPException, Depends

async def get_api_key(api_key: str = Header(..., description="API Key for authentication")):
    """
    Dependency เพื่อตรวจสอบ API Key จาก X-API-Key header
    """
    if api_key != "SECRET_API_KEY":
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return api_key

@app.get("/secure-data/", dependencies=[Depends(get_api_key)])
async def read_secure_data():
    """
    Endpoint ที่ต้องใช้ API Key ในการเข้าถึง
    """
    return {"message": "You have access to secure data!"}

@app.get("/items/special/", response_model=Item)
async def read_special_item(api_key: str = Depends(get_api_key)):
    """
    Endpoint สำหรับดึงข้อมูล item พิเศษที่ต้องใช้ API Key
    และแสดง API Key ที่ใช้ในการเข้าถึง (สำหรับการทดสอบ)
    """
    # ในสถานการณ์จริง เราอาจจะใช้ API Key เพื่อดึงข้อมูลเฉพาะผู้ใช้
    return Item(name="Special Item", price=999.99, description=f"Accessed with API Key: {api_key}")

คำอธิบาย:

• async def get_api_key(...): นี่คือฟังก์ชัน dependency ของเรา มันจะรับ api_key จาก HTTP Header ชื่อ X-API-Key (ตั้งชื่อพารามิเตอร์ให้ตรงกับชื่อ Header หรือใช้ Header(alias="X-API-Key"))
• raise HTTPException(status_code=401, detail="Invalid API Key"): หาก API Key ไม่ถูกต้อง จะส่งข้อผิดพลาด 401 Unauthorized กลับไป
• @app.get("/secure-data/", dependencies=[Depends(get_api_key)]): เราสามารถใช้ dependencies=[Depends(dependency_function)] เพื่อเพิ่ม dependency ให้กับ Path Operation ได้
• api_key: str = Depends(get_api_key): หรือสามารถฉีดค่าที่ return จาก dependency เข้ามาใน Path Operation function ได้โดยตรงครับ

ลองทดสอบโดยเข้า http://127.0.0.1:8000/docs แล้วลองส่งคำขอไปที่ /secure-data/ โดยใส่และไม่ใส่ X-API-Key ที่ถูกต้องดูนะครับ

Security และ Authentication: การรักษาความปลอดภัย API ของคุณ

FastAPI มีเครื่องมือสำหรับจัดการความปลอดภัยหลายรูปแบบ โดยใช้มาตรฐาน OpenAPI Security Schemes ครับ ที่นิยมใช้กันคือ OAuth2 (ซึ่งรวมถึง JWT tokens) และ API Key

มาดูตัวอย่างการใช้ OAuth2 with Password Flow (ซึ่งเป็นพื้นฐานสำหรับการใช้งาน JWT) ครับ

# main.py (เพิ่มส่วนนี้)
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from datetime import datetime, timedelta
from passlib.context import CryptContext

# การตั้งค่าสำหรับ JWT
SECRET_KEY = "YOUR_SUPER_SECRET_KEY" # ควรเก็บใน Environment Variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

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

# ฟังก์ชันสำหรับ Hash Password
def hash_password(password: str):
    return pwd_context.hash(password)

# ฟังก์ชันสำหรับตรวจสอบ Password
def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

# ฟังก์ชันสำหรับสร้าง Access Token
def create_access_token(data: dict, expires_delta: timedelta | None = 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

# ฟังก์ชันสำหรับตรวจสอบ Token และดึง User
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=401,
        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 = {"username": username, "full_name": "Test User", "email": "[email protected]"}
        if user is None:
            raise credentials_exception
        return user
    except JWTError:
        raise credentials_exception

# Endpoint สำหรับ Login (รับ username, password)
@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # ในสถานการณ์จริง คุณจะตรวจสอบ username/password กับฐานข้อมูล
    if form_data.username == "testuser" and form_data.password == "testpass":
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"sub": form_data.username}, expires_delta=access_token_expires
        )
        return {"access_token": access_token, "token_type": "bearer"}
    raise HTTPException(
        status_code=400, detail="Incorrect username or password"
    )

# Endpoint ที่ต้องใช้ Token ในการเข้าถึง
@app.get("/users/me/")
async def read_users_me(current_user: User = Depends(get_current_user)):
    """
    ดึงข้อมูลผู้ใช้ปัจจุบันที่ Login อยู่
    """
    return current_user

คำอธิบาย:

• pip install python-jose[cryptography] passlib[bcrypt]: คุณต้องติดตั้งไลบรารีเหล่านี้ก่อนครับ
• OAuth2PasswordBearer(tokenUrl="token"): บอก FastAPI ว่าจะใช้ OAuth2 และ endpoint สำหรับขอ token คือ /token
• create_access_token: ฟังก์ชันสำหรับสร้าง JWT token ที่มีข้อมูลผู้ใช้ (sub) และเวลาหมดอายุ (exp)
• get_current_user: Dependency ที่จะดึง token จาก Header, ตรวจสอบความถูกต้อง, และ decode token เพื่อดึงข้อมูลผู้ใช้ หาก token ไม่ถูกต้อง จะส่ง HTTP 401
• @app.post("/token"): Endpoint สำหรับ Login ที่รับ username และ password ผ่าน OAuth2PasswordRequestForm แล้วสร้าง Access Token กลับไป
• @app.get("/users/me/"): Endpoint ที่ Protected ซึ่งจะใช้ get_current_user เพื่อตรวจสอบและฉีดข้อมูลผู้ใช้ปัจจุบันเข้ามาในฟังก์ชัน

ลองทดสอบโดย:

1. ไปที่ http://127.0.0.1:8000/docs
2. ไปที่ /token endpoint, คลิก “Try it out”
3. ใส่ username: testuser และ password: testpass แล้วกด “Execute”
4. คุณจะได้รับ access_token คัดลอกค่านี้ไว้
5. คลิกปุ่ม “Authorize” (รูปกุญแจ) ที่มุมขวาบนของหน้าเว็บ
6. เลือก “oauth2_scheme”, วาง access_token ที่ได้มาลงในช่อง “Value” โดยมีคำว่า “Bearer ” นำหน้า (เช่น Bearer eyJhbGciOiJIUzI1Ni...) แล้วกด “Authorize”
7. ลองเรียก /users/me/ อีกครั้ง คุณจะเห็นข้อมูลผู้ใช้ครับ

นี่เป็นเพียงพื้นฐานของการทำ Authentication ด้วย FastAPI ครับ คุณสามารถขยายไปใช้ Database ในการจัดเก็บผู้ใช้และ Password ได้ครับ

Database Integration: เชื่อมต่อและจัดการข้อมูล (SQLAlchemy)

FastAPI ไม่ได้มี ORM (Object-Relational Mapper) ในตัวเหมือน Django แต่สามารถทำงานร่วมกับ ORM ยอดนิยมอื่นๆ ได้ดีเยี่ยม เช่น SQLAlchemy, SQLModel (ซึ่งสร้างบน SQLAlchemy และ Pydantic) หรือ Tortoise-ORM ครับ ในที่นี้เราจะใช้ SQLAlchemy (แบบ Asynchronous) เพื่อเชื่อมต่อกับ SQLite ครับ

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

pip install sqlalchemy aiosqlite # aiosqlite สำหรับ async SQLite

สร้างไฟล์ database.py และ models.py

database.py:

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# ตั้งค่า URL สำหรับฐานข้อมูล SQLite
# asyncpg สำหรับ PostgreSQL, aiomysql สำหรับ MySQL
DATABASE_URL = "sqlite+aiosqlite:///./sql_app.db"

# สร้าง Async Engine
engine = create_async_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}, echo=True
)

# สร้าง Async Session Local
AsyncSessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=engine, class_=AsyncSession
)

Base = declarative_base()

# Dependency สำหรับการรับ Database Session
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

models.py:

# models.py
from sqlalchemy import Column, Integer, String, Boolean, Float
from .database import Base

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)
    is_offer = Column(Boolean, default=False)

แก้ไข main.py เพื่อรวม Database และ Pydantic Models สำหรับ Schema:

# main.py (เพิ่ม imports)
from typing import List
from sqlalchemy.future import select
from sqlalchemy.exc import NoResultFound

from .database import Base, engine, get_db
from .models import DBItem

# Pydantic Schemas (สำหรับ Request/Response)
class ItemBase(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    is_offer: bool = False

    class ConfigDict: # สำหรับ Pydantic v2
        from_attributes = True # alias: orm_mode = True ใน Pydantic v1

# สร้างตารางในฐานข้อมูล
@app.on_event("startup")
async def startup_event():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

# CRUD Operations กับ Database
@app.post("/db_items/", response_model=Item, status_code=201)
async def create_db_item(item: ItemCreate, db: AsyncSession = Depends(get_db)):
    db_item = DBItem(**item.model_dump())
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

@app.get("/db_items/", response_model=List[Item])
async def read_db_items(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(DBItem).offset(skip).limit(limit))
    items = result.scalars().all()
    return items

@app.get("/db_items/{item_id}", response_model=Item)
async def read_db_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(DBItem).where(DBItem.id == item_id))
    try:
        item = result.scalar_one()
    except NoResultFound:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.put("/db_items/{item_id}", response_model=Item)
async def update_db_item(item_id: int, item: ItemCreate, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(DBItem).where(DBItem.id == item_id))
    try:
        db_item = result.scalar_one()
    except NoResultFound:
        raise HTTPException(status_code=404, detail="Item not found")

    item_data = item.model_dump(exclude_unset=True)
    for key, value in item_data.items():
        setattr(db_item, key, value)
    
    await db.commit()
    await db.refresh(db_item)
    return db_item

@app.delete("/db_items/{item_id}", status_code=204)
async def delete_db_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(DBItem).where(DBItem.id == item_id))
    try:
        db_item = result.scalar_one()
    except NoResultFound:
        raise HTTPException(status_code=404, detail="Item not found")
    
    await db.delete(db_item)
    await db.commit()
    return {"message": "Item deleted successfully"}

คำอธิบาย:

• DBItem: SQLAlchemy Model ที่แมปกับตาราง items ในฐานข้อมูล
• ItemBase, ItemCreate, Item: Pydantic Schemas ที่ใช้ในการกำหนดโครงสร้างข้อมูลสำหรับ Request และ Response โดย Item มี id และ is_offer เพิ่มเติม และมี ConfigDict(from_attributes=True) เพื่อให้ Pydantic สามารถอ่านข้อมูลจาก SQLAlchemy Model ได้โดยตรง
• @app.on_event("startup"): Event ที่จะถูกเรียกเมื่อแอปพลิเคชันเริ่มต้น ใช้สำหรับสร้างตารางในฐานข้อมูลหากยังไม่มี
• db: AsyncSession = Depends(get_db): ใช้ Dependency Injection เพื่อรับ database session สำหรับแต่ละ request
• การดำเนินการกับ db (db.add, db.commit, db.refresh, db.execute, db.delete) ทั้งหมดใช้ await เนื่องจากเรากำลังใช้ SQLAlchemy ในโหมด Asynchronous ครับ
• result.scalars().all() และ result.scalar_one(): วิธีการดึงข้อมูลจาก SQLAlchemy query

ตอนนี้คุณมี API ที่สามารถเชื่อมต่อและจัดการข้อมูลในฐานข้อมูลได้แล้วครับ!

Error Handling: การจัดการข้อผิดพลาดอย่างมืออาชีพ

FastAPI มีระบบการจัดการข้อผิดพลาดในตัวที่ยอดเยี่ยม โดยจะส่ง HTTP 422 สำหรับ validation errors และคุณสามารถใช้ HTTPException เพื่อส่งข้อผิดพลาดอื่นๆ ได้

นอกจากนี้ คุณยังสามารถสร้าง Custom Exception Handlers เพื่อจัดการกับข้อผิดพลาดบางประเภทในลักษณะที่คุณต้องการได้ครับ

# main.py (เพิ่มส่วนนี้)
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

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

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something wrong. But I'm a teapot!"},
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    # นี่คือ Default Exception Handler ของ Starlette
    # คุณสามารถปรับแต่ง Response ได้ตามต้องการ
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "custom_message": "An error occurred, please try again."}
    )

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

คำอธิบาย:

• UnicornException: Custom Exception ที่เราสร้างขึ้นเอง
• @app.exception_handler(UnicornException): Decorator ที่ลงทะเบียนฟังก์ชัน unicorn_exception_handler ให้เป็นตัวจัดการเมื่อ UnicornException ถูกเรียกใช้งาน
• JSONResponse: ใช้สำหรับสร้าง JSON response ที่มี status code และ content ที่กำหนดเอง
• เรายังสามารถ override default exception handler ของ HTTPException ได้ด้วย @app.exception_handler(StarletteHTTPException) เพื่อปรับแต่ง response ของ HTTP errors ทั่วไปได้ครับ

ทดสอบโดยไปที่ http://127.0.0.1:8000/unicorns/yolo คุณจะเห็น response ที่เรากำหนดเองครับ

Background Tasks: การทำงานเบื้องหลังที่ไม่บล็อก API

บางครั้งเราต้องการให้ API ทำงานบางอย่างที่ใช้เวลานาน (เช่น ส่งอีเมล, ประมวลผลภาพ) แต่ไม่ต้องการให้ผู้ใช้ต้องรอนาน FastAPI มี BackgroundTasks สำหรับจัดการเรื่องนี้ครับ

# main.py (เพิ่ม imports)
from fastapi import BackgroundTasks

def write_notification(email: str, message=""):
    with open("log.txt", mode="a") as email_file:
        content = f"notification for {email}: {message}\n"
        email_file.write(content)
    print(f"Notification written for {email}")

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks, message: str = "Hello from FastAPI!"):
    """
    ส่ง Notification โดยใช้ Background Task
    """
    background_tasks.add_task(write_notification, email, message)
    return {"message": "Notification sent in the background"}

คำอธิบาย:

• background_tasks: BackgroundTasks: FastAPI จะฉีดอินสแตนซ์ของ BackgroundTasks เข้ามาในฟังก์ชัน Path Operation
• background_tasks.add_task(write_notification, email, message): เพิ่มฟังก์ชัน write_notification เข้าไปในคิวของงานเบื้องหลัง โดยส่ง email และ message เป็นอาร์กิวเมนต์

เมื่อคุณเรียก POST /send-notification/[email protected] API จะส่ง response กลับมาทันที แต่ฟังก์ชัน write_notification จะทำงานในเบื้องหลังโดยไม่บล็อกการตอบสนองของ API ครับ คุณสามารถตรวจสอบไฟล์ log.txt ที่สร้างขึ้นได้

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

Middleware คือฟังก์ชันที่สามารถทำงานก่อนและหลัง Path Operation ได้ มีประโยชน์สำหรับการจัดการข้ามตัด (cross-cutting concerns) เช่น การบันทึก Log, การจัดการ CORS (Cross-Origin Resource Sharing), การตรวจสอบสิทธิ์เบื้องต้น หรือการบีบอัดข้อมูลครับ

# main.py (เพิ่ม imports)

import time

from fastapi.middleware.cors import CORSMiddleware

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint

from starlette.responses import Response

from starlette.requests import Request

# เพิ่ม CORS Middleware

app.add_middleware(

    CORSMiddleware,

    allow_origins=["http://localhost:3000", "https://yourfrontend.com"], # หรือ ["*"] สำหรับทุกโดเมน

    allow_credentials=True,

    allow_methods=["*"],

    allow_headers=["*"],

)
# Custom Middleware สำหรับ Logging Request Time

class TimingMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:

        start_time = time.time()

        response = await call_next(request)

        process_time = time.time() - start_time
    	
บทความที่เกี่ยวข้อง
Python Automation สร้าง Bot อัตโนมัติสำหรับงาน IT
PostgreSQL Performance Tuning คู่มือปรับแต่งฐานข้อมูล
MikroTik Router คู่มือตั้งค่า MikroTik RouterOS สำหรับองค์กร
SiamCafe.net — ชุมชน IT ที่ใหญ่ที่สุด · Siam2R.com — Portfolio งาน IT