
ในโลกของการพัฒนาซอฟต์แวร์ยุคใหม่ การเชื่อมต่อระหว่างระบบต่างๆ เป็นหัวใจสำคัญที่ขับเคลื่อนนวัตกรรมและบริการดิจิทัลมากมายครับ และ REST API ก็คือกลไกหลักที่ทำให้การเชื่อมต่อเหล่านั้นเป็นไปได้อย่างราบรื่นและมีประสิทธิภาพ วันนี้ SiamLancard.com ขอพาคุณดำดิ่งสู่โลกของการสร้าง REST API ด้วย FastAPI ซึ่งเป็น Python web framework ที่กำลังมาแรงที่สุดในขณะนี้ ด้วยความเร็ว ประสิทธิภาพ และความง่ายในการพัฒนาที่โดดเด่น FastAPI ได้รับการยกย่องว่าเป็นหนึ่งในเครื่องมือที่ดีที่สุดสำหรับการสร้าง API ที่ทันสมัยและปรับขนาดได้ เราจะมาเรียนรู้กันตั้งแต่พื้นฐานไปจนถึงการสร้าง API ที่ใช้งานได้จริง พร้อมฟีเจอร์ขั้นสูงต่างๆ ที่จะช่วยให้คุณเป็นนักพัฒนา FastAPI มืออาชีพได้แบบครบจบในบทความเดียวเลยครับ
สารบัญ
- REST API คืออะไร และทำไมถึงสำคัญกับการพัฒนาเว็บสมัยใหม่?
- ทำไมต้องเลือก FastAPI? จุดเด่นที่เหนือกว่าเฟรมเวิร์กอื่น
- เตรียมความพร้อมก่อนเริ่มสร้าง API ด้วย FastAPI
- สร้าง API แรกด้วย FastAPI: Hello World!
- ทำความเข้าใจส่วนประกอบหลักของ FastAPI
- Path Operations: จัดการ HTTP Methods (GET, POST, PUT, DELETE)
- Path Parameters: ส่งข้อมูลผ่าน URL
- Query Parameters: ส่งข้อมูลเพิ่มเติมผ่าน URL
- Request Body: ส่งข้อมูลในรูปแบบ JSON ด้วย Pydantic
- Type Hints: หัวใจสำคัญของการตรวจสอบข้อมูลและเอกสารอัตโนมัติ
- การจัดการข้อผิดพลาด (Error Handling) ด้วย HTTPException
- สร้าง REST API ระบบจัดการสินค้า (CRUD) แบบครบวงจร
- โครงสร้างโปรเจกต์
- สร้าง Pydantic Models สำหรับข้อมูลสินค้า
- จำลองฐานข้อมูลในหน่วยความจำ
- สร้าง Endpoint สำหรับการเพิ่มสินค้า (Create – POST)
- สร้าง Endpoint สำหรับการดึงข้อมูลสินค้าทั้งหมด (Read All – GET)
- สร้าง Endpoint สำหรับการดึงข้อมูลสินค้าตาม ID (Read One – GET)
- สร้าง Endpoint สำหรับการอัปเดตข้อมูลสินค้า (Update – PUT)
- สร้าง Endpoint สำหรับการลบข้อมูลสินค้า (Delete – DELETE)
- ฟีเจอร์ขั้นสูงของ FastAPI ที่ควรรู้
- การทดสอบ API ด้วย FastAPI TestClient
- การ Deploy API ของคุณสู่ Production
- คำถามที่พบบ่อย (FAQ)
- สรุปและก้าวต่อไป
REST API คืออะไร และทำไมถึงสำคัญกับการพัฒนาเว็บสมัยใหม่?
REST API (Representational State Transfer Application Programming Interface) คือชุดของกฎและข้อกำหนดที่ใช้ในการสื่อสารระหว่างแอปพลิเคชันต่างๆ บนอินเทอร์เน็ตครับ ลองจินตนาการว่าคุณมีแอปพลิเคชันมือถือที่ต้องการดึงข้อมูลสินค้าจากเว็บไซต์ e-commerce หรือระบบหลังบ้าน (backend) ที่ต้องการส่งข้อมูลไปยังแอปพลิเคชันหน้าบ้าน (frontend) API นี่แหละครับคือ “สะพาน” ที่ทำให้ข้อมูลเหล่านี้เดินทางไปมาระหว่างกันได้
หัวใจสำคัญของ REST API คือการใช้ HTTP Methods (เช่น GET, POST, PUT, DELETE) เพื่อดำเนินการกับ “ทรัพยากร” (Resources) ซึ่งอาจเป็นข้อมูลสินค้า, ผู้ใช้งาน, หรือข้อมูลอื่นๆ ที่เราต้องการจัดการ โดยแต่ละ Method จะมีความหมายเฉพาะเจาะจง:
- GET: ใช้สำหรับขอข้อมูลจากเซิร์ฟเวอร์ (เช่น ดึงข้อมูลสินค้าทั้งหมด หรือสินค้าชิ้นเดียว)
- POST: ใช้สำหรับส่งข้อมูลใหม่ไปยังเซิร์ฟเวอร์เพื่อสร้างทรัพยากร (เช่น เพิ่มสินค้าใหม่)
- PUT: ใช้สำหรับอัปเดตข้อมูลของทรัพยากรที่มีอยู่บนเซิร์ฟเวอร์ (เช่น แก้ไขข้อมูลสินค้า)
- DELETE: ใช้สำหรับลบทรัพยากรออกจากเซิร์ฟเวอร์ (เช่น ลบสินค้า)
REST API มีความสำคัญอย่างยิ่งในการพัฒนาเว็บสมัยใหม่ด้วยเหตุผลหลายประการครับ:
- ความยืดหยุ่น: ช่วยให้ Frontend (เช่น เว็บไซต์, แอปมือถือ) สามารถแยกการพัฒนาออกจาก Backend ได้อย่างอิสระ
- การนำกลับมาใช้ใหม่ (Reusability): API เดียวกันสามารถถูกใช้งานโดยแอปพลิเคชันหลายตัวได้ เช่น API ดึงข้อมูลสินค้าอาจถูกใช้โดยเว็บไซต์, แอป iOS, และแอป Android
- การปรับขนาด (Scalability): ช่วยให้การขยายระบบทำได้ง่ายขึ้น เนื่องจากแต่ละส่วนทำงานแยกกัน
- มาตรฐานเปิด: ใช้โปรโตคอล HTTP และรูปแบบข้อมูลที่เป็นมาตรฐาน (ส่วนใหญ่เป็น JSON) ทำให้ง่ายต่อการทำความเข้าใจและใช้งานข้ามแพลตฟอร์ม
ดังนั้น การเข้าใจและสามารถสร้าง REST API ได้จึงเป็นทักษะที่จำเป็นอย่างยิ่งสำหรับนักพัฒนาทุกคนในยุคดิจิทัลครับ
ทำไมต้องเลือก FastAPI? จุดเด่นที่เหนือกว่าเฟรมเวิร์กอื่น
FastAPI คือ Python web framework ที่ถูกออกแบบมาเพื่อสร้าง API โดยเฉพาะ ด้วยประสิทธิภาพที่รวดเร็ว (Fast) และใช้งานง่าย (Easy) ที่มาพร้อมกับฟีเจอร์ที่ทันสมัย ทำให้มันกลายเป็นตัวเลือกยอดนิยมอย่างรวดเร็ว นี่คือเหตุผลหลักๆ ที่คุณควรพิจารณาใช้ FastAPI ครับ:
- ประสิทธิภาพสูง: FastAPI สร้างขึ้นบน Starlette (สำหรับ web parts) และ Pydantic (สำหรับ data parts) ซึ่งเป็นไลบรารีที่รวดเร็วและมีประสิทธิภาพสูงมาก มันใช้ Asynchronous programming (
async/await) ทำให้สามารถจัดการคำขอพร้อมกันได้จำนวนมาก - รวดเร็วในการพัฒนา: ด้วย Type Hints ของ Python และ Pydantic ทำให้การเขียนโค้ดชัดเจนขึ้น และลดข้อผิดพลาดในการพัฒนาลงอย่างมาก
- ตรวจสอบข้อมูลอัตโนมัติ (Automatic Data Validation): Pydantic ช่วยให้ FastAPI สามารถตรวจสอบข้อมูลที่เข้ามาใน API (Request Body, Query Parameters, Path Parameters) ได้โดยอัตโนมัติ หากข้อมูลไม่ตรงตาม Model ที่กำหนด ก็จะมีการแจ้งข้อผิดพลาดให้ทันที ช่วยประหยัดเวลาในการเขียนโค้ดตรวจสอบข้อมูล
- เอกสาร API อัตโนมัติ: FastAPI สร้างเอกสาร API แบบ Interactive (OpenAPI / Swagger UI) และ ReDoc ให้โดยอัตโนมัติจากโค้ดของคุณทันทีที่คุณรันแอปพลิเคชัน คุณไม่จำเป็นต้องเขียนเอกสารแยกต่างหาก ซึ่งช่วยลดภาระงานและมั่นใจได้ว่าเอกสารจะอัปเดตตรงกับโค้ดเสมอ
- ใช้ Python Type Hints: FastAPI ใช้ประโยชน์จาก Type Hints ของ Python 3.6+ อย่างเต็มที่ ซึ่งช่วยให้ IDEs (เช่น VS Code, PyCharm) สามารถให้คำแนะนำ (autocompletion) และตรวจสอบข้อผิดพลาด (type checking) ได้ดียิ่งขึ้น
- Dependency Injection System: ระบบ Dependency Injection ที่ทรงพลังและใช้งานง่าย ช่วยให้การจัดการการเชื่อมต่อฐานข้อมูล, การตรวจสอบสิทธิ์, หรือการ inject ค่าต่างๆ เข้าไปในฟังก์ชัน Path Operation ทำได้ง่ายและเป็นระเบียบ
- รองรับ Asynchronous Code: ออกแบบมาเพื่อรองรับ
asyncและawaitโดยตรง ทำให้การจัดการ I/O-bound operations (เช่น การอ่าน/เขียนฐานข้อมูล, การเรียก API ภายนอก) มีประสิทธิภาพสูงมาก
FastAPI กับ Flask และ Django: ตารางเปรียบเทียบ
เพื่อให้เห็นภาพชัดเจนขึ้น ลองมาดูตารางเปรียบเทียบ FastAPI กับเฟรมเวิร์ก Python ยอดนิยมอื่นๆ อย่าง Flask และ Django กันครับ
| คุณสมบัติ | FastAPI | Flask | Django |
|---|---|---|---|
| ประเภท | Micro-framework (เน้น API) | Micro-framework (ทั่วไป) | Full-stack framework |
| จุดเด่น | ประสิทธิภาพสูง, Type Hints, Pydantic, เอกสารอัตโนมัติ, Async/Await | เรียบง่าย, ยืดหยุ่น, เหมาะกับโปรเจกต์ขนาดเล็กถึงกลาง, มี Extension มากมาย | ครบวงจร, ORM ในตัว, Admin Panel, ระบบ Auth, เหมาะกับโปรเจกต์ขนาดใหญ่ |
| ความเร็ว/ประสิทธิภาพ | ยอดเยี่ยม (สร้างบน Starlette, Async) | ดี (ขึ้นอยู่กับการใช้งาน) | ดี (แต่มี Overhead มากกว่า) |
| การตรวจสอบข้อมูล (Validation) | อัตโนมัติด้วย Pydantic | ต้องใช้ไลบรารีภายนอก (เช่น Marshmallow) | Forms, Django REST Framework serializers |
| เอกสาร API | สร้างอัตโนมัติ (Swagger UI, ReDoc) | ต้องใช้ Extension หรือเขียนเอง | ต้องใช้ Extension (เช่น drf-yasg) |
| การจัดการฐานข้อมูล | ไม่รวมมาโดยตรง (ใช้ SQLAlchemy, Tortoise-ORM) | ไม่รวมมาโดยตรง (ใช้ SQLAlchemy) | ORM ในตัว |
| การเรียนรู้ | ปานกลาง (ต้องเข้าใจ Type Hints, Pydantic) | ง่าย (สำหรับพื้นฐาน) | ปานกลางถึงยาก (มีแนวคิดเฉพาะตัวเยอะ) |
| เหมาะสำหรับ | REST APIs, Microservices, Real-time APIs | เว็บแอปพลิเคชันขนาดเล็ก, APIs, Prototype | เว็บแอปพลิเคชันขนาดใหญ่, CMS, CRM |
| ชุมชน | กำลังเติบโตอย่างรวดเร็ว | ขนาดใหญ่และแข็งแกร่ง | ขนาดใหญ่และแข็งแกร่ง |
จากตารางจะเห็นได้ว่า FastAPI มีจุดเด่นในเรื่องของประสิทธิภาพ การตรวจสอบข้อมูล และการสร้างเอกสาร API อัตโนมัติ ซึ่งเป็นสิ่งที่สำคัญอย่างยิ่งในการสร้าง REST API ที่มีคุณภาพและรวดเร็วครับ
เตรียมความพร้อมก่อนเริ่มสร้าง API ด้วย FastAPI
ก่อนที่เราจะลงมือเขียนโค้ด FastAPI กัน มาเตรียมเครื่องมือที่จำเป็นกันก่อนนะครับ
ติดตั้ง Python และ Pip
FastAPI ต้องการ Python เวอร์ชั่น 3.7 ขึ้นไป ถ้าคุณยังไม่มี Python บนเครื่อง ให้ดาวน์โหลดและติดตั้งได้จาก เว็บไซต์ทางการของ Python ครับ ในระหว่างการติดตั้ง อย่าลืมทำเครื่องหมายที่ช่อง “Add Python to PATH” เพื่อให้สามารถเรียกใช้ Python จาก Command Line ได้ง่ายขึ้น
เมื่อติดตั้ง Python แล้ว Pip (Python’s package installer) จะถูกติดตั้งมาให้โดยอัตโนมัติครับ คุณสามารถตรวจสอบเวอร์ชันของ Python และ Pip ได้โดยเปิด Command Prompt หรือ Terminal แล้วพิมพ์คำสั่งเหล่านี้:
python --version
pip --version
สร้าง Virtual Environment (สภาพแวดล้อมเสมือน)
การใช้ Virtual Environment เป็น Best Practice ที่สำคัญในการพัฒนา Python ครับ เพราะมันช่วยให้เราแยกแพ็คเกจและ dependencies ของแต่ละโปรเจกต์ออกจากกัน ไม่ให้ไปชนกับโปรเจกต์อื่น หรือระบบหลักของคุณ
สร้างโฟลเดอร์สำหรับโปรเจกต์ของคุณแล้วเข้าไปในโฟลเดอร์นั้น:
mkdir my-fastapi-app
cd my-fastapi-app
สร้าง Virtual Environment (ในที่นี้เราจะตั้งชื่อว่า venv):
python -m venv venv
เปิดใช้งาน Virtual Environment:
- บน Windows:
.\venv\Scripts\activate - บน macOS / Linux:
source venv/bin/activate
เมื่อเปิดใช้งานแล้ว คุณจะเห็น (venv) นำหน้า Prompt ของคุณ ซึ่งหมายความว่าคุณกำลังอยู่ในสภาพแวดล้อมเสมือนแล้วครับ
ติดตั้ง FastAPI และ Uvicorn
ตอนนี้เราพร้อมที่จะติดตั้ง FastAPI และ Uvicorn แล้วครับ Uvicorn คือ ASGI server ที่ FastAPI ใช้ในการรันแอปพลิเคชัน
pip install fastapi "uvicorn[standard]"
คำสั่ง "uvicorn[standard]" จะติดตั้ง Uvicorn พร้อมกับ dependencies เพิ่มเติมที่จำเป็นสำหรับการใช้งานที่ครอบคลุมมากขึ้น เช่น python-dotenv และ watchfiles ซึ่งมีประโยชน์ในการพัฒนาครับ
ตรวจสอบว่าติดตั้งสำเร็จหรือไม่ โดยการพิมพ์ pip list คุณควรจะเห็น fastapi และ uvicorn ในรายการแพ็คเกจครับ
เยี่ยมมากครับ! ตอนนี้เรามีทุกอย่างพร้อมแล้วที่จะเริ่มสร้าง API ตัวแรกของเรา
สร้าง API แรกด้วย FastAPI: Hello World!
มาเริ่มต้นด้วย API ที่ง่ายที่สุด นั่นคือการส่งข้อความ “Hello World!” กลับไปเมื่อมีการเรียกใช้งานครับ
สร้างไฟล์ชื่อ main.py ในโฟลเดอร์โปรเจกต์ my-fastapi-app ของคุณ แล้วเพิ่มโค้ดต่อไปนี้:
# main.py
from fastapi import FastAPI
# สร้าง instance ของ FastAPI
# นี่คือจุดเริ่มต้นของแอปพลิเคชัน FastAPI ของเราครับ
app = FastAPI()
# กำหนด Path Operation Decorator สำหรับ HTTP GET request
# เมื่อมีคนเรียก URL ที่ root "/" ด้วย method GET ฟังก์ชันด้านล่างจะถูกเรียกใช้งาน
@app.get("/")
async def read_root():
"""
Endpoint สำหรับการทดสอบว่า API ทำงานได้หรือไม่
จะคืนค่าเป็น dictionary ที่มี key คือ "message" และ value คือ "Hello World!"
"""
return {"message": "Hello World!"}
# คุณสามารถเพิ่ม endpoint อื่นๆ ได้อีก เช่น
@app.get("/items/{item_id}")
async def read_item(item_id: int):
"""
Endpoint ตัวอย่างสำหรับการดึงข้อมูล item ตาม item_id
"""
return {"item_id": item_id, "description": "This is a sample item."}
คำอธิบายโค้ด:
from fastapi import FastAPI: นำเข้าคลาสFastAPIที่เป็นหัวใจหลักของเฟรมเวิร์กapp = FastAPI(): สร้างออบเจกต์appซึ่งเป็นตัวแทนของแอปพลิเคชัน FastAPI ของเรา@app.get("/"): นี่คือ Path Operation Decorator ที่บอก FastAPI ว่าฟังก์ชันread_rootด้านล่างควรถูกเรียกใช้เมื่อมีคำขอ HTTPGETไปยัง URL/(root path)async def read_root():: ฟังก์ชันนี้จะถูกเรียกใช้เมื่อมีคำขอGET /คำว่าasyncหมายความว่าฟังก์ชันนี้เป็น asynchronous function ซึ่ง FastAPI ถูกออกแบบมาให้รองรับโดยธรรมชาติครับreturn {"message": "Hello World!"}: FastAPI จะแปลง Python dictionary นี้ให้เป็น JSON response โดยอัตโนมัติ
รันแอปพลิเคชัน FastAPI ของคุณ
เปิด Command Prompt หรือ Terminal ของคุณ (ตรวจสอบให้แน่ใจว่าคุณยังอยู่ใน Virtual Environment นะครับ) แล้วรันคำสั่ง Uvicorn:
uvicorn main:app --reload
คำอธิบายคำสั่ง:
uvicorn: คือชื่อคำสั่งของ ASGI servermain:app: บอก Uvicorn ว่าให้หาออบเจกต์appภายในไฟล์main.py--reload: เป็นแฟล็กที่มีประโยชน์มากในการพัฒนาครับ มันจะทำให้ Uvicorn ตรวจจับการเปลี่ยนแปลงของโค้ดในไฟล์โปรเจกต์ของคุณโดยอัตโนมัติ และรีโหลดเซิร์ฟเวอร์ใหม่ทุกครั้งที่คุณบันทึกไฟล์ ทำให้คุณไม่ต้องหยุดและรันเซิร์ฟเวอร์ใหม่ด้วยตัวเอง
เมื่อรันคำสั่งแล้ว คุณจะเห็นข้อความประมาณนี้ใน Terminal:
INFO: Will watch for changes in these directories: ['/path/to/your/my-fastapi-app']
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.
ตอนนี้ API ของคุณกำลังทำงานอยู่ที่ http://127.0.0.1:8000 แล้วครับ! ลองเปิดเว็บบราวเซอร์ของคุณแล้วเข้า URL นี้ดู คุณควรจะเห็นข้อความ JSON: {"message": "Hello World!"}
สำรวจเอกสาร API อัตโนมัติ (Swagger UI / ReDoc)
นี่คือหนึ่งในฟีเจอร์ที่เจ๋งที่สุดของ FastAPI ครับ! มันสร้างเอกสาร API แบบ Interactive ให้คุณโดยอัตโนมัติ
- Swagger UI: เปิด
http://127.0.0.1:8000/docsในบราวเซอร์ของคุณ คุณจะเห็นหน้าเอกสารที่สวยงามและสามารถทดสอบ API ได้โดยตรงจากหน้าเว็บ - ReDoc: เปิด
http://127.0.0.1:8000/redocคุณจะเห็นเอกสารในอีกรูปแบบหนึ่งที่อ่านง่ายและมีรายละเอียดครบถ้วน
การมีเอกสาร API ที่อัปเดตอยู่เสมอและสามารถโต้ตอบได้ ช่วยให้การทำงานร่วมกันระหว่างทีมง่ายขึ้นมาก และลดเวลาในการสื่อสารลงได้อย่างมหาศาลเลยครับ
ทำความเข้าใจส่วนประกอบหลักของ FastAPI
หลังจากที่เราสร้าง “Hello World!” API ได้แล้ว มาเจาะลึกส่วนประกอบสำคัญที่ทำให้ FastAPI ทรงพลังและใช้งานง่ายกันครับ
Path Operations: จัดการ HTTP Methods (GET, POST, PUT, DELETE)
Path Operation คือฟังก์ชัน Python ที่ FastAPI จะเรียกใช้เมื่อมี HTTP request เข้ามาที่ URL และ HTTP method ที่ตรงกันครับ เราใช้ Decorator เพื่อกำหนด Path Operation
from fastapi import FastAPI
app = FastAPI()
# GET: ใช้สำหรับดึงข้อมูล
@app.get("/users")
async def get_users():
return [{"username": "alice"}, {"username": "bob"}]
# POST: ใช้สำหรับสร้างข้อมูลใหม่
@app.post("/users")
async def create_user(user: dict): # ในอนาคตเราจะใช้ Pydantic model แทน dict
return {"message": "User created", "user": user}
# PUT: ใช้สำหรับอัปเดตข้อมูลทั้งหมดของทรัพยากร
@app.put("/users/{user_id}")
async def update_user(user_id: int, user: dict):
return {"message": f"User {user_id} updated", "user": user}
# DELETE: ใช้สำหรับลบข้อมูล
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
return {"message": f"User {user_id} deleted"}
# PATCH: ใช้สำหรับอัปเดตข้อมูลบางส่วนของทรัพยากร (ไม่ครอบคลุมในตัวอย่างนี้ แต่ใช้งานคล้าย PUT)
# @app.patch("/users/{user_id}")
# async def patch_user(user_id: int, user_data: dict):
# return {"message": f"User {user_id} partially updated", "data": user_data}
Path Parameters: ส่งข้อมูลผ่าน URL
Path Parameters คือส่วนหนึ่งของ URL ที่เปลี่ยนแปลงได้ และใช้ในการระบุทรัพยากรเฉพาะเจาะจงครับ เช่น /items/1 โดย 1 คือ Path Parameter ที่เป็น ID ของ Item
เรากำหนด Path Parameter ใน Path ด้วยวงเล็บปีกกา {} และระบุ Type Hint ในฟังก์ชัน:
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int): # item_id จะถูกแปลงเป็น int โดยอัตโนมัติ
return {"item_id": item_id}
@app.get("/users/{username}")
async def read_user(username: str): # username จะเป็น string
return {"username": username}
# Path Parameters สามารถมี Type Hint ที่หลากหลายได้ เช่น float, bool, UUID
@app.get("/files/{file_id}")
async def read_file(file_id: str): # หากไม่ระบุ Type Hint จะถือว่าเป็น str
return {"file_id": file_id}
FastAPI จะตรวจสอบ Type ของ Path Parameter ให้โดยอัตโนมัติครับ หากค่าที่ส่งมาไม่ตรงกับ Type ที่กำหนด (เช่น ส่ง “abc” ไปยัง item_id: int) FastAPI จะคืนค่าข้อผิดพลาด 422 Unprocessable Entity โดยอัตโนมัติ ซึ่งเป็นประโยชน์อย่างมากในการทำ Data Validation ครับ
Query Parameters: ส่งข้อมูลเพิ่มเติมผ่าน URL
Query Parameters คือคู่ของ Key-Value ที่ปรากฏต่อจากเครื่องหมาย ? ใน URL และคั่นด้วย & ใช้สำหรับส่งข้อมูลเสริมที่ไม่ใช่ส่วนหนึ่งของ Path ครับ เช่น /items?skip=0&limit=10
ใน FastAPI เราสามารถกำหนด Query Parameters ได้ง่ายๆ โดยการเพิ่มพารามิเตอร์ลงในฟังก์ชัน Path Operation ที่มีค่าเริ่มต้น (default value) หรือใช้ Optional จากโมดูล typing
from fastapi import FastAPI
from typing import Optional
app = FastAPI()
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
"""
ดึงข้อมูล items โดยมี pagination:
- skip: จำนวน item ที่จะข้ามไป (default 0)
- limit: จำนวน item ที่จะดึงมา (default 10)
"""
return {"skip": skip, "limit": limit}
@app.get("/users/")
async def read_users(name: Optional[str] = None, age: Optional[int] = None):
"""
ดึงข้อมูล users โดยสามารถกรองด้วยชื่อและอายุได้
"""
result = {"message": "Fetching users"}
if name:
result.update({"name_filter": name})
if age:
result.update({"age_filter": age})
return result
หากพารามิเตอร์ไม่มีค่าเริ่มต้น (เช่น name: str โดยไม่มี = None) FastAPI จะถือว่ามันเป็น Query Parameter ที่จำเป็น (required) ครับ
Request Body: ส่งข้อมูลในรูปแบบ JSON ด้วย Pydantic
เมื่อเราต้องการสร้างหรืออัปเดตข้อมูลที่มีโครงสร้างซับซ้อน เราจะส่งข้อมูลเหล่านั้นไปใน Request Body ซึ่งส่วนใหญ่มักจะเป็นรูปแบบ JSON ครับ FastAPI ใช้ Pydantic ในการกำหนดโครงสร้างข้อมูลและทำ Data Validation โดยอัตโนมัติ
สร้าง Pydantic Model โดยการสืบทอดจาก BaseModel:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# สร้าง Pydantic Model สำหรับ Item
# นี่คือการกำหนดโครงสร้างและ Type ของข้อมูลที่เราคาดหวังจาก Request Body
class Item(BaseModel):
name: str
description: str | None = None # description อาจจะเป็น None ได้
price: float
tax: float | None = None
@app.post("/items/")
async def create_item(item: Item): # FastAPI จะรับ Request Body และแปลงเป็น Item object ให้โดยอัตโนมัติ
"""
สร้าง item ใหม่
"""
item_dict = item.model_dump() # แปลง Pydantic model เป็น dict
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({"price_with_tax": price_with_tax})
return item_dict
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
"""
อัปเดต item ตาม ID
"""
return {"item_id": item_id, **item.model_dump()}
เมื่อคุณส่ง JSON request body มา FastAPI จะตรวจสอบว่าข้อมูลนั้นตรงตาม Model Item หรือไม่ หากไม่ตรง เช่น ไม่มีฟิลด์ name หรือ price ไม่ใช่ตัวเลข FastAPI จะคืนค่าข้อผิดพลาด 422 Unprocessable Entity ทันที ซึ่งเป็นความสามารถที่ทรงพลังและช่วยประหยัดเวลาในการเขียนโค้ด Validation ได้มหาศาลครับ
Type Hints: หัวใจสำคัญของการตรวจสอบข้อมูลและเอกสารอัตโนมัติ
Type Hints (หรือ Type Annotations) คือการระบุชนิดข้อมูลของตัวแปร, พารามิเตอร์ฟังก์ชัน, และค่าที่ฟังก์ชันคืนกลับใน Python ครับ FastAPI ใช้ประโยชน์จาก Type Hints เหล่านี้อย่างเต็มที่เพื่อ:
- Data Validation: ตรวจสอบว่าข้อมูลที่รับเข้ามามี Type ตรงตามที่กำหนดหรือไม่
- Serialization: แปลงข้อมูลจาก Python objects ไปเป็น JSON และในทางกลับกัน
- Automatic Documentation: ใช้ Type Hints ในการสร้างเอกสาร OpenAPI (Swagger UI, ReDoc) ให้โดยอัตโนมัติ
- Editor Support: IDEs (เช่น VS Code, PyCharm) สามารถให้ Autocompletion และตรวจสอบข้อผิดพลาดเกี่ยวกับ Type ได้
ตัวอย่างการใช้ Type Hints ที่เราเห็นมาแล้ว:
# Path Parameter: item_id: int
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
# Query Parameters: skip: int = 0, limit: int = 10
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# Request Body: item: Item (Pydantic Model)
@app.post("/items/")
async def create_item(item: Item):
return item
การใช้ Type Hints ไม่ได้แค่ช่วยให้ FastAPI ทำงานได้ดีขึ้น แต่ยังช่วยให้โค้ดของคุณอ่านง่ายขึ้น บำรุงรักษาง่ายขึ้น และลดข้อผิดพลาดในระยะยาวอีกด้วยครับ
การจัดการข้อผิดพลาด (Error Handling) ด้วย HTTPException
เมื่อเกิดเหตุการณ์ที่ไม่คาดฝัน เช่น ผู้ใช้ส่ง ID ที่ไม่มีอยู่จริง หรือพยายามเข้าถึงทรัพยากรที่ไม่มีสิทธิ์ FastAPI มีคลาส HTTPException ที่ช่วยให้เราสามารถส่ง HTTP error response กลับไปได้อย่างง่ายดายและสอดคล้องกับมาตรฐาน
from fastapi import FastAPI, HTTPException, status # นำเข้า status เพื่อความชัดเจนของ HTTP status codes
app = FastAPI()
items_db = {
"foo": {"name": "Foo", "price": 50.2},
"bar": {"name": "Bar", "price": 62.0},
}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items_db:
# หาก item_id ไม่มีอยู่ในฐานข้อมูล ให้ raise HTTPException
# status.HTTP_404_NOT_FOUND คือรหัส 404
# detail คือข้อความที่จะส่งกลับไปใน JSON response
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
return items_db[item_id]
@app.post("/items/")
async def create_item(item_name: str):
if item_name in items_db:
# หากพยายามสร้าง item ที่มีอยู่แล้ว
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Item already exists")
items_db[item_name] = {"name": item_name, "price": 100.0}
return {"message": "Item created", "item": items_db[item_name]}
เมื่อ HTTPException ถูก raise, FastAPI จะจัดการสร้าง JSON response ที่มี status_code และ detail ตามที่คุณกำหนดโดยอัตโนมัติ ทำให้การจัดการข้อผิดพลาดเป็นมาตรฐานและง่ายต่อการใช้งานในฝั่ง Client ครับ
สร้าง REST API ระบบจัดการสินค้า (CRUD) แบบครบวงจร
ตอนนี้เรามาสร้าง API ที่ซับซ้อนขึ้นอีกนิด ด้วยระบบจัดการสินค้า (Products) ที่รองรับการดำเนินการแบบ CRUD (Create, Read, Update, Delete) ซึ่งเป็นพื้นฐานของ API ส่วนใหญ่ครับ เพื่อให้บทความกระชับและเน้นที่ FastAPI เราจะใช้ Dictionary ในหน่วยความจำเพื่อจำลองฐานข้อมูลนะครับ
โครงสร้างโปรเจกต์
เพื่อความเป็นระเบียบ เราจะแยกไฟล์ออกเป็นส่วนๆ ดังนี้ครับ
my-fastapi-app/
├── venv/
├── main.py
├── models.py
└── database.py
สร้าง Pydantic Models สำหรับข้อมูลสินค้า
สร้างไฟล์ models.py เพื่อกำหนดโครงสร้างข้อมูลของสินค้าครับ
# models.py
from pydantic import BaseModel, Field
from typing import Optional
# ProductBase: โมเดลพื้นฐานสำหรับข้อมูลสินค้าที่ต้องมี
class ProductBase(BaseModel):
name: str = Field(..., example="Notebook Pro X") # ... หมายถึง required
description: Optional[str] = Field(None, example="Powerful laptop for professionals")
price: float = Field(..., example=1200.00)
# เพิ่ม Field example เพื่อให้เอกสาร API ดูดีขึ้น
# ProductCreate: โมเดลสำหรับตอนสร้างสินค้าใหม่ (อาจจะเหมือน ProductBase)
# หรืออาจมีฟิลด์เพิ่มเติมที่จำเป็นตอนสร้างเท่านั้น
class ProductCreate(ProductBase):
pass
# Product: โมเดลสำหรับแสดงข้อมูลสินค้าที่ถูกสร้างแล้ว (รวม ID)
class Product(ProductBase):
id: int = Field(..., example=1) # id จะถูกสร้างโดยระบบ
# Pydantic Configuration:
# from_attributes = True ช่วยให้ Pydantic สามารถอ่านข้อมูลจาก ORM หรือ object ที่มี attribute
# แทนที่จะเป็นแค่ dictionary keys ทำให้ทำงานร่วมกับฐานข้อมูลได้ดีขึ้น
class Config:
from_attributes = True
json_schema_extra = {
"example": {
"id": 1,
"name": "Notebook Pro X",
"description": "Powerful laptop for professionals",
"price": 1200.00
}
}
เราสร้างโมเดล 3 ตัวเพื่อแยกความรับผิดชอบ:
ProductBase: ฟิลด์พื้นฐานที่สินค้าทุกตัวต้องมีProductCreate: ใช้สำหรับรับข้อมูลเมื่อผู้ใช้ต้องการสร้างสินค้าใหม่ (ในที่นี้เหมือนProductBase)Product: ใช้สำหรับส่งข้อมูลสินค้ากลับไปให้ผู้ใช้งาน ซึ่งจะรวมidที่ถูกสร้างขึ้นมาด้วย
จำลองฐานข้อมูลในหน่วยความจำ
สร้างไฟล์ database.py เพื่อจำลองฐานข้อมูลของเราครับ
# database.py
from typing import Dict
from models import Product
# จำลองฐานข้อมูลสินค้าในหน่วยความจำ
# ใช้ Dict เพื่อเก็บสินค้า โดย key คือ product ID
products_db: Dict[int, Product] = {}
next_product_id: int = 1 # ตัวนับ ID สำหรับสินค้าใหม่
def get_next_id() -> int:
global next_product_id
current_id = next_product_id
next_product_id += 1
return current_id
เราใช้ products_db เป็น Dictionary เพื่อเก็บข้อมูลสินค้า และ next_product_id เพื่อสร้าง ID สำหรับสินค้าใหม่ครับ
สร้าง Endpoint สำหรับการเพิ่มสินค้า (Create – POST)
แก้ไขไฟล์ main.py ของเราครับ
# main.py
from fastapi import FastAPI, HTTPException, status
from typing import List
from models import Product, ProductCreate
from database import products_db, get_next_id
app = FastAPI(
title="SiamLancard Product API",
description="API สำหรับจัดการข้อมูลสินค้าด้วย FastAPI.",
version="1.0.0"
)
@app.post("/products/", response_model=Product, status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate):
"""
สร้างสินค้าใหม่ในระบบ
- **name**: ชื่อสินค้า (จำเป็น)
- **description**: รายละเอียดสินค้า (ไม่จำเป็น)
- **price**: ราคาสินค้า (จำเป็น)
"""
new_id = get_next_id()
new_product_data = product.model_dump() # แปลง Pydantic model เป็น dict
# สร้าง instance ของ Product โดยใส่ id เข้าไป
new_product = Product(id=new_id, **new_product_data)
products_db[new_id] = new_product
return new_product
คำอธิบาย:
response_model=Product: บอก FastAPI ว่า Response ที่ส่งกลับไปจะมีโครงสร้างเหมือน Pydantic ModelProductซึ่งจะช่วยในการทำ Serialization และสร้างเอกสาร APIstatus_code=status.HTTP_201_CREATED: กำหนด HTTP status code ให้เป็น 201 Created เมื่อสร้างทรัพยากรสำเร็จproduct: ProductCreate: FastAPI จะรับ Request Body (JSON) มาตรวจสอบกับProductCreateModel ให้โดยอัตโนมัติ
สร้าง Endpoint สำหรับการดึงข้อมูลสินค้าทั้งหมด (Read All – GET)
เพิ่มโค้ดนี้ลงใน main.py
# main.py (ต่อจากโค้ดด้านบน)
@app.get("/products/", response_model=List[Product])
async def get_all_products():
"""
ดึงข้อมูลสินค้าทั้งหมดจากระบบ
"""
# คืนค่าเป็น List ของ Product objects
return list(products_db.values())
response_model=List[Product] บอก FastAPI ว่าจะคืนค่าเป็น List ของ Pydantic Model Product
สร้าง Endpoint สำหรับการดึงข้อมูลสินค้าตาม ID (Read One – GET)
เพิ่มโค้ดนี้ลงใน main.py
# main.py (ต่อจากโค้ดด้านบน)
@app.get("/products/{product_id}", response_model=Product)
async def get_product_by_id(product_id: int):
"""
ดึงข้อมูลสินค้าตาม ID
- **product_id**: รหัสสินค้า (จำเป็น)
"""
if product_id not in products_db:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
return products_db[product_id]
เราใช้ Path Parameter product_id: int และทำการตรวจสอบว่า ID นั้นมีอยู่ในฐานข้อมูลหรือไม่ หากไม่พบ ก็จะส่ง HTTPException 404 กลับไปครับ
สร้าง Endpoint สำหรับการอัปเดตข้อมูลสินค้า (Update – PUT)
เพิ่มโค้ดนี้ลงใน main.py
# main.py (ต่อจากโค้ดด้านบน)
@app.put("/products/{product_id}", response_model=Product)
async def update_product(product_id: int, product: ProductCreate):
"""
อัปเดตข้อมูลสินค้าที่มีอยู่
- **product_id**: รหัสสินค้าที่ต้องการอัปเดต (จำเป็น)
- **product**: ข้อมูลสินค้าใหม่ (name, description, price)
"""
if product_id not in products_db:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
# อัปเดตข้อมูลในฐานข้อมูล
updated_product_data = product.model_dump()
updated_product = Product(id=product_id, **updated_product_data)
products_db[product_id] = updated_product
return updated_product
คล้ายกับการสร้าง แต่เราจะตรวจสอบ product_id ก่อน แล้วจึงอัปเดตข้อมูลครับ
สร้าง Endpoint สำหรับการลบข้อมูลสินค้า (Delete – DELETE)
เพิ่มโค้ดนี้ลงใน main.py
# main.py (ต่อจากโค้ดด้านบน)
@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_product(product_id: int):
"""
ลบสินค้าออกจากระบบ
- **product_id**: รหัสสินค้าที่ต้องการลบ (จำเป็น)
"""
if product_id not in products_db:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
del products_db[product_id]
# HTTP 204 No Content ใช้เมื่อการดำเนินการสำเร็จแต่ไม่มีข้อมูลจะส่งคืน
return {"message": "Product deleted successfully"}
เมื่อลบสำเร็จ เราจะคืนค่าเป็น status_code=status.HTTP_204_NO_CONTENT ซึ่งหมายถึง “No Content” โดยทั่วไปแล้วจะไม่มี Body ใน Response สำหรับ 204 ครับ แต่ FastAPI จะส่ง JSON body เล็กๆ กลับไปให้ถ้าเรา return ค่า แต่ HTTP client ที่ดีจะละเว้นมันไปเองครับ
ตอนนี้คุณมี REST API ระบบจัดการสินค้าแบบ CRUD ที่ทำงานได้สมบูรณ์แล้วครับ! ลองรัน Uvicorn ด้วย uvicorn main:app --reload และไปที่ http://127.0.0.1:8000/docs เพื่อทดลองใช้งาน API ของคุณได้เลยครับ
เคล็ดลับ: การใช้เครื่องมืออย่าง Postman หรือ Insomnia จะช่วยให้การทดสอบ API ที่ใช้ HTTP POST, PUT, DELETE ทำได้ง่ายขึ้นมากครับ
สำหรับผู้ที่สนใจศึกษาเรื่องการจัดการฐานข้อมูลจริงๆ จังๆ ร่วมกับ FastAPI สามารถศึกษา การใช้งาน SQLAlchemy กับ FastAPI หรือ การเชื่อมต่อ MongoDB กับ FastAPI เพิ่มเติมได้ครับ
ฟีเจอร์ขั้นสูงของ FastAPI ที่ควรรู้
FastAPI ไม่ได้มีแค่พื้นฐานเท่านั้น แต่ยังมีฟีเจอร์ขั้นสูงอีกมากมายที่จะช่วยให้คุณสร้าง API ที่ซับซ้อนและมีประสิทธิภาพได้ดียิ่งขึ้นครับ
Dependency Injection: การพึ่งพิงฟังก์ชัน
Dependency Injection (DI) เป็น Pattern ที่ช่วยให้เราสามารถ “inject” หรือส่ง Dependencies (เช่น การเชื่อมต่อฐานข้อมูล, การตรวจสอบสิทธิ์ผู้ใช้ปัจจุบัน) เข้าไปในฟังก์ชัน Path Operation ของเราได้อย่างง่ายดายและเป็นระเบียบ
# main.py (ตัวอย่าง Dependency Injection)
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional, List, Dict
app = FastAPI()
# จำลองฐานข้อมูลผู้ใช้
fake_users_db = {
"alice": {"username": "alice", "email": "[email protected]"},
"bob": {"username": "bob", "email": "[email protected]"},
}
# Dependency Function: สมมติว่านี่คือฟังก์ชันที่ต้องตรวจสอบ API Key
# หรือเชื่อมต่อกับฐานข้อมูล
async def get_current_user(api_key: str = Depends(lambda x: x)): # ในความเป็นจริงจะใช้ Header หรือ Query
"""
Dependency นี้จะจำลองการตรวจสอบผู้ใช้จาก API Key
"""
if api_key not in fake_users_db:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API Key",
headers={"WWW-Authenticate": "Bearer"},
)
return fake_users_db[api_key]
@app.get("/users/me/")
async def read_users_me(current_user: Dict = Depends(get_current_user)):
"""
ดึงข้อมูลผู้ใช้ปัจจุบัน (ต้องมี API Key ที่ถูกต้อง)
"""
return current_user
# อีกตัวอย่างหนึ่งของ Dependency ที่ส่งค่าคงที่
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items_with_common_params/")
async def read_items_with_common_params(commons: Dict = Depends(common_parameters)):
"""
ดึงข้อมูล items โดยใช้ common parameters
"""
return commons
ข้อดีของ Dependency Injection:
- นำกลับมาใช้ใหม่ได้: ฟังก์ชัน Dependency สามารถถูกใช้ในหลาย Path Operation
- ทดสอบง่าย: สามารถ Mock Dependencies ได้ง่ายในการเขียน Unit Test
- จัดการทรัพยากร: สามารถใช้สำหรับจัดการ Session ฐานข้อมูล (open/close) ได้อย่างสะอาด
Security & Authentication: การรักษาความปลอดภัย API
FastAPI มีเครื่องมือช่วยในการสร้างระบบ Authentication และ Authorization ที่ดีเยี่ยม โดยรองรับมาตรฐาน OAuth2 และ JWT (JSON Web Tokens) ครับ
ตัวอย่างการใช้ OAuth2PasswordBearer สำหรับ Bearer Token Authentication:
# main.py (ตัวอย่าง Security)
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Dict
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # "token" คือ URL ที่จะใช้สำหรับรับ Token
# จำลองฐานข้อมูลผู้ใช้และ API Key/Password
fake_users_db_creds = {
"user_a": {"username": "user_a", "hashed_password": "supersecretpassword"},
"user_b": {"username": "user_b", "hashed_password": "anothersecret"},
}
# Dependency ที่ใช้สำหรับตรวจสอบ Token
async def get_current_active_user(token: str = Depends(oauth2_scheme)):
"""
ตรวจสอบ Token ที่ส่งมาใน Header (Authorization: Bearer <token>)
และคืนค่าผู้ใช้ที่เกี่ยวข้อง
"""
# ในโลกจริง ที่นี่จะเป็นการถอดรหัส JWT และตรวจสอบความถูกต้อง
# แต่สำหรับตัวอย่างนี้ เราจะใช้ token เป็น username ตรงๆ
user = fake_users_db_creds.get(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return user
@app.get("/secure_data/")
async def read_secure_data(current_user: Dict = Depends(get_current_active_user)):
"""
Endpoint ที่ต้องการการยืนยันตัวตน (Authentication)
"""
return {"message": f"Hello, {current_user['username']}! This is secure data."}
เมื่อเข้าถึง /secure_data/ โดยไม่มี Token หรือ Token ไม่ถูกต้อง FastAPI จะส่ง HTTP 401 Unauthorized กลับไปโดยอัตโนมัติ และในเอกสาร Swagger UI คุณจะเห็นปุ่ม “Authorize” สำหรับทดสอบการส่ง Token ได้ครับ
APIRouter: การจัดโครงสร้างโปรเจกต์ขนาดใหญ่ให้เป็นระเบียบ
สำหรับโปรเจกต์ขนาดใหญ่ การเก็บ Path Operation ทั้งหมดไว้ในไฟล์ main.py ไฟล์เดียวอาจทำให้โค้ดยาวและจัดการยาก APIRouter ช่วยให้เราสามารถแยก Path Operation ไปยังไฟล์โมดูลอื่นๆ และนำมารวมกันใน main.py ได้อย่างเป็นระเบียบ
สมมติว่าเรามีไฟล์ routers/products.py และ routers/users.py
routers/products.py:
# routers/products.py
from fastapi import APIRouter, HTTPException, status
from typing import List
from models import Product, ProductCreate
from database import products_db, get_next_id
router = APIRouter(
prefix="/products", # กำหนด prefix สำหรับทุก Path ใน Router นี้
tags=["products"], # ใช้สำหรับจัดกลุ่มในเอกสาร API
responses={404: {"description": "Product not found"}},
)
@router.post("/", response_model=Product, status_code=status.HTTP_201_CREATED)
async def create_product(product: ProductCreate):
new_id = get_next_id()
new_product = Product(id=new_id, **product.model_dump())
products_db[new_id] = new_product
return new_product
@router.get("/", response_model=List[Product])
async def get_all_products():
return list(products_db.values())
@router.get("/{product_id}", response_model=Product)
async def get_product_by_id(product_id: int):
if product_id not in products_db:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found")
return products_db[product_id]
# ... เพิ่ม PUT, DELETE ตามตัวอย่าง CRUD
main.py:
# main.py
from fastapi import FastAPI
from routers import products # นำเข้า router ที่เราสร้าง
app = FastAPI(
title="SiamLancard Product API (Modular)",
description="API จัดการสินค้าแบบแยกโมดูล",
version="1.0.0"
)
# นำ router เข้ามารวมกับแอปพลิเคชันหลัก
app.include_router(products.router)
@app.get("/")
async def root():
return {"message": "Welcome to SiamLancard Product API!"}
เมื่อรันแอปพลิเคชัน Path ทั้งหมดใน products.router จะถูกเพิ่มเข้าใน /products โดยอัตโนมัติ ทำให้โครงสร้างโค้ดสะอาดและจัดการง่ายขึ้นมากครับ
Middlewares: การดักจับและประมวลผลคำขอก่อนถึง Endpoint
Middleware คือฟังก์ชันที่ทำงานก่อนและหลัง Path Operation ครับ สามารถใช้สำหรับ:
- การบันทึก Log ของทุกๆ Request
- การจัดการ Cross-Origin Resource Sharing (CORS)
- การเพิ่ม Header บางอย่างให้กับ Response
- การตรวจสอบสิทธิ์เบื้องต้น
ตัวอย่างการใช้งาน CORS Middleware (จำเป็นสำหรับ Frontend ที่รันคนละ Origin กับ Backend):
# main.py (ตัวอย่าง Middleware)
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.middleware.cors import CORSMiddleware
import time
app = FastAPI()
# เพิ่ม CORS Middleware
origins = [
"http://localhost:3000", # ตัวอย่าง Frontend URL
"http://127.0.0.1:3000",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # อนุญาต Origin ที่กำหนด
allow_credentials=True, # อนุญาต Cookie หรือ Authorization headers
allow_methods=["*"], # อนุญาตทุก HTTP methods
allow_headers=["*"], # อนุญาตทุก Headers
)
# ตัวอย่าง Custom Middleware สำหรับบันทึกเวลาประมวลผล
@app.middleware("http")
async def add_process_time_header(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 to {request.url.path} processed in {process_time:.4f} seconds")
return response
@app.get("/ping")
async def ping():
return {"message": "pong"}
เมื่อเรียก /ping คุณจะเห็น Header X-Process-Time ใน Response และ Log ใน Terminal ครับ
Background Tasks: การทำงานแบบเบื้องหลัง
บางครั้งเราต้องการให้ API ส่ง Response กลับไปให้ Client ทันที แต่ยังมีการทำงานบางอย่างที่ต้องทำต่อในเบื้องหลัง เช่น การส่งอีเมลแจ้งเตือน หรือการบันทึก Log ที่ไม่จำเป็นต้องรอให้เสร็จสิ้น FastAPI มี BackgroundTasks สำหรับจัดการสิ่งนี้ครับ
# main.py (ตัวอย่าง Background Tasks)
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def write_log(message: str):
with open("api.log", mode="a") as log_file:
log_file.write(f"{message}\n")
@app.post("/send-notification/")
async def send_notification(email: str, message: str, background_tasks: BackgroundTasks):
"""
ส่งข้อความแจ้งเตือนและบันทึก Log ในเบื้องหลัง
"""
background_tasks.add_task(write_log, f"Notification sent to {email}: {message}")
# สามารถเพิ่ม task อื่นๆ เช่น ส่งอีเมลจริง
# background_tasks.add_task(send_email, email, "Notification", message)
return {"message": "Notification will be sent in the background"}
เมื่อเรียก Endpoint นี้ ผู้ใช้จะได้รับ Response ทันที และฟังก์ชัน write_log จะทำงานในเบื้องหลังโดยไม่บล็อกการตอบสนองของ API ครับ
การทดสอบ API ด้วย FastAPI TestClient
การทดสอบเป็นส่วนสำคัญของการพัฒนาซอฟต์แวร์ครับ FastAPI มี TestClient ที่สร้างขึ้นบน Starlette’s TestClient ซึ่งช่วยให้เราสามารถทดสอบ Path Operations ได้อย่างง่ายดายและมีประสิทธิภาพ โดยไม่ต้องรันเซิร์ฟเวอร์จริง
สร้างไฟล์ test_main.py ในโฟลเดอร์โปรเจกต์ของคุณ:
# test_main.py
from fastapi.testclient import TestClient
from main import app # นำเข้า app instance จากไฟล์ main.py ของเรา
from models import Product # นำเข้า Product model จาก models.py
from database import products_db # นำเข้าฐานข้อมูลจำลอง
# สร้าง TestClient จาก app instance
client = TestClient(app)
# ก่อนเริ่มการทดสอบแต่ละครั้ง หรือก่อนชุดการทดสอบทั้งหมด
# เราสามารถเคลียร์ฐานข้อมูลจำลองได้ เพื่อให้การทดสอบเป็นอิสระต่อกัน
def setup_function():
products_db.clear()
# รีเซ็ต next_product_id ด้วยหากต้องการ
from database import next_product_id # ต้องอ้างอิงจากโมดูลเพื่อแก้ไข global variable
next_product_id = 1
def test_read_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Welcome to SiamLancard Product API!"}
def test_create_product():
setup_function() # เคลียร์ DB ก่อนทดสอบ
response = client.post(
"/products/",
json={"name": "Test Product", "description": "A product for testing", "price": 99.99}
)
assert response.status_code == 201
assert response.json()["name"] == "Test Product"
assert response.json()["id"] == 1 # ตรวจสอบ ID ที่ถูกสร้าง
assert len(products_db) == 1 # ตรวจสอบว่ามีสินค้าใน DB
def test_get_all_products_empty():
setup_function()
response = client.get("/products/")
assert response.status_code == 200
assert