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

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

March 17, 2026
0 Views

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

สารบัญ

REST API คืออะไร?

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

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

ส่วน REST (Representational State Transfer) คือสถาปัตยกรรม (Architectural Style) สำหรับการออกแบบ API บนเว็บ ที่ได้รับความนิยมอย่างแพร่หลายเนื่องจากความเรียบง่ายและยืดหยุ่น โดยมีหลักการสำคัญดังนี้ครับ:

  • Client-Server (ไคลเอนต์-เซิร์ฟเวอร์): การแยกส่วนไคลเอนต์ (ผู้ใช้งาน) และเซิร์ฟเวอร์ (ผู้ให้บริการ) ออกจากกันอย่างชัดเจน ทำให้แต่ละส่วนสามารถพัฒนาได้อย่างอิสระ
  • Stateless (ไร้สถานะ): แต่ละคำขอจากไคลเอนต์ไปยังเซิร์ฟเวอร์ต้องมีข้อมูลที่จำเป็นทั้งหมดในการประมวลผลคำขอนั้น ๆ เซิร์ฟเวอร์จะไม่เก็บข้อมูลสถานะของไคลเอนต์ระหว่างคำขอ
  • Cacheable (แคชได้): ข้อมูลที่ส่งคืนจากเซิร์ฟเวอร์สามารถระบุได้ว่าสามารถแคชได้หรือไม่ เพื่อปรับปรุงประสิทธิภาพการทำงาน
  • Layered System (ระบบหลายชั้น): ไคลเอนต์อาจไม่รู้ว่ากำลังเชื่อมต่อโดยตรงกับเซิร์ฟเวอร์ปลายทาง หรือผ่านตัวกลางอย่าง Load Balancer หรือ Proxy
  • Uniform Interface (ส่วนต่อประสานที่สอดคล้องกัน): นี่คือหลักการที่สำคัญที่สุด โดย API จะใช้ชุดของส่วนต่อประสานมาตรฐาน ทำให้ระบบต่าง ๆ เข้าใจและสื่อสารกันได้ง่ายขึ้น ซึ่งรวมถึง:
    • Resources (ทรัพยากร): ทุกอย่างใน REST API ถือเป็นทรัพยากร (เช่น ผู้ใช้, สินค้า, รายการงาน) และมี URI (Uniform Resource Identifier) เฉพาะในการเข้าถึง
    • HTTP Methods (วิธีการ HTTP): ใช้ Verb มาตรฐานของ HTTP ในการระบุการกระทำที่ต้องการกับทรัพยากร ได้แก่:
      • GET: ดึงข้อมูลทรัพยากร
      • POST: สร้างทรัพยากรใหม่
      • PUT: อัปเดตทรัพยากรทั้งหมด
      • PATCH: อัปเดตทรัพยากรบางส่วน
      • DELETE: ลบทรัพยากร
    • Representation (การนำเสนอ): ข้อมูลจะถูกนำเสนอในรูปแบบที่อ่านง่าย เช่น JSON (JavaScript Object Notation) หรือ XML

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

ทำไมต้องเลือก FastAPI สำหรับการพัฒนา REST API?

ในบรรดาเฟรมเวิร์ก Python สำหรับการพัฒนาเว็บและ API นั้น FastAPI โดดเด่นขึ้นมาด้วยเหตุผลหลายประการครับ

  • ประสิทธิภาพสูง: FastAPI สร้างขึ้นบน Starlette (สำหรับเว็บ) และ Pydantic (สำหรับการตรวจสอบข้อมูล) ทำให้เป็นหนึ่งในเฟรมเวิร์ก Python ที่เร็วที่สุด สามารถรองรับการทำงานแบบ Asynchronous (async/await) ได้เต็มรูปแบบ ทำให้จัดการคำขอพร้อมกันได้จำนวนมาก
  • ความเร็วในการพัฒนา: ด้วยคุณสมบัติอย่าง Type Hints ของ Python และ Pydantic ทำให้โค้ดมีความชัดเจน ตรวจสอบข้อผิดพลาดได้ง่าย และลดเวลาในการดีบักอย่างมาก นอกจากนี้ การสร้างเอกสาร API อัตโนมัติยังช่วยลดภาระงานส่วนนี้ไปได้เยอะครับ
  • การตรวจสอบข้อมูลอัตโนมัติ (Data Validation): Pydantic ช่วยให้คุณสามารถกำหนดรูปแบบข้อมูล (Schema) ได้อย่างง่ายดาย และ FastAPI จะทำการตรวจสอบข้อมูลที่เข้ามา (Request Body) หรือข้อมูลที่ส่งออกไป (Response) ให้โดยอัตโนมัติ ทำให้มั่นใจได้ว่าข้อมูลถูกต้องตามที่คาดหวัง
  • เอกสาร API อัตโนมัติ: FastAPI สร้างเอกสาร API แบบอินเทอร์แอกทีฟ (Interactive Documentation) ให้โดยอัตโนมัติ ทั้งในรูปแบบ Swagger UI และ ReDoc ซึ่งช่วยให้นักพัฒนาสามารถทำความเข้าใจ ทดสอบ และโต้ตอบกับ API ได้อย่างง่ายดาย
  • Dependency Injection (การฉีดพึ่งพิง): ระบบ DI ที่ทรงพลังของ FastAPI ช่วยให้การจัดการการพึ่งพิง (Dependencies) ทำได้ง่ายและมีประสิทธิภาพ ทำให้โค้ดโมดูลาร์ขึ้น ทดสอบง่ายขึ้น และลดความซับซ้อน
  • รองรับ Asynchronous (Async/Await): FastAPI สร้างขึ้นมาเพื่อรองรับการทำงานแบบ Asynchronous โดยเฉพาะ ทำให้เหมาะอย่างยิ่งสำหรับแอปพลิเคชันที่ต้องการประสิทธิภาพสูงและสามารถจัดการ I/O-bound operations (เช่น การเชื่อมต่อฐานข้อมูล, การเรียกใช้ API ภายนอก) ได้อย่างมีประสิทธิภาพ
  • ความปลอดภัย: มีเครื่องมือและแนวทางสำหรับการรักษาความปลอดภัย เช่น OAuth2 with Bearer Tokens ที่ช่วยให้การสร้างระบบยืนยันตัวตนและการอนุญาตเป็นไปได้ง่าย
  • ชุมชนและการเติบโต: FastAPI มีชุมชนผู้ใช้งานที่เติบโตอย่างรวดเร็ว พร้อมเอกสารประกอบที่ยอดเยี่ยมและตัวอย่างการใช้งานมากมาย

ลองดูตารางเปรียบเทียบ FastAPI กับเฟรมเวิร์ก Python อื่น ๆ ที่นิยมใช้ในการสร้าง API ครับ:

คุณสมบัติ FastAPI Flask Django REST Framework
ความเร็ว/ประสิทธิภาพ สูงมาก (รองรับ Async/Await) ปานกลาง (Sync เป็นหลัก) ปานกลาง (Sync เป็นหลัก, แต่มี async รองรับในเวอร์ชันใหม่)
ความเร็วในการพัฒนา สูง (Type Hints, Pydantic, Auto-docs) สูง (Minimalist, แต่ต้องเพิ่ม lib เอง) ปานกลาง (Boilerplate เยอะกว่า)
การตรวจสอบข้อมูล อัตโนมัติด้วย Pydantic ต้องใช้ libraries เพิ่มเติม (เช่น Marshmallow) มีในตัว (Serializers)
เอกสาร API อัตโนมัติ มีในตัว (Swagger UI, ReDoc) ต้องใช้ libraries เพิ่มเติม (เช่น Flask-RESTX) ต้องใช้ libraries เพิ่มเติม (เช่น drf-yasg)
Dependency Injection มีในตัว (ทรงพลัง) ต้องสร้างเองหรือใช้ libraries มีในตัว (View-based dependencies)
เหมาะสำหรับ API, Microservices, Real-time apps Web apps ขนาดเล็ก, APIs, Microservices Web apps ขนาดใหญ่, APIs (Full-stack)
Learning Curve ปานกลาง (ต้องเข้าใจ Type Hints, Async) ต่ำ (Minimalist) สูง (เยอะกว่า Flask, ต้องเข้าใจ Django)

เริ่มต้นกับ FastAPI: การติดตั้งและการตั้งค่าเบื้องต้น

มาเริ่มกันเลยครับ! ในส่วนนี้เราจะทำการติดตั้ง FastAPI และสร้าง API “Hello World” แรกของเรา

สิ่งที่ต้องเตรียม

ก่อนอื่น คุณต้องแน่ใจว่าได้ติดตั้ง Python ไว้ในเครื่องแล้ว แนะนำให้ใช้ Python 3.7 ขึ้นไปครับ

คุณสามารถตรวจสอบเวอร์ชัน Python ได้โดยเปิด Terminal (หรือ Command Prompt) และพิมพ์:

python --version

หรือถ้ามี Python 2 และ 3 ติดตั้งอยู่ ให้ใช้:

python3 --version

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

FastAPI ไม่ได้เป็นเซิร์ฟเวอร์เว็บโดยตรง แต่เป็นเฟรมเวิร์กสำหรับสร้าง API ดังนั้นเราจึงต้องใช้ ASGI server (Asynchronous Server Gateway Interface) อย่าง Uvicorn เพื่อรันแอปพลิเคชัน FastAPI ของเราครับ

แนะนำให้สร้าง Virtual Environment เพื่อแยกการจัดการแพ็กเกจของโปรเจกต์ออกจากระบบหลัก:

  1. สร้าง Virtual Environment: 
    python3 -m venv venv

    คำสั่งนี้จะสร้างโฟลเดอร์ชื่อ venv ที่มีสภาพแวดล้อม Python แยกต่างหาก

  2. เปิดใช้งาน Virtual Environment:
    • บน macOS/Linux: 
      source venv/bin/activate
    • บน Windows (Command Prompt): 
      venv\Scripts\activate.bat
    • บน Windows (PowerShell): 
      .\venv\Scripts\Activate.ps1

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

  3. ติดตั้ง FastAPI และ Uvicorn: 
    pip install fastapi "uvicorn[standard]"

    "uvicorn[standard]" จะติดตั้ง Uvicorn พร้อมกับแพ็กเกจเสริมที่จำเป็น (เช่น python-dotenv, httptools, watchfiles) เพื่อการทำงานที่ครบถ้วนและสะดวกสบาย

FastAPI “Hello World” แรกของคุณ

มาสร้างไฟล์แรกของเรากันครับ ตั้งชื่อไฟล์ว่า main.py แล้วใส่โค้ดต่อไปนี้:

# main.py
from fastapi import FastAPI

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

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

# กำหนด Path Operation สำหรับ HTTP GET ที่ "/items/{item_id}"
# {item_id} เป็น Path Parameter
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None):
    """
    Endpoint สำหรับดึงข้อมูล item โดยใช้ item_id และ query parameter q
    """
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

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

  • from fastapi import FastAPI: นำเข้าคลาส FastAPI
  • app = FastAPI(): สร้างอินสแตนซ์ของแอปพลิเคชัน FastAPI
  • @app.get("/"): นี่คือ Path Operation Decorator ที่บอก FastAPI ว่าฟังก์ชัน read_root ควรถูกเรียกเมื่อมีการร้องขอ HTTP GET ไปยัง URL root (/)
  • async def read_root():: ฟังก์ชันนี้ถูกกำหนดให้เป็น async เพราะ FastAPI ถูกออกแบบมาเพื่อรองรับ Asynchronous programming ซึ่งช่วยให้แอปพลิเคชันสามารถจัดการคำขอพร้อมกันได้ดียิ่งขึ้น
  • return {"message": "Hello, FastAPI World!"}: ฟังก์ชันจะส่งคืน Python dictionary ซึ่ง FastAPI จะแปลงเป็น JSON response โดยอัตโนมัติ
  • @app.get("/items/{item_id}"): ตัวอย่างของ Path Parameter โดย {item_id} จะถูกส่งผ่านเป็นอาร์กิวเมนต์ไปยังฟังก์ชัน read_item
  • item_id: int: นี่คือ Type Hint ที่บอก FastAPI ว่า item_id ควรจะเป็นชนิดข้อมูล int ซึ่ง FastAPI จะใช้ Pydantic ในการตรวจสอบข้อมูลนี้โดยอัตโนมัติ
  • q: str | None = None: นี่คือ Query Parameter ที่เป็นทางเลือก (Optional) และมีค่าเริ่มต้นเป็น None

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

เปิด Terminal (ที่เปิด Virtual Environment อยู่) และรันคำสั่ง Uvicorn:

uvicorn main:app --reload

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

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

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

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 [yyyyy]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

ตอนนี้ API ของคุณกำลังทำงานอยู่ที่ http://127.0.0.1:8000 แล้วครับ

ลองเปิดเบราว์เซอร์และเข้าไปที่:

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

คุณจะเห็น JSON response ที่ถูกส่งกลับมาครับ

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

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

ในขณะที่ Uvicorn กำลังทำงานอยู่ ลองเปิดเบราว์เซอร์แล้วไปที่:

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

    นี่คือ UI ที่ช่วยให้คุณสามารถดูเอกสาร API ของคุณ ทดสอบ Path Operations ต่าง ๆ และส่ง Request ได้โดยตรงจากหน้าเว็บ

  • ReDoc: http://127.0.0.1:8000/redoc

    เป็นอีกรูปแบบหนึ่งของเอกสาร API ที่อ่านง่ายและสวยงาม เหมาะสำหรับการเป็นเอกสารอ้างอิง

คุณจะเห็นว่า Endpoint ที่เราสร้างไปเมื่อสักครู่ถูกแสดงอยู่ในเอกสารเหล่านี้ พร้อมทั้งข้อมูลของ Path Parameters, Query Parameters และ Response Model ซึ่งช่วยให้นักพัฒนาทั้งฝั่ง Backend และ Frontend ทำงานร่วมกันได้ง่ายขึ้นมากครับ

แก่นแนวคิดสำคัญของ FastAPI

มาเจาะลึกแนวคิดหลัก ๆ ที่ทำให้ FastAPI ทรงพลังและใช้งานง่ายกันครับ

Path Operations (การดำเนินการบนเส้นทาง)

เราได้เห็น Path Operations ไปแล้วในตัวอย่าง “Hello World” ครับ โดยพื้นฐานแล้ว Path Operations คือฟังก์ชัน Python ที่ถูกกำหนดให้รันเมื่อมีการร้องขอ HTTP ที่ตรงกับเส้นทาง (Path) และวิธีการ (Method) ที่กำหนดไว้

FastAPI มี decorator สำหรับ HTTP method หลัก ๆ ดังนี้:

  • @app.get(): สำหรับการดึงข้อมูล
  • @app.post(): สำหรับการสร้างข้อมูลใหม่
  • @app.put(): สำหรับการอัปเดตข้อมูลทั้งหมด
  • @app.patch(): สำหรับการอัปเดตข้อมูลบางส่วน
  • @app.delete(): สำหรับการลบข้อมูล
  • @app.options(), @app.head(), @app.trace(): สำหรับ HTTP method อื่น ๆ ที่ใช้น้อยกว่า

Path Parameters:

ดังที่เราเห็นใน /items/{item_id}, Path Parameters จะถูกระบุในวงเล็บปีกกา {} ใน decorator และจะถูกส่งผ่านเป็นอาร์กิวเมนต์ไปยังฟังก์ชัน Path Operation ครับ

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
    return {"user_id": user_id, "item_id": item_id}

Query Parameters:

Query Parameters คือพารามิเตอร์ที่ปรากฏหลังเครื่องหมาย ? ใน URL (เช่น /items?skip=0&limit=10) ใน FastAPI คุณสามารถกำหนด Query Parameters ได้โดยการประกาศอาร์กิวเมนต์ในฟังก์ชัน Path Operation ที่ไม่ได้เป็นส่วนหนึ่งของ Path Parameters

@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

ในตัวอย่างนี้ skip และ limit มีค่าเริ่มต้นเป็น 0 และ 10 ตามลำดับ ทำให้เป็น Optional Query Parameters ครับ

Parameter Types:

FastAPI ใช้ Python Type Hints ในการตรวจสอบและแปลงข้อมูลโดยอัตโนมัติ ไม่ว่าจะเป็น int, str, float, bool, หรือแม้กระทั่ง UUID, datetime ซึ่งเป็นประโยชน์อย่างมากในการลดข้อผิดพลาดและเพิ่มความชัดเจนของโค้ด

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

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

Pydantic Model คืออะไร?

Pydantic เป็นไลบรารี Python สำหรับการตรวจสอบข้อมูล (Data Validation) และการตั้งค่า (Settings Management) โดยใช้ Python Type Hints ทำให้คุณสามารถกำหนดโครงสร้างข้อมูลที่ต้องการได้อย่างง่ายดาย

ตัวอย่างการใช้งาน Request Body:

มาลองสร้าง API สำหรับสร้าง Item ใหม่กันครับ

# main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# ... (Path Operations เดิม) ...

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

@app.post("/items/")
async def create_item(item: Item):
    """
    Endpoint สำหรับสร้าง Item ใหม่
    """
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

# ลองทดสอบด้วย POST request ไปที่ /items/
# Body (JSON):
# {
#   "name": "Foo",
#   "description": "A very nice Item",
#   "price": 35.4,
#   "tax": 3.2
# }

คำอธิบาย:

  • class Item(BaseModel):: เราสร้างคลาส Item ที่สืบทอดมาจาก pydantic.BaseModel
  • ภายในคลาส Item เรากำหนดฟิลด์ต่าง ๆ พร้อม Type Hints (name: str, price: float) และฟิลด์ที่เป็น Optional (description: str | None = None, tax: float | None = None)
  • ในฟังก์ชัน create_item เราประกาศอาร์กิวเมนต์ item: Item ซึ่ง FastAPI จะรู้ว่านี่คือ Request Body และจะใช้ Pydantic ในการ:
    1. อ่าน Request Body ที่เข้ามาในรูปแบบ JSON
    2. ตรวจสอบข้อมูลว่าตรงกับโครงสร้างของ Item Model หรือไม่
    3. ถ้าข้อมูลถูกต้อง จะแปลงเป็นอินสแตนซ์ของ Item และส่งไปยังฟังก์ชัน
    4. ถ้าข้อมูลไม่ถูกต้อง FastAPI จะส่ง HTTP 422 Unprocessable Entity กลับไปโดยอัตโนมัติพร้อมรายละเอียดข้อผิดพลาด

คุณสามารถลองส่ง POST Request ไปยัง http://127.0.0.1:8000/items/ จาก Swagger UI (/docs) หรือใช้เครื่องมืออย่าง Postman ก็ได้ครับ

การกำหนด Response Model

เช่นเดียวกับการจัดการ Request Body คุณสามารถกำหนดรูปแบบของข้อมูลที่จะส่งคืนเป็น Response ได้ด้วยการใช้พารามิเตอร์ response_model ใน Path Operation decorator ครับ นี่มีประโยชน์อย่างมากในการ:

  • กรองข้อมูล: หากคุณมี Pydantic Model ที่มีฟิลด์ข้อมูลที่ละเอียดอ่อน (เช่น รหัสผ่าน) คุณสามารถสร้าง Response Model ที่มีเพียงฟิลด์ที่ต้องการเปิดเผยต่อสาธารณะได้
  • การสร้างเอกสาร: FastAPI จะใช้ response_model ในการสร้างเอกสาร API ที่ถูกต้อง ทำให้ผู้ใช้ API ทราบถึงโครงสร้างของข้อมูลที่จะได้รับ

ตัวอย่างการใช้ Response Model:

สมมติว่าเรามี User Model ที่มีฟิลด์ hashed_password ที่ไม่ควรส่งกลับไปใน Response

# main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# ... (Path Operations เดิม) ...

class UserInDB(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    hashed_password: str # ไม่ควรเปิดเผย

class UserPublic(BaseModel): # Response Model
    username: str
    email: str | None = None
    full_name: str | None = None

@app.post("/users/", response_model=UserPublic)
async def create_user(user: UserInDB): # รับข้อมูล UserInDB
    # ในความเป็นจริงเราจะ hash รหัสผ่านและบันทึกลงฐานข้อมูล
    # ที่นี่เราแค่จำลองการสร้าง user
    fake_db_user = UserInDB(
        username=user.username,
        email=user.email,
        full_name=user.full_name,
        hashed_password="some_hashed_password" # สมมติว่านี่คือรหัสผ่านที่ถูก hash
    )
    # FastAPI จะใช้ UserPublic ในการกรองข้อมูลก่อนส่งกลับ
    return fake_db_user

เมื่อคุณเรียกใช้ Endpoint /users/ ด้วย POST Request และส่งข้อมูล username, email, full_name และ hashed_password (ที่เราสมมติว่ามี) FastAPI จะส่งคืนเฉพาะข้อมูลที่อยู่ใน UserPublic Model เท่านั้น นั่นคือ username, email และ full_name โดยไม่มี hashed_password ปรากฏใน Response เลยครับ

Dependency Injection (การฉีดพึ่งพิง)

Dependency Injection (DI) เป็นหนึ่งในคุณสมบัติที่ทรงพลังที่สุดของ FastAPI ช่วยให้คุณสามารถจัดการการพึ่งพิง (Dependencies) ระหว่างส่วนต่าง ๆ ของแอปพลิเคชันได้อย่างเป็นระเบียบและมีประสิทธิภาพ

มันทำงานอย่างไร?

คุณสามารถกำหนดฟังก์ชัน (หรือคลาส) ที่เป็น “Dependency” โดยฟังก์ชัน Path Operation ของคุณสามารถประกาศว่าต้องการ Dependency เหล่านี้ FastAPI จะรับผิดชอบในการเรียกใช้ Dependency และส่งค่าที่ส่งคืนมาให้กับฟังก์ชัน Path Operation ของคุณโดยอัตโนมัติ

ประโยชน์ของ DI:

  • ลดการทำซ้ำโค้ด: ใช้ฟังก์ชัน Dependency ซ้ำได้หลาย Path Operations
  • ทดสอบง่ายขึ้น: สามารถ Mock Dependencies ได้ง่ายในการทดสอบ
  • โค้ดสะอาดขึ้น: แยกส่วนการทำงานออกจากกันชัดเจน
  • จัดการทรัพยากร: เช่น การเชื่อมต่อฐานข้อมูล, การยืนยันตัวตน

ตัวอย่าง Dependency Injection:

สมมติว่าเราต้องการ Query Parameter token ที่ใช้ในการยืนยันตัวตนเบื้องต้น

# main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# ... (Path Operations และ Models เดิม) ...

# สร้าง Dependency function
async def get_current_user(token: str):
    if token != "johndoe": # สมมติว่า "johndoe" คือ token ที่ถูกต้อง
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"username": "johndoe", "email": "[email protected]"}

@app.get("/users/me/")
async def read_current_user(current_user: dict = Depends(get_current_user)):
    """
    Endpoint สำหรับดึงข้อมูลผู้ใช้งานปัจจุบัน โดยต้องมี token ที่ถูกต้อง
    """
    return current_user

คำอธิบาย:

  • async def get_current_user(token: str):: นี่คือฟังก์ชัน Dependency ของเรา โดยจะรับ token เป็น Query Parameter
  • raise HTTPException(...): ถ้า token ไม่ถูกต้อง ฟังก์ชันจะยกข้อผิดพลาด HTTPException ซึ่ง FastAPI จะจับและส่ง HTTP 401 Unauthorized กลับไป
  • current_user: dict = Depends(get_current_user): ในฟังก์ชัน read_current_user เราประกาศว่าต้องการ current_user ซึ่งเป็นผลลัพธ์จาก Dependency get_current_user

เมื่อคุณเรียก http://127.0.0.1:8000/users/me/?token=johndoe คุณจะได้รับข้อมูลผู้ใช้งาน แต่ถ้าใช้ token อื่น คุณจะได้รับ HTTP 401 ครับ

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

FastAPI มีระบบจัดการข้อผิดพลาดที่มีประสิทธิภาพ คุณสามารถยก HTTPException เพื่อส่งคืน HTTP status code และข้อความที่กำหนดเองได้

# main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import FastAPI, HTTPException, status

app = FastAPI()

# ... (Path Operations และ Models เดิม) ...

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

@app.get("/items/{item_name}")
async def read_item_by_name(item_name: str):
    if item_name not in items_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
            headers={"X-Error": "There goes my custom header"},
        )
    return items_db[item_name]

ในตัวอย่างนี้ หาก item_name ที่ร้องขอไม่พบใน items_db เราจะยก HTTPException พร้อม status_code=404 Not Found และข้อความ "Item not found" นอกจากนี้ยังสามารถเพิ่ม Headers ที่กำหนดเองได้ด้วยครับ

FastAPI ยังช่วยให้คุณสามารถกำหนด Custom Exception Handler สำหรับจัดการข้อผิดพลาดบางประเภทได้อย่างละเอียดอีกด้วยครับ

คุณสมบัติขั้นสูงของ FastAPI

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

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

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

ตัวอย่างการใช้ APIRouter:

  1. สร้างไฟล์ routers/items.py: 
    # routers/items.py
from fastapi import APIRouter

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

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

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

@router.get("/{item_name}")
async def read_item(item_name: str):
    if item_name not in fake_items_db:
        return {"error": "Item not found"} # ในความเป็นจริงควร raise HTTPException
    return fake_items_db[item_name]
  2. สร้างไฟล์ routers/users.py: 
    # routers/users.py
from fastapi import APIRouter

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

fake_users_db = [
    {"username": "john", "email": "[email protected]"},
    {"username": "jane", "email": "[email protected]"}
]

@router.get("/")
async def read_users():
    return fake_users_db

@router.get("/{username}")
async def read_user(username: str):
    for user in fake_users_db:
        if user["username"] == username:
            return user
    return {"error": "User not found"}
  3. แก้ไข main.py เพื่อรวม Router: 
    # main.py
from fastapi import FastAPI
from routers import items, users # นำเข้า routers ที่สร้างขึ้น

app = FastAPI()

# เพิ่ม routers เข้ามาในแอปพลิเคชัน
app.include_router(items.router)
app.include_router(users.router)

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

ตอนนี้คุณสามารถเข้าถึง /items/ และ /users/ ได้แยกกัน โค้ดของคุณก็จะดูสะอาดและจัดการง่ายขึ้นมากครับ

การรักษาความปลอดภัย (Authentication & Authorization)

FastAPI มีเครื่องมือช่วยในการจัดการการยืนยันตัวตน (Authentication) และการอนุญาต (Authorization) โดยมี OAuth2 with Password (and Bearer Token) เป็นสิ่งที่รองรับในตัว

แนวคิดพื้นฐาน:

  • OAuth2: เป็นโปรโตคอลมาตรฐานสำหรับการอนุญาต
  • Password Flow: ผู้ใช้ส่ง username/password เพื่อรับ Access Token
  • Bearer Token: Access Token ที่ได้รับ จะถูกส่งไปใน Header ของทุก Request (Authorization: Bearer YOUR_TOKEN)
  • Dependency Injection: ใช้ DI ในการตรวจสอบ Token และดึงข้อมูลผู้ใช้งาน

ตัวอย่าง (แบบย่อ):

# main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

# ตัวอย่าง User Model (สำหรับจำลอง)
class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

class UserInDB(User):
    hashed_password: str

# จำลองฐานข้อมูลผู้ใช้งาน
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": "fakehashedsecret", # ในความเป็นจริงต้อง hash ด้วย bcrypt
        "disabled": False,
    }
}

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

def verify_password(plain_password: str, hashed_password: str):
    # ในความเป็นจริงใช้ bcrypt.checkpw(plain_password.encode('utf-8'), hashed_password.encode('utf-8'))
    return plain_password + "hash" == hashed_password # จำลองการตรวจสอบรหัสผ่าน

# กำหนด OAuth2PasswordBearer ที่จะใช้
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = get_user(token) # สมมติว่า token คือ username
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user(form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # ในความเป็นจริงต้องสร้าง JWT Token
    access_token = user.username # จำลอง access token
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

โค้ดนี้แสดงแนวคิดพื้นฐานของการใช้ OAuth2 ใน FastAPI โดยมี Endpoint /token สำหรับเข้าสู่ระบบและรับ Token และ /users/me/ ที่ต้องใช้ Token เพื่อเข้าถึงครับ

หมายเหตุ: ในการใช้งานจริง คุณจะต้องใช้ไลบรารีสำหรับสร้าง JWT (JSON Web Token) และ hash รหัสผ่านด้วย Bcrypt หรือ Argon2 เพื่อความปลอดภัยที่แท้จริงครับ

การเชื่อมต่อฐานข้อมูลด้วย SQLAlchemy ORM

FastAPI ไม่ได้กำหนดว่าคุณต้องใช้ฐานข้อมูลประเภทใด หรือ ORM (Object-Relational Mapper) ตัวใด แต่ SQLAlchemy เป็น ORM ยอดนิยมใน Python ที่ทำงานร่วมกับ FastAPI ได้ดีเยี่ยม

ในตัวอย่างนี้ เราจะใช้ SQLite เป็นฐานข้อมูลแบบง่าย และ SQLAlchemy ORM ในการจัดการ

  1. ติดตั้ง SQLAlchemy และ Pydantic-SQLAlchemy (SQLModel):

    สำหรับ SQLModel ซึ่งเป็นไลบรารีที่สร้างโดยผู้สร้าง FastAPI เพื่อรวม Pydantic และ SQLAlchemy เข้าด้วยกันอย่างลงตัวครับ

    pip install sqlmodel
  2. ตั้งค่าฐานข้อมูลและ Model: 
    # database.py
from sqlmodel import Field, SQLModel, create_engine, Session
from typing import Optional

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session
  3. สร้าง CRUD Operations ใน main.py: 
    # main.py (เพิ่มโค้ดส่วนนี้)
from fastapi import FastAPI, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List

from database import Hero, create_db_and_tables, get_session # นำเข้าจากไฟล์ database.py

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()

@app.post("/heroes/", response_model=Hero)
async def create_hero(*, session: Session = Depends(get_session), hero: Hero):
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero

@app.get("/heroes/", response_model=List[Hero])
async def read_heroes(*, session: Session = Depends(get_session)):
    heroes = session.exec(select(Hero)).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=Hero)
async def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.put("/heroes/{hero_id}", response_model=Hero)
async def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: Hero):
    db_hero = session.get(Hero, hero_id)
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.dict(exclude_unset=True) # อัปเดตเฉพาะฟิลด์ที่มีการส่งมา
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    
    session.add(db_hero)
    session.commit()
    session.refresh(db_hero)
    return db_hero

@app.delete("/heroes/{hero_id}")
async def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    session.delete(hero)
    session.commit()
    return {"message": "Hero deleted successfully"}

    คำอธิบาย:

    • create_db_and_tables(): ถูกเรียกเมื่อแอปพลิเคชันเริ่มต้น (@app.on_event("startup")) เพื่อสร้างตารางในฐานข้อมูลหากยังไม่มี
    • get_session(): เป็น Dependency function ที่ใช้ในการสร้างและปิด Session ของฐานข้อมูล ช่วยให้จัดการกับ Session ได้อย่างถูกต้อง
    • CRUD Operations (Create, Read, Update, Delete) ถูกสร้างขึ้นโดยใช้ SQLAlchemy functions และ Pydantic Models เพื่อจัดการข้อมูล

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

    การทดสอบเป็นส่วนสำคัญของการพัฒนาซอฟต์แวร์ FastAPI มี TestClient ที่ช่วยให้คุณสามารถทดสอบ API ของคุณได้อย่างง่ายดายและรวดเร็ว

    1. ติดตั้ง pytest: 
      pip install pytest
    2. สร้างไฟล์ test_main.py: 
      # test_main.py
from fastapi.testclient import TestClient
from main import app # นำเข้า app instance ของคุณ

client = TestClient(app)

def test_read_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, FastAPI World!"}

def test_read_item():
    response = client.get("/items/5")
    assert response.status_code == 200
    assert response.json() == {"item_id": 5}

def test_create_item():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "A test item", "price": 10.0, "tax": 1.0},
    )
    assert response.status_code == 200
    assert response.json()["name"] == "Test Item"
    assert response.json()["price_with_tax"] == 11.0 # จาก logic ใน create_item

def test_create_item_invalid_price():
    response = client.post(
        "/items/",
        json={"name": "Test Item", "description": "A test item", "price": "invalid", "tax": 1.0},
    )
    assert response.status_code == 422 # Unprocessable Entity สำหรับ validation error
    3. รันการทดสอบ: 
      pytest

    TestClient ทำงานเหมือนกับการส่ง Request HTTP จริง ๆ แต่ทำงานในหน่วยความจำ โดยไม่ต้องรันเซิร์ฟเวอร์ Uvicorn แยกต่างหากครับ

    ข้อควรพิจารณาในการ Deploy API

    เมื่อ API ของคุณพร้อมสำหรับการใช้งานจริง มีหลายสิ่งที่คุณควรพิจารณาในการ Deploy:

    • ASGI Server: ใช้ Uvicorn ในโหมด Production (ไม่ใช้ --reload) หรือใช้ Gunicorn (ซึ่งสามารถจัดการ Worker Processes ได้) ร่วมกับ Uvicorn worker class เช่น: 
      gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
    • Reverse Proxy: ใช้ Nginx หรือ Apache เป็น Reverse Proxy เพื่อจัดการ SSL/TLS, Load Balancing, และการเสิร์ฟ Static Files
    • Dockerization: การใช้ Docker เป็นวิธีที่นิยมในการแพ็คเกจแอปพลิเคชันและ Dependencies ทั้งหมดเข้าด้วยกัน ทำให้ Deploy ได้ง่ายและสอดคล้องกันในทุกสภาพแวดล้อม
    • Environmental Variables: เก็บการตั้งค่าที่ละเอียดอ่อน เช่น Database Credentials, API Keys ใน Environmental Variables แทนที่จะ hardcode ไว้ในโค้ด
    • Monitoring & Logging: ตั้งค่าระบบ Monitoring และ Logging เพื่อตรวจสอบประสิทธิภาพและข้อผิดพลาดของ API

    สร้าง REST API สำหรับระบบจัดการงาน (Task Management) แบบครบวงจร

    มาถึงส่วนที่เราจะนำความรู้ที่เรียนมาทั้งหมดมาสร้าง REST API สำหรับระบบจัดการงาน (Task Management) อย่างง่ายกันครับ

    ฟังก์ชันการทำงาน:

    • สร้าง, อ่าน, อัปเดต, ลบงาน (CRUD)
    • เก็บงานในฐานข้อมูล SQLite
    • มีระบบยืนยันตัวตนแบบง่าย ๆ (Basic Authentication)

    การตั้งค่าโปรเจกต์

    สร้างโครงสร้างโฟลเดอร์ดังนี้:

    task_api/
├── main.py
├── database.py
├── routers/
│   └── tasks.py
├── models.py
├── schemas.py
├── auth.py
└── requirements.txt

    requirements.txt:

    fastapi
uvicorn[standard]
sqlmodel
passlib[bcrypt] # สำหรับ hash รหัสผ่าน
python-jose[cryptography] # สำหรับ JWT (ใช้ใน auth.py)

    สร้าง Virtual Environment และติดตั้ง Dependencies:

    python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

    สร้าง Pydantic Models สำหรับ Task

    เราจะสร้าง Pydantic Models ในไฟล์ schemas.py เพื่อใช้ในการรับและส่งข้อมูล

    schemas.py:

    # schemas.py
from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class TaskBase(BaseModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class TaskCreate(TaskBase):
    pass # ไม่ต้องเพิ่มอะไรเป็นพิเศษสำหรับการสร้าง

class TaskUpdate(TaskBase):
    title: Optional[str] = None
    completed: Optional[bool] = None

class Task(TaskBase): # Task model ที่มี ID และ created_at
    id: int
    owner_id: int # เพิ่ม field สำหรับเจ้าของงาน
    created_at: datetime

    class Config:
        orm_mode = True # สำคัญสำหรับ SQLModel/SQLAlchemy เพื่อให้สามารถแปลงจาก ORM object เป็น Pydantic ได้

    ตั้งค่าฐานข้อมูลและ SQLAlchemy Models

    เราจะใช้ SQLModel เพื่อรวม Pydantic และ SQLAlchemy เข้าด้วยกัน

    models.py: (นี่คือ SQLModel models ที่จะใช้สร้างตารางในฐานข้อมูล)

    # models.py
from sqlmodel import Field, SQLModel, Relationship
from typing import Optional, List
from datetime import datetime

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(unique=True, index=True)
    hashed_password: str
    email: Optional[str] = Field(default=None, unique=True, index=True)
    is_active: bool = Field(default=True)

    tasks: List["TaskDB"] = Relationship(back_populates="owner") # One-to-Many relationship

class TaskDB(SQLModel, table=True): # ตั้งชื่อเป็น TaskDB เพื่อไม่ให้ชนกับ Pydantic Task
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)

    owner_id: Optional[int] = Field(default=None, foreign_key="user.id")
    owner: Optional[User] = Relationship(back_populates="tasks")

    database.py: (สำหรับจัดการการเชื่อมต่อฐานข้อมูลและ Session)

    # database.py
from sqlmodel import create_engine, Session, SQLModel
from models import User, TaskDB # นำเข้า SQLModel ของเรา

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

engine = create_engine(sqlite_url, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

def get_session():
    with Session(engine) as session:
        yield session

    เพิ่มระบบยืนยันตัวตน (Basic Authentication)

    เราจะใช้ JWT (JSON Web Token) สำหรับการยืนยันตัวตน

    auth.py:

    # auth.py
from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

from sqlmodel import Session, select
from database import get_session
from models import User

# การตั้งค่าสำหรับ JWT
SECRET_KEY = "super-secret-key" # ในโปรดักชันควรเก็บใน env var
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme), session: Session = Depends(get_session)):
    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
    except JWTError:
        raise credentials_exception
    
    user = session.exec(select(User).where(User.username == username)).first()
    if user is None:
        raise credentials_exception
    return user

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

    สร้าง CRUD Operations สำหรับ Task

    เราจะสร้าง Router สำหรับ Tasks และ User Authentication

    routers/tasks.py:

    # routers/tasks.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from typing import List

from database import get_session
from models import TaskDB, User # ใช้ TaskDB model
from schemas import Task, TaskCreate, TaskUpdate # ใช้ Pydantic schemas
from auth import get_current_active_user

router = APIRouter(
    prefix="/tasks",
    tags=["tasks"],
    dependencies=[Depends(get_current_active_user)], # ทุก endpoint ใน router นี้ต้องการการยืนยันตัวตน
    responses={401: {"description": "Unauthorized"}},
)

@router.post("/", response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task(
    *,
    session: Session = Depends(get_session),
    task_in: TaskCreate,
    current_user: User = Depends(get_current_active_user)
):
    """
    สร้างงานใหม่
    """
    db_task = TaskDB.from_orm(task_in, update={'owner_id': current_user.id}) # สร้าง TaskDB จาก TaskCreate
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task

@router.get("/", response_model=List[Task])
async def read_tasks(
    *,
    session: Session = Depends(get_session),
    offset: int = 0,
    limit: int = 100,
    current_user: User = Depends(get_current_active_user)
):
    """
    ดึงรายการงานทั้งหมดของผู้ใช้งานปัจจุบัน
    """
    tasks = session.exec(
        select(TaskDB).where(TaskDB.owner_id == current_user.id).offset(offset).limit(limit)
    ).all()
    return tasks

@router.get("/{task_id}", response_model=Task)
async def read_task(
    *,
    session: Session = Depends(get_session),
    task_id: int,
    current_user: User = Depends(get_current_active_user)
):
    """
    ดึงงานเดี่ยวตาม ID
    """
    task = session.exec(
        select(TaskDB).where(TaskDB.id == task_id, TaskDB.owner_id == current_user.id)
    ).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found or you don't have access to it")
    return task

@router.put("/{task_id}", response_model=Task)
async def update_task(
    *,
    session: Session = Depends(get_session),
    task_id: int,
    task_in: TaskUpdate,
    current_user: User = Depends(get_current_active_user)
):
    """
    อัปเดตงาน
    """
    db_task = session.exec(
        select(TaskDB).where(TaskDB.id == task_id, TaskDB.owner_id == current_user.id)
    ).first()
    if not db_task:
        raise HTTPException(status_code=404, detail="Task not found or you don't have access to it")
    
    task_data = task_in.dict(exclude_unset=True)
    for key, value in task_data.items():
        setattr(db_task, key, value)
    
    session.add(db_task)
    session.commit()
    session.refresh(db_task)
    return db_task

@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    *,
    session: Session = Depends(get_session),
    task_id: int,
    current_user: User = Depends(get_current_active_user)
):
    """
    ลบงาน
    """
    task = session.exec(
        select(TaskDB).where(TaskDB.id == task_id, TaskDB.owner_id == current_user.id)
    ).first()
    if not task:
        raise HTTPException(status_code=404, detail="Task not found or you don't have access to it")
    
    session.delete(task)
    session.commit()
    return {"message": "Task deleted successfully"} # ในกรณี 204 ควรส่งคืน empty body แต่ส่ง message เพื่อความชัดเจนในการทดสอบ

    routers/users.py: (สำหรับลงทะเบียนผู้ใช้และเข้าสู่ระบบ)

    # routers/users.py
    
from fastapi import APIRouter, Depends, HTTPException, status
    
from fastapi.security import OAuth2PasswordRequestForm
    
from sqlmodel import Session, select
    
from datetime import timedelta
    

    from database import get_session
    
from models import User
    
from schemas import Task
    
from auth import (
    
    ACCESS_TOKEN_EXPIRE_MINUTES,
    
    create_access_token,
    
    get_password_hash,
    
    verify_password,
    
    get_current_active_user,
    
    	

    บทความที่เกี่ยวข้อง
    SiamCafe.net — ชุมชน IT ที่ใหญ่ที่สุด · Siam2R.com — Portfolio งาน IT

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

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

Facebook | YouTube | LINE OA

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

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

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