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

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

March 16, 2026
0 Views

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

สารบัญ

บทนำ: ทำไมต้อง FastAPI สำหรับ REST API ของคุณ?

ก่อนที่เราจะลงมือสร้าง API ด้วย FastAPI เรามาทำความเข้าใจพื้นฐานเกี่ยวกับ REST API และเหตุผลว่าทำไม FastAPI ถึงเป็นตัวเลือกที่ยอดเยี่ยมสำหรับการสร้าง API ในยุคปัจจุบันกันก่อนครับ

REST API คืออะไร?

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

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

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

REST API ใช้ HTTP methods (GET, POST, PUT, DELETE) ในการดำเนินการกับทรัพยากรต่าง ๆ เช่น การดึงข้อมูล (GET), การสร้างข้อมูลใหม่ (POST), การอัปเดตข้อมูล (PUT), และการลบข้อมูล (DELETE) ครับ

FastAPI คืออะไร? จุดเด่นที่ทำให้เหนือกว่า

FastAPI คือเว็บเฟรมเวิร์กสำหรับสร้าง API ด้วย Python ที่มาพร้อมกับประสิทธิภาพสูง ใช้งานง่าย และมีฟีเจอร์มากมายที่ช่วยให้นักพัฒนาสร้าง API ได้รวดเร็วและมีคุณภาพ โดยอาศัย Python Type Hints และ Pydantic ในการจัดการข้อมูล มาดูกันครับว่าจุดเด่นของ FastAPI มีอะไรบ้าง

  • ความเร็วและประสิทธิภาพ: FastAPI ถูกสร้างขึ้นบนพื้นฐานของ Starlette (สำหรับเว็บพาร์ท) และ Pydantic (สำหรับการจัดการข้อมูล) ซึ่งเป็นไลบรารีที่มีประสิทธิภาพสูง ทำให้ FastAPI สามารถประมวลผลคำขอได้รวดเร็วเทียบเท่ากับ Node.js และ Go เลยทีเดียวครับ
  • เอกสาร API อัตโนมัติ (Swagger UI/ReDoc): จุดเด่นที่นักพัฒนาชื่นชอบมากที่สุดคือ FastAPI สามารถสร้างเอกสาร API แบบอินเทอร์แอคทีฟ (Interactive API Documentation) ให้โดยอัตโนมัติ ไม่ว่าจะเป็น Swagger UI (OpenAPI) หรือ ReDoc ซึ่งช่วยให้การทดสอบและทำความเข้าใจ API ทำได้ง่ายขึ้นมาก โดยไม่ต้องเขียนเอกสารเพิ่มเองเลยครับ
  • การตรวจสอบข้อมูลที่ทรงพลังด้วย Pydantic: FastAPI ใช้ Pydantic ในการกำหนด Schema และตรวจสอบความถูกต้องของข้อมูล (Data Validation) ทั้งในส่วนของ Request Body, Query Parameters, Path Parameters และ Response Models ซึ่งช่วยลดข้อผิดพลาดและทำให้โค้ดมีความน่าเชื่อถือมากขึ้นครับ
  • ระบบ Dependency Injection ที่ยืดหยุ่น: FastAPI มีระบบ Dependency Injection ที่ใช้งานง่ายและทรงพลัง ช่วยให้คุณสามารถจัดการกับการพึ่งพาของฟังก์ชันต่าง ๆ ได้อย่างมีระเบียบ เช่น การเชื่อมต่อฐานข้อมูล, การตรวจสอบสิทธิ์ผู้ใช้ หรือการดึงค่า Configuration ครับ
  • รองรับ Asynchronous Programming (async/await): FastAPI ถูกออกแบบมาให้รองรับ Asynchronous Programming โดยใช้ async และ await ของ Python ทำให้สามารถจัดการกับการร้องขอจำนวนมากพร้อมกันได้อย่างมีประสิทธิภาพ เหมาะสำหรับการสร้าง API ที่ต้องการความเร็วสูงและรองรับผู้ใช้งานจำนวนมากครับ
  • คอมมูนิตี้ที่เติบโตเร็ว: แม้จะเป็นเฟรมเวิร์กที่ค่อนข้างใหม่ แต่ FastAPI มีคอมมูนิตี้ที่เติบโตอย่างรวดเร็ว มีเอกสารประกอบที่ชัดเจน และมีตัวอย่างการใช้งานมากมาย ทำให้ง่ายต่อการเรียนรู้และขอความช่วยเหลือครับ

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

เตรียมความพร้อม: ติดตั้งและตั้งค่าสภาพแวดล้อม

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

ข้อกำหนดเบื้องต้น

  • Python 3.7+ (แนะนำ Python 3.9 หรือสูงกว่า)
  • pip (ตัวจัดการแพ็กเกจของ Python)

การติดตั้ง Python และ pip

หากคุณยังไม่มี Python บนเครื่อง ให้ดาวน์โหลดและติดตั้งจากเว็บไซต์ทางการของ Python (python.org/downloads/) โดยอย่าลืมเลือกช่อง “Add Python to PATH” ระหว่างการติดตั้ง เพื่อให้สามารถเรียกใช้งาน Python จาก Command Line ได้ง่ายขึ้นครับ

หลังจากติดตั้งแล้ว คุณสามารถตรวจสอบเวอร์ชันของ Python และ pip ได้โดยใช้คำสั่ง:

python --version
pip --version

การสร้าง Virtual Environment

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

สร้าง Virtual Environment:

python -m venv venv

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

  • บน Windows:
    • .\venv\Scripts\activate
  • บน macOS/Linux:
    • source venv/bin/activate

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

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

FastAPI ต้องการ ASGI server เช่น Uvicorn ในการรันแอปพลิเคชันครับ เราจะติดตั้งทั้งสองอย่างพร้อมกัน

pip install fastapi uvicorn[standard]

uvicorn[standard] จะติดตั้ง Uvicorn พร้อมกับแพ็กเกจเสริมที่จำเป็น เช่น python-dotenv และ watchgod ซึ่งมีประโยชน์สำหรับการพัฒนาครับ

ถ้าคุณต้องการติดตั้งเฉพาะ Uvicorn แบบพื้นฐาน สามารถใช้ pip install uvicorn ได้ครับ

โครงสร้างพื้นฐานของ FastAPI API

เมื่อเตรียมพร้อมแล้ว เรามาเริ่มสร้าง API แรกของเรากันเลยครับ

Hello World ด้วย FastAPI

สร้างไฟล์ชื่อ main.py และเพิ่มโค้ดต่อไปนี้:

# main.py
from fastapi import FastAPI

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

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

# กำหนด Path Operation Decorator สำหรับ HTTP GET method ที่ Path "/items/{item_id}"
@app.get("/items/{item_id}")
async def read_item(item_id: int, query_param: str = None):
    """
    Endpoint สำหรับดึงข้อมูล Item ด้วย ID
    รับ Path Parameter: item_id (int)
    รับ Query Parameter: query_param (str, optional)
    """
    if query_param:
        return {"item_id": item_id, "query_param": query_param}
    return {"item_id": item_id}

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

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

การรันแอปพลิเคชัน

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

uvicorn main:app --reload

คำอธิบาย:

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

คุณจะเห็นข้อความประมาณว่า “Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)”

การเข้าถึงเอกสาร API

เมื่อแอปพลิเคชันทำงานอยู่ คุณสามารถเข้าถึงเอกสาร API ที่ FastAPI สร้างให้โดยอัตโนมัติได้ที่:

ลองเข้าดูนะครับ คุณจะเห็นรายละเอียดของ API ที่เราสร้างไปเมื่อสักครู่ พร้อมปุ่มให้ทดสอบเรียก API ได้ทันที!

HTTP Methods พื้นฐาน: GET, POST, PUT, DELETE

REST API ใช้ HTTP Methods ในการบ่งบอกประเภทของการดำเนินการกับทรัพยากรครับ

  • GET: ใช้สำหรับดึงข้อมูล (Read) จาก Server โดยไม่เปลี่ยนแปลงสถานะของ Server
  • POST: ใช้สำหรับสร้างข้อมูลใหม่ (Create) บน Server โดยส่งข้อมูลไปใน Request Body
  • PUT: ใช้สำหรับอัปเดตข้อมูลที่มีอยู่แล้วทั้งหมด (Update/Replace) บน Server โดยส่งข้อมูลใหม่ทั้งหมดไปแทนที่
  • DELETE: ใช้สำหรับลบข้อมูล (Delete) จาก Server
  • PATCH: ใช้สำหรับอัปเดตข้อมูลบางส่วน (Update/Modify) บน Server

FastAPI มี Path Operation Decorator สำหรับแต่ละ HTTP Method เช่น @app.get(), @app.post(), @app.put(), @app.delete(), @app.patch()

การจัดการ Path Parameters และ Query Parameters

API ส่วนใหญ่มักจะต้องรับค่าจาก Client เพื่อระบุทรัพยากรหรือกรองข้อมูลครับ FastAPI มีวิธีจัดการกับ Path Parameters และ Query Parameters ได้อย่างง่ายดาย

Path Parameters

Path Parameters คือค่าที่ฝังอยู่ใน URL Path เพื่อระบุทรัพยากรเฉพาะ เช่น /items/1 เพื่อดึง item ที่มี ID เป็น 1

ตัวอย่าง:

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

app = FastAPI()

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

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """
    ดึงข้อมูลผู้ใช้ด้วย ID
    """
    return {"user_id": user_id, "message": f"Fetching user with ID: {user_id}"}

@app.get("/files/{file_path:path}")
async def get_file(file_path: str):
    """
    ดึงข้อมูลไฟล์ด้วย Path (ตัวอย่างการใช้ Path Parameter ที่มี "/" ได้)
    """
    return {"file_path": file_path, "message": f"Fetching file from path: {file_path}"}

ในตัวอย่าง /users/{user_id}, user_id จะถูกจับคู่จาก URL และส่งเป็นอาร์กิวเมนต์ไปยังฟังก์ชัน get_user ครับ

สำหรับ /files/{file_path:path}, :path เป็นการบอก FastAPI ว่า Path Parameter นี้สามารถมีเครื่องหมาย / ได้ ทำให้เหมาะสำหรับการส่ง Path ของไฟล์เข้ามาครับ

Query Parameters

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

ตัวอย่าง:

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI, Query

app = FastAPI()

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

@app.get("/products/")
async def get_products(skip: int = 0, limit: int = 10, search: str = None):
    """
    ดึงข้อมูลสินค้าพร้อมกับการ Pagination และ Search
    - skip: จำนวนรายการที่จะข้ามไป (default: 0)
    - limit: จำนวนรายการที่จะดึง (default: 10)
    - search: คำค้นหา (optional)
    """
    products = [{"id": i, "name": f"Product {i}"} for i in range(1, 101)]
    
    if search:
        products = [p for p in products if search.lower() in p["name"].lower()]

    return products[skip : skip + limit]

@app.get("/items_with_validation/")
async def read_items_with_validation(
    q: str = Query(
        None, 
        min_length=3, 
        max_length=50, 
        regex="^fast.*", 
        description="Query string for items",
        alias="item-query"
    )
):
    """
    ตัวอย่าง Query Parameter พร้อมการ Validation และ Metadata
    """
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

ในตัวอย่าง get_products, skip และ limit จะถูกกำหนดค่า Default เป็น 0 และ 10 ตามลำดับ ทำให้เป็น Optional Query Parameters ครับ search ก็เป็น Optional เช่นกัน

สำหรับ read_items_with_validation, เราใช้ฟังก์ชัน Query() จาก fastapi เพื่อเพิ่มการ Validation และ Metadata ให้กับ Query Parameter ครับ เช่น min_length, max_length, regex และ description ที่จะไปปรากฏในเอกสาร API โดยอัตโนมัติ นอกจากนี้ยังสามารถกำหนด alias เพื่อให้ชื่อ Parameter ใน URL แตกต่างจากชื่อตัวแปรในโค้ดได้ด้วยครับ

Type Hinting และการตรวจสอบข้อมูล

คุณคงสังเกตเห็นว่าเราใช้ Type Hinting (เช่น user_id: int, skip: int) อย่างสม่ำเสมอในโค้ด FastAPI ซึ่งเป็นหัวใจสำคัญในการทำงานของ FastAPI เลยครับ

ประโยชน์ของ Type Hinting ใน FastAPI:

  • Data Validation: FastAPI ใช้ Type Hinting ร่วมกับ Pydantic ในการตรวจสอบความถูกต้องของข้อมูลที่ส่งเข้ามา หากข้อมูลไม่ตรงตาม Type ที่กำหนด จะส่ง HTTP 422 Unprocessable Entity กลับไปพร้อมรายละเอียดข้อผิดพลาด
  • Data Conversion: FastAPI จะแปลงข้อมูลที่รับเข้ามาให้เป็น Type ที่ถูกต้องโดยอัตโนมัติ เช่น หากรับ "123" มาสำหรับ item_id: int ก็จะแปลงเป็น Integer 123 ให้ครับ
  • IDE Support: Type Hinting ช่วยให้ IDE (เช่น VS Code, PyCharm) สามารถให้คำแนะนำโค้ด (Autocompletion), ตรวจจับข้อผิดพลาด และ Refactor โค้ดได้ดีขึ้น
  • API Documentation: ข้อมูล Type Hinting จะถูกนำไปใช้ในการสร้างเอกสาร API (Swagger UI/ReDoc) ให้โดยอัตโนมัติ ทำให้ Client ทราบว่าต้องส่งข้อมูลรูปแบบใด

คุณสามารถใช้ Type Hinting ได้กับ Python Types มาตรฐาน (str, int, float, bool, list, dict) และ Types จากโมดูล typing (เช่น Optional, List, Dict) ครับ

Pydantic Model: การสร้าง Request Body และ Response Model

เมื่อเราต้องการส่งข้อมูลที่ซับซ้อน เช่น JSON Object ไปพร้อมกับ Request หรือกำหนดโครงสร้างของข้อมูลที่จะส่งกลับไป Pydantic Model จะเข้ามามีบทบาทสำคัญครับ

ทำไมต้องใช้ Pydantic?

Pydantic คือไลบรารีสำหรับ Data Validation และ Settings Management โดยใช้ Python Type Hinting เป็นหลักครับ FastAPI ใช้ Pydantic เป็นแกนหลักในการจัดการกับ Request Body และ Response Model ด้วยเหตุผลดังนี้:

  • Validation อัตโนมัติ: Pydantic ตรวจสอบความถูกต้องของข้อมูลตาม Type Hinting ที่กำหนดไว้ หากข้อมูลไม่ถูกต้อง จะส่งคืนข้อผิดพลาดโดยละเอียด
  • Serialization/Deserialization: Pydantic สามารถแปลงข้อมูล JSON เป็น Python Object (Deserialization) และแปลง Python Object เป็น JSON (Serialization) ได้อย่างราบรื่น
  • Autocompletion: การกำหนด Model ช่วยให้ IDE สามารถให้ Autocompletion ได้เมื่อคุณทำงานกับข้อมูลเหล่านั้น
  • Schema Generation: Pydantic Model สามารถแปลงเป็น JSON Schema ได้ ซึ่ง FastAPI นำไปใช้ในการสร้างเอกสาร API

การสร้าง Pydantic Model

เราจะสร้าง Pydantic Model โดยการสืบทอดจาก BaseModel ของ Pydantic ครับ

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI
from pydantic import BaseModel, Field # เพิ่ม Field เพื่อ Validation ที่ซับซ้อนขึ้น
from typing import Optional, List

app = FastAPI()

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

# Pydantic Model สำหรับ Item
class Item(BaseModel):
    name: str = Field(..., example="Latest Smartphone", description="ชื่อของสินค้า")
    description: Optional[str] = Field(None, example="รุ่นล่าสุดพร้อมกล้อง 108MP", description="รายละเอียดสินค้า")
    price: float = Field(..., gt=0, description="ราคาสินค้า ต้องมากกว่า 0")
    tax: Optional[float] = Field(None, example=0.1, description="ภาษีสินค้า (ถ้ามี)")
    tags: List[str] = Field([], example=["electronics", "mobile"], description="หมวดหมู่สินค้า")

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

ใน Item Model:

  • name: str = Field(..., example="...", description="..."): ... หมายถึง Required Field. example และ description จะไปปรากฏในเอกสาร API
  • description: Optional[str] = None: Field นี้เป็น Optional และมีค่า Default เป็น None
  • price: float = Field(..., gt=0): gt=0 เป็น Pydantic Validation ที่ระบุว่าค่าของ price ต้องมากกว่า 0
  • tags: List[str] = []: ระบุว่าเป็น List ของ String และมีค่า Default เป็น List ว่าง

การใช้งาน Pydantic ใน POST Request

เมื่อต้องการรับข้อมูลใน Request Body เช่น เพื่อสร้าง Item ใหม่ เราจะใช้ Pydantic Model เป็น Type Hint ของอาร์กิวเมนต์ใน Path Operation Function ครับ

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, List

app = FastAPI()

# ... (โค้ด Item และ User Model) ...

@app.post("/items/", response_model=Item, status_code=201)
async def create_item(item: Item):
    """
    สร้าง Item ใหม่
    """
    # ในโลกจริง คุณอาจจะบันทึก item ลงฐานข้อมูลที่นี่
    # และคืนค่า item พร้อม ID ที่สร้างขึ้นมา
    item_dict = item.dict()
    item_dict.update({"id": len(fake_items_db) + 1}) # สมมติ ID
    fake_items_db.append(item_dict)
    return item_dict

# สมมติฐานข้อมูลในหน่วยความจำ
fake_items_db = []

คำอธิบาย:

  • @app.post("/items/", response_model=Item, status_code=201):
    • @app.post(): ระบุว่าเป็น HTTP POST request
    • response_model=Item: บอก FastAPI ว่า Response ที่ส่งกลับไปควรมีโครงสร้างตาม Item Model สิ่งนี้ช่วยในการ Serialization และสร้างเอกสาร API
    • status_code=201: กำหนด HTTP Status Code เป็น 201 Created สำหรับการสร้างทรัพยากรสำเร็จ
  • async def create_item(item: Item):: FastAPI จะอ่าน Request Body, ตรวจสอบความถูกต้องตาม Item Model, และสร้าง Instance ของ Item Model ให้คุณโดยอัตโนมัติ

หากคุณส่ง Request Body ที่ไม่ตรงตาม Item Model (เช่น price เป็น Negative หรือ name หายไป) FastAPI จะส่ง HTTP 422 Unprocessable Entity กลับมาพร้อมข้อผิดพลาดที่ชัดเจนครับ

การใช้งาน Pydantic สำหรับ Response Model

การใช้ response_model ใน Path Operation Decorator ไม่เพียงแต่ช่วยในการสร้างเอกสาร API เท่านั้น แต่ยังช่วยในการ Filtering และ Serialization ของข้อมูลที่จะส่งกลับไปด้วยครับ

ตัวอย่าง: การซ่อนข้อมูลบางอย่างใน Response

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, List

app = FastAPI()

# ... (โค้ด Item และ User Model) ...

# Pydantic Model สำหรับ User ที่จะใช้ใน Response (ซ่อน hashed_password)
class UserInDB(User):
    hashed_password: str

class UserResponse(BaseModel):
    username: str
    email: str
    full_name: Optional[str] = None

# สมมติฐานข้อมูลผู้ใช้
fake_users_db = {
    "john_doe": {
        "username": "john_doe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": "thisisasecretpasswordhashed"
    },
    "jane_smith": {
        "username": "jane_smith",
        "email": "[email protected]",
        "full_name": "Jane Smith",
        "hashed_password": "anothersecretpasswordhashed"
    }
}

@app.get("/users/{username}", response_model=UserResponse)
async def get_user_data(username: str):
    """
    ดึงข้อมูลผู้ใช้ (โดยซ่อนรหัสผ่าน)
    """
    user_data = fake_users_db.get(username)
    if user_data:
        return user_data
    return {"message": "User not found"}

ในตัวอย่างนี้ แม้ว่า fake_users_db จะมี hashed_password อยู่ แต่เนื่องจากเรากำหนด response_model=UserResponse ซึ่ง UserResponse ไม่มี Field hashed_password, FastAPI จะทำการกรอง Field hashed_password ออกไปจาก Response โดยอัตโนมัติ ทำให้ Client ไม่เห็นข้อมูล sensitive นี้ครับ

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

Dependency Injection: จัดการกับการพึ่งพา

Dependency Injection (DI) เป็น Pattern การออกแบบที่ช่วยให้โค้ดของคุณมีความยืดหยุ่น, ทดสอบง่าย, และดูแลรักษาง่ายขึ้นครับ FastAPI มีระบบ DI ที่ยอดเยี่ยมและใช้งานง่าย

แนวคิดของ Dependency Injection

แนวคิดหลักคือแทนที่จะให้ฟังก์ชันหรือคลาสสร้างหรือหา “สิ่งที่พึ่งพา” (Dependencies) ด้วยตัวเอง เราจะให้ “สิ่งที่พึ่งพา” เหล่านั้นถูก “ฉีด” (Injected) เข้าไปในฟังก์ชันหรือคลาสจากภายนอกครับ

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

สมมติว่าคุณมีฟังก์ชันที่ต้องการค่าจาก Header ของ HTTP Request หรือต้องการ Database Connection ฟังก์ชันนั้นไม่จำเป็นต้องเขียนโค้ดเพื่อดึงค่าเหล่านั้นด้วยตัวเอง แต่สามารถ “ขอ” ให้ FastAPI จัดการ “ฉีด” ค่าเหล่านั้นเข้ามาให้เมื่อฟังก์ชันถูกเรียกใช้ครับ

ตัวอย่างการใช้งาน Dependency Injection

เรามาสร้าง Dependency ง่าย ๆ ที่รับค่าจาก Header กันครับ

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI, Header, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional, List

app = FastAPI()

# ... (โค้ด Item และ User Model) ...
# ... (โค้ด fake_items_db และ fake_users_db) ...

# Dependency Function สำหรับตรวจสอบ X-Token Header
async def get_x_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")
    return x_token

@app.get("/items_protected/", dependencies=[Depends(get_x_token)])
async def read_items_protected():
    """
    Endpoint ที่ได้รับการป้องกันด้วย X-Token Header
    """
    return fake_items_db

@app.get("/users_protected/", response_model=UserResponse)
async def read_users_protected(x_token: str = Depends(get_x_token)):
    """
    Endpoint ที่ได้รับการป้องกันด้วย X-Token Header และแสดงข้อมูลผู้ใช้
    """
    # x_token ถูกตรวจสอบและส่งค่ากลับมาแล้ว ไม่ต้องตรวจสอบซ้ำ
    return list(fake_users_db.values())

คำอธิบาย:

  • async def get_x_token(x_token: str = Header(...))::
    • นี่คือฟังก์ชัน Dependency ของเรา
    • x_token: str = Header(...): บอก FastAPI ว่าให้ดึงค่าจาก HTTP Header ชื่อ X-Token และเป็น Required (...)
    • หาก X-Token ไม่ถูกต้อง จะมีการยก HTTPException ซึ่ง FastAPI จะจัดการแปลงเป็น HTTP Response Code 400 ให้โดยอัตโนมัติ
  • @app.get("/items_protected/", dependencies=[Depends(get_x_token)]):
    • เราใช้ dependencies=[Depends(get_x_token)] เพื่อบอก FastAPI ว่าก่อนที่จะเรียก read_items_protected ให้รันฟังก์ชัน get_x_token ก่อน
    • หาก get_x_token ยก HTTPException, read_items_protected จะไม่ถูกเรียก
  • async def read_users_protected(x_token: str = Depends(get_x_token))::
    • อีกวิธีในการใช้ Dependency คือการส่งเป็น Argument ในฟังก์ชัน Path Operation โดยตรง
    • ค่าที่ get_x_token คืนกลับมา (ในกรณีที่ไม่มีข้อผิดพลาด) จะถูกส่งเข้ามาใน x_token ของฟังก์ชัน read_users_protected ครับ

Dependency Injection ทำให้โค้ดของคุณสะอาดขึ้น เนื่องจากฟังก์ชัน Path Operation ไม่ต้องกังวลว่าจะได้ค่า X-Token มาอย่างไร เพียงแค่ระบุว่าต้องการอะไร FastAPI ก็จะจัดหามาให้ครับ

การสร้าง Dependency สำหรับ Database Connection

หนึ่งในการใช้งาน Dependency Injection ที่พบบ่อยที่สุดคือการจัดการ Database Session ครับ เพื่อให้แต่ละ Request ได้รับ Database Session ที่เป็นอิสระและถูกปิดอย่างเหมาะสม

# database.py (ไฟล์ใหม่)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# ตั้งค่า Database URL สำหรับ SQLite
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# สร้าง SQLAlchemy Engine
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} # จำเป็นสำหรับ SQLite ใน FastAPI
)

# สร้าง SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# สร้าง Base class สำหรับ SQLAlchemy Models
Base = declarative_base()

# Dependency สำหรับการจัดการ Database Session
def get_db():
    db = SessionLocal()
    try:
        yield db # ส่ง db session ให้กับ Path Operation
    finally:
        db.close() # ปิด db session เมื่อ Path Operation ทำงานเสร็จ

ในไฟล์ main.py คุณสามารถใช้ get_db เป็น Dependency ได้ดังนี้:

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI, Depends
# ... (imports อื่นๆ) ...
from sqlalchemy.orm import Session
from .database import get_db, engine, Base # import จากไฟล์ database.py

# ... (โค้ด Item และ User Model) ...

# สร้างตารางในฐานข้อมูล (ถ้ายังไม่มี)
# Base.metadata.create_all(bind=engine) # จะทำในส่วนของการเชื่อมต่อฐานข้อมูลด้านล่าง

@app.get("/items_from_db/")
async def read_items_from_db(db: Session = Depends(get_db)):
    """
    ตัวอย่างการดึงข้อมูลจากฐานข้อมูลโดยใช้ Dependency Injection
    """
    # ในโลกจริง คุณจะใช้ db session ในการ query ข้อมูลจากฐานข้อมูล
    # เช่น db.query(models.Item).all()
    return {"message": "Database session injected successfully!", "db_type": str(type(db))}

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

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

API ส่วนใหญ่ต้องการการโต้ตอบกับฐานข้อมูลเพื่อจัดเก็บและดึงข้อมูล ในส่วนนี้ เราจะมาดูวิธีการเชื่อมต่อ FastAPI กับฐานข้อมูล โดยใช้ SQLAlchemy ORM เป็นตัวอย่างครับ

เลือกฐานข้อมูล: SQLite, PostgreSQL, MySQL, MongoDB

FastAPI ไม่ได้ผูกติดกับฐานข้อมูลใดเป็นพิเศษ คุณสามารถใช้ฐานข้อมูลใดก็ได้ที่คุณต้องการ:

  • Relational Databases:
    • SQLite: เหมาะสำหรับโปรเจกต์ขนาดเล็ก หรือใช้ใน Development Environment เพราะเป็นฐานข้อมูลแบบไฟล์ ไม่ต้องติดตั้ง Server เพิ่มเติม
    • PostgreSQL: เป็นตัวเลือกยอดนิยมสำหรับโปรเจกต์ขนาดใหญ่ มีความสามารถสูง และรองรับคุณสมบัติขั้นสูง
    • MySQL: ฐานข้อมูลยอดนิยมอีกตัว มีเครื่องมือและทรัพยากรมากมาย

    สำหรับฐานข้อมูลเชิงสัมพันธ์ มักจะใช้ร่วมกับ ORM (Object-Relational Mapper) เช่น SQLAlchemy หรือ Tortoise ORM ครับ

  • NoSQL Databases:
    • MongoDB: ฐานข้อมูลแบบ Document-oriented เหมาะสำหรับข้อมูลที่มีโครงสร้างยืดหยุ่น
    • Cassandra, Redis: ฐานข้อมูลเฉพาะทางอื่น ๆ

    สำหรับ NoSQL มักจะใช้ Driver เฉพาะของแต่ละฐานข้อมูล เช่น pymongo สำหรับ MongoDB

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

ตัวอย่างการใช้ SQLAlchemy ORM กับ SQLite

การติดตั้ง SQLAlchemy และ Pydantic

เราต้องติดตั้งไลบรารีที่จำเป็นครับ

pip install sqlalchemy "pydantic[email]" # [email] สำหรับ validation email

จากนั้น เราจะสร้างไฟล์ database.py และ models.py (สำหรับ SQLAlchemy Models) และ schemas.py (สำหรับ Pydantic Schemas) เพื่อจัดระเบียบโค้ดครับ

การสร้าง Base Model และ Session (ไฟล์: database.py)

สร้างไฟล์ sql_app/database.py (สร้างโฟลเดอร์ sql_app ก่อนนะครับ)

# sql_app/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # ไฟล์ฐานข้อมูลจะอยู่ที่ ./sql_app.db

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

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

การสร้าง Schema (Pydantic) และ Model (SQLAlchemy)

ไฟล์: sql_app/models.py (SQLAlchemy Models)

# sql_app/models.py
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship

from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

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

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

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

ไฟล์: sql_app/schemas.py (Pydantic Schemas)

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

# Base Schema สำหรับ Item
class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

# Schema สำหรับการสร้าง Item (มี title และ description)
class ItemCreate(ItemBase):
    pass

# Schema สำหรับ Item ที่อ่านจาก DB (มี id, owner_id และ is_active)
class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True # สำคัญ: บอก Pydantic ให้ทำงานกับ ORM objects ได้
# sql_app/schemas.py (ต่อ)
# Base Schema สำหรับ User
class UserBase(BaseModel):
    email: EmailStr # ใช้ EmailStr สำหรับ email validation

# Schema สำหรับการสร้าง User (มี email และ password)
class UserCreate(UserBase):
    password: str

# Schema สำหรับ User ที่อ่านจาก DB (มี id, is_active และ items)
class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = [] # User สามารถมี List ของ Items ได้

    class Config:
        orm_mode = True

CRUD Operations (ไฟล์: main.py)

ตอนนี้เราสามารถนำ Database Models และ Schemas มาใช้ใน main.py เพื่อสร้าง CRUD API ได้แล้วครับ

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

# Import โมดูลที่เราสร้างขึ้น
from .sql_app import models, schemas
from .sql_app.database import SessionLocal, engine, get_db

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

app = FastAPI()

# --- User CRUD Operations ---

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

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

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

# --- Item CRUD Operations ---

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    user = db.query(models.User).filter(models.User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")

    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

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

คำอธิบาย:

  • models.Base.metadata.create_all(bind=engine): บรรทัดนี้จะสร้างตารางในฐานข้อมูล sql_app.db ตามที่เรากำหนดไว้ใน models.py หากยังไม่มีตารางนั้น ๆ
  • db: Session = Depends(get_db): ในแต่ละ Path Operation Function เราจะใช้ Dependency get_db เพื่อรับ Database Session มาใช้งาน Session นี้จะถูกปิดโดยอัตโนมัติเมื่อ Request เสร็จสิ้น
  • db.query(models.User): ใช้ SQLAlchemy ORM ในการสร้าง Query
  • .filter(), .offset(), .limit(), .first(), .all(): เป็นเมธอดของ SQLAlchemy สำหรับกรอง, แบ่งหน้า, และดึงข้อมูล
  • db.add(db_user), db.commit(), db.refresh(db_user): เป็นขั้นตอนมาตรฐานในการบันทึกข้อมูลใหม่ลงฐานข้อมูลด้วย SQLAlchemy
  • response_model=schemas.User หรือ response_model=List[schemas.User]: FastAPI จะใช้ Pydantic Schema เหล่านี้ในการ Serialization ข้อมูลจาก SQLAlchemy Model ให้เป็น JSON Response โดยอัตโนมัติ และกรองข้อมูลที่ไม่ตรงกับ Schema ออกไป
  • class Config: orm_mode = True ใน Pydantic Schema (schemas.py) มีความสำคัญอย่างยิ่ง เพราะมันบอก Pydantic ให้สามารถอ่านข้อมูลจาก ORM Model ได้โดยตรง ไม่ใช่แค่จาก Dict หรือ JSON ครับ

ลองรัน uvicorn main:app --reload แล้วเข้าดู Swagger UI ที่ http://127.0.0.1:8000/docs คุณจะเห็น API สำหรับ User และ Item พร้อมให้ทดสอบ CRUD ได้เลยครับ

อ่านเพิ่มเติมเกี่ยวกับการใช้ SQLAlchemy กับ FastAPI

การตรวจสอบสิทธิ์และการอนุญาต (Authentication & Authorization)

API ที่ดีต้องมีระบบรักษาความปลอดภัย การตรวจสอบสิทธิ์ (Authentication) คือการยืนยันว่าผู้ใช้งานคือใคร ส่วนการอนุญาต (Authorization) คือการกำหนดว่าผู้ใช้งานนั้นมีสิทธิ์ทำอะไรได้บ้าง

แนวคิดเบื้องต้น

  • Authentication:
    • Basic Auth: ชื่อผู้ใช้และรหัสผ่านถูกส่งใน Header (ไม่ปลอดภัยหากไม่มี HTTPS)
    • API Key: Client ส่ง Key ลับไปกับทุก Request
    • OAuth2/JWT: Client ได้รับ Token (เช่น JWT) หลังจาก Login แล้วส่ง Token นั้นไปกับทุก Request เพื่อยืนยันตัวตน
  • Authorization:
    • Role-Based Access Control (RBAC): กำหนดบทบาท (เช่น Admin, User) ให้ผู้ใช้ และกำหนดสิทธิ์ตามบทบาท
    • Permissions-Based Access Control: กำหนดสิทธิ์เฉพาะ (เช่น ‘can_edit_item’, ‘can_delete_user’) ให้ผู้ใช้

FastAPI มีเครื่องมือและ Dependencies ในตัวสำหรับจัดการ OAuth2 ซึ่งเป็นมาตรฐานยอดนิยมสำหรับการตรวจสอบสิทธิ์ครับ

การใช้ OAuth2 with Password (JWT Token)

เราจะใช้ JWT (JSON Web Token) สำหรับการตรวจสอบสิทธิ์ OAuth2 ครับ

การติดตั้ง python-multipart และ python-jose[cryptography]

python-multipart จำเป็นสำหรับ Forms (เช่น Login Forms) และ python-jose[cryptography] สำหรับการสร้างและตรวจสอบ JWT

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

passlib[bcrypt] ใช้สำหรับ Hash รหัสผ่าน

โครงสร้างโค้ดสำหรับ Auth (ไฟล์: sql_app/auth.py และ main.py)

สร้างไฟล์ sql_app/auth.py:

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

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

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

from .database import get_db
from sqlalchemy.orm import Session
from . import models, schemas

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

# สำหรับ Hash รหัสผ่าน
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# สำหรับ OAuth2
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), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = schemas.TokenData(username=username)
    except JWTError:
        raise credentials_exception
    
    user = db.query(models.User).filter(models.User.email == token_data.username).first()
    if user is None:
        raise credentials_exception
    return user

ปรับปรุง sql_app/schemas.py เพิ่ม Token Schema:

# sql_app/schemas.py (เพิ่ม)
class Token(BaseModel):
    access_token: str
    token_type: str

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

ปรับปรุง sql_app/models.py เพื่อให้ User มี email เป็น `username` สำหรับ JWT (ถ้ายังไม่ได้ทำ):

# sql_app/models.py (ส่วนของ User model)
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True) # ใช้ email เป็น username
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

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

เพิ่ม Endpoint สำหรับ Login และ Protected Routes ใน main.py:

# main.py (เพิ่ม)
from .sql_app import auth # เพิ่ม import

# ... (โค้ดอื่นๆ) ...

@app.post("/token", response_model=schemas.Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.email == form_data.username).first()
    if not user or not auth.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"},
        )
    access_token_expires = timedelta(minutes=auth.ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = auth.create_access_token(
        data={"sub": user.email}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=schemas.User)
async def read_users_me(current_user: models.User = Depends(auth.get_current_user)):
    """
    Endpoint ที่ได้รับการป้องกัน: ดึงข้อมูลผู้ใช้งานปัจจุบัน
    """
    return current_user

@app.get("/items_protected_by_auth/", response_model=List[schemas.Item])
async def read_items_protected_by_auth(current_user: models.User = Depends(auth.get_current_user), db: Session = Depends(get_db)):
    """
    Endpoint ที่ได้รับการป้องกัน: ดึงข้อมูล Item ทั้งหมด
    """
    # ตรวจสอบสิทธิ์เพิ่มเติม ถ้า current_user.is_admin:
    # if not current_user.is_admin:
    #    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized")
    items = db.query(models.Item).all()
    return items

ขั้นตอนการทำงาน:

  1. Client ส่ง username (email) และ password ไปที่ /token (POST Request)
  2. login_for_access_token รับข้อมูลผ่าน OAuth2PasswordRequestForm, ตรวจสอบชื่อผู้ใช้/รหัสผ่านกับฐานข้อมูล
  3. หากถูกต้อง จะสร้าง JWT Access Token ด้วย auth.create_access_token และคืนกลับให้ Client
  4. Client จะเก็บ Access Token ไว้ และส่ง Token นี้ไปใน HTTP Header Authorization: Bearer <access_token> สำหรับทุก Request ไปยัง Protected Endpoints
  5. ใน Protected Endpoints (เช่น /users/me/), auth.get_current_user จะถูกเรียกเป็น Dependency
    • get_current_user ดึง Token จาก Header โดยใช้ oauth2_scheme
    • ถอดรหัส Token และตรวจสอบความถูกต้องของ JWT
    • ดึงข้อมูลผู้ใช้จากฐานข้อมูลด้วย username ที่ได้จาก Token
    • คืนค่า models.User Object ของผู้ใช้งานปัจจุบัน หรือยก HTTPException หาก Token ไม่ถูกต้องหรือไม่ได้รับอนุญาต
  6. ฟังก์ชัน Path Operation จะได้รับ current_user Object ซึ่งสามารถนำไปใช้ในการกำหนดสิทธิ์ (Authorization) เพิ่มเติมได้ครับ

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

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

การจัดการข้อผิดพลาดที่ดีเป็นสิ่งสำคัญสำหรับ API ที่ใช้งานง่ายและเสถียร FastAPI มีวิธีจัดการข้อผิดพลาดทั้งแบบ Built-in และ Custom

HTTPException

FastAPI มีคลาส HTTPException ที่ใช้งานง่ายสำหรับการส่ง HTTP Error Responses

# main.py (ตัวอย่าง)

from fastapi import FastAPI, HTTPException, status # เพิ่ม status


app = FastAPI()


# ... (โค้ดอื่นๆ) ...


@app.get("/items/{item_id}")

async def read_item_with_error(item_id: int):

    if item_id == 404:

        raise HTTPException(status_code=404, detail="Item not found", headers={"X-Error": "Item-ID-4

    	

บทความที่เกี่ยวข้อง
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