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

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

March 16, 2026
0 Views

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

สารบัญ

ทำไมต้อง FastAPI? ข้อดีที่ทำให้คุณหลงรัก

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

ประสิทธิภาพระดับสุดยอด

FastAPI ถูกสร้างขึ้นบน Starlette สำหรับส่วนของเว็บและ Pydantic สำหรับส่วนของการตรวจสอบข้อมูล ซึ่งทั้งสองไลบรารีนี้ขึ้นชื่อเรื่องประสิทธิภาพที่สูงมากครับ ทำให้ FastAPI เป็นหนึ่งในเฟรมเวิร์ก Python ที่เร็วที่สุดในการสร้าง API สามารถรองรับ Request ได้จำนวนมากพร้อมกัน และเหมาะสำหรับแอปพลิเคชันที่ต้องการความเร็วและการตอบสนองที่ฉับไวครับ

ใช้งานง่ายและโค้ดอ่านออกได้

ด้วยการใช้ Type Hints ของ Python 3.6+ ทำให้โค้ด FastAPI มีความชัดเจน อ่านง่าย และสามารถตรวจสอบข้อผิดพลาดได้ตั้งแต่ตอนเขียนโค้ด (IDE support) ช่วยลดเวลาในการพัฒนาและ Debug ได้อย่างมหาศาลครับ คุณไม่จำเป็นต้องเขียนโค้ด Boilerplate ซ้ำๆ เพื่อทำสิ่งพื้นฐาน ทำให้โฟกัสไปที่ Business Logic ของแอปพลิเคชันได้เต็มที่ครับ

การตรวจสอบข้อมูลอัตโนมัติ (Pydantic)

นี่คือจุดเด่นที่สำคัญมากอีกข้อหนึ่งครับ! FastAPI ใช้ Pydantic ในการตรวจสอบ (Validate) และแปลง (Serialize) ข้อมูลขาเข้า (Request Body, Query Parameters, Path Parameters) และข้อมูลขาออก (Response) โดยอัตโนมัติ เพียงแค่คุณกำหนด Type Hints ให้กับพารามิเตอร์ของฟังก์ชัน และสร้าง Pydantic Model เท่านั้นครับ หากข้อมูลที่ส่งมาไม่ตรงตามรูปแบบที่กำหนด FastAPI จะส่งข้อผิดพลาดที่เป็นมาตรฐานกลับไปให้โดยอัตโนมัติ ซึ่งช่วยประหยัดเวลาในการเขียนโค้ด Validation ได้อย่างมหาศาลครับ

เอกสาร API แบบโต้ตอบ (Swagger UI / ReDoc)

FastAPI สร้างเอกสาร API แบบโต้ตอบได้โดยอัตโนมัติจากโค้ดของคุณ โดยใช้มาตรฐาน OpenAPI (หรือที่รู้จักกันในชื่อ Swagger) ครับ เมื่อคุณรันแอปพลิเคชัน คุณสามารถเข้าถึงเอกสารเหล่านี้ได้ที่ /docs (Swagger UI) และ /redoc (ReDoc) ซึ่งช่วยให้นักพัฒนา Frontend หรือผู้ใช้ API สามารถทำความเข้าใจ ทดสอบ และทำงานร่วมกับ API ของคุณได้อย่างง่ายดาย ไม่ต้องเขียนเอกสาร API แยกต่างหากอีกต่อไปครับ

รองรับ Asynchronous (async/await)

FastAPI รองรับการทำงานแบบ Asynchronous (async/await) อย่างเต็มรูปแบบ ทำให้สามารถจัดการกับ I/O-bound tasks เช่น การเรียกใช้ฐานข้อมูล หรือการเรียกใช้ API ภายนอก ได้อย่างมีประสิทธิภาพ โดยไม่บล็อกการทำงานหลักของแอปพลิเคชัน ช่วยให้ API ของคุณสามารถรองรับผู้ใช้งานพร้อมกันได้จำนวนมากขึ้นครับ

ด้วยข้อดีเหล่านี้ FastAPI จึงเป็นตัวเลือกที่ยอดเยี่ยมสำหรับการสร้าง REST API ไม่ว่าจะเป็นโปรเจกต์ขนาดเล็กหรือโปรเจกต์ขนาดใหญ่ที่ซับซ้อนครับ

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

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

ติดตั้ง Python

FastAPI ต้องการ Python เวอร์ชัน 3.7 ขึ้นไปครับ หากคุณยังไม่มี Python ติดตั้งอยู่ในเครื่อง สามารถดาวน์โหลดได้จากเว็บไซต์ทางการของ Python (https://www.python.org/downloads/) และทำตามขั้นตอนการติดตั้งได้เลยครับ

หลังจากติดตั้งเสร็จแล้ว ลองตรวจสอบเวอร์ชันของ Python โดยใช้คำสั่งใน Terminal/Command Prompt:

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

สร้าง Virtual Environment

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

1. สร้าง Virtual Environment ในโฟลเดอร์โปรเจกต์ของคุณ (สมมติว่าชื่อโฟลเดอร์โปรเจกต์คือ my_fastapi_app):

mkdir my_fastapi_app
cd my_fastapi_app
python -m venv venv

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

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

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

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

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

pip install fastapi "uvicorn[standard]"

"uvicorn[standard]" จะติดตั้ง Uvicorn พร้อมกับ Dependencies ที่จำเป็นสำหรับการทำงานพื้นฐานครับ

เครื่องมือที่แนะนำ

  • Visual Studio Code (VS Code): เป็น IDE ที่ยอดเยี่ยมสำหรับ Python มี Extension ที่ช่วยให้การพัฒนา FastAPI ง่ายขึ้นมากครับ เช่น Python Extension, Pylance.
  • Postman/Insomnia: ใช้สำหรับทดสอบ API โดยการส่ง HTTP Request ไปยัง Endpoint ต่างๆ ของคุณครับ
  • Git: สำหรับการจัดการเวอร์ชันของโค้ดครับ

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

เมื่อเตรียมสภาพแวดล้อมพร้อมแล้ว เรามาเริ่มสร้าง API แรกกันเลยครับ เพื่อให้เห็นภาพรวมของโครงสร้างโค้ด FastAPI ครับ

Hello World แรกของคุณ

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

from fastapi import FastAPI

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

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

# กำหนด Path Operation Decorator สำหรับ HTTP GET request ที่ path "/items/{item_id}"
@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 จากไลบรารี fastapi ครับ
  • app = FastAPI(): สร้าง instance ของแอปพลิเคชัน FastAPI ครับ นี่คือจุดเริ่มต้นของ API ของเรา
  • @app.get("/"): นี่คือ Path Operation Decorator ครับ มันบอก FastAPI ว่าฟังก์ชัน read_root ด้านล่างควรถูกเรียกใช้เมื่อมี HTTP GET request เข้ามาที่ Path "/" ครับ
  • async def read_root():: นี่คือ Path Operation Function ครับ เป็นฟังก์ชัน Python ปกติ แต่เราใช้ async def เพื่อให้รองรับ Asynchronous Operation ครับ ในที่นี้ฟังก์ชันจะคืนค่าเป็น Python Dictionary ซึ่ง FastAPI จะแปลงเป็น JSON Response ให้โดยอัตโนมัติครับ
  • @app.get("/items/{item_id}"): ตัวอย่างการสร้าง Endpoint ที่มี Path Parameter item_id ครับ
  • async def read_item(item_id: int, q: str | None = None):: ฟังก์ชันนี้รับ item_id ซึ่งเรากำหนด Type Hint เป็น int และ q ซึ่งเป็น Query Parameter ที่เป็น str และเป็น Optional (None เป็นค่าเริ่มต้น) ครับ FastAPI จะทำการ Validate Type ของข้อมูลเหล่านี้ให้โดยอัตโนมัติครับ

การรัน FastAPI ด้วย Uvicorn

เมื่อสร้างไฟล์ main.py แล้ว เราสามารถรันแอปพลิเคชันได้โดยใช้ Uvicorn ครับ

ตรวจสอบให้แน่ใจว่าคุณอยู่ในโฟลเดอร์โปรเจกต์ my_fastapi_app และ Virtual Environment ถูก Activate อยู่

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

เมื่อรันคำสั่งนี้ คุณจะเห็นข้อความคล้ายๆ แบบนี้ใน Terminal:

INFO:     Will watch for changes in these directories: ['/path/to/my_fastapi_app']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx] using statreload
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

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

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

  • http://127.0.0.1:8000/: คุณจะเห็น {"message": "Hello FastAPI! Welcome to SiamLancard.com"}
  • http://127.0.0.1:8000/items/5: คุณจะเห็น {"item_id": 5}
  • http://127.0.0.1:8000/items/5?q=somequery: คุณจะเห็น {"item_id": 5, "q": "somequery"}

และที่สำคัญ ลองเข้าไปที่ http://127.0.0.1:8000/docs ครับ คุณจะเห็นเอกสาร API แบบโต้ตอบที่สวยงามที่สร้างขึ้นโดยอัตโนมัติจากโค้ดของคุณ! นี่คือความเจ๋งของ FastAPI ครับ

การสร้าง Endpoint และ Path Operations

ในส่วนนี้ เราจะมาเจาะลึกวิธีการสร้าง Endpoint ต่างๆ และการจัดการ Path Operations ซึ่งเป็นหัวใจสำคัญของการสร้าง REST API ครับ

HTTP Methods (GET, POST, PUT, DELETE)

REST API จะใช้ HTTP Methods ในการระบุประเภทของการดำเนินการที่ต้องการทำกับ Resource ครับ FastAPI มี Decorator สำหรับแต่ละ Method ให้ใช้งานง่ายๆ ครับ

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

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

# สร้าง In-memory database เพื่อเก็บข้อมูล (สำหรับตัวอย่าง)
items_db = {}
item_id_counter = 0

# GET: ดึงข้อมูลทั้งหมด
@app.get("/items/", summary="ดึงรายการสินค้าทั้งหมด")
async def read_items():
    """
    ดึงข้อมูลรายการสินค้าทั้งหมดที่มีอยู่ในระบบ
    """
    return items_db

# GET: ดึงข้อมูล item ตาม ID
@app.get("/items/{item_id}", summary="ดึงข้อมูลสินค้าตาม ID")
async def read_item_by_id(item_id: int):
    """
    ดึงข้อมูลของสินค้าที่ระบุด้วย `item_id`
    """
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

# POST: สร้าง item ใหม่
@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED, summary="สร้างสินค้าใหม่")
async def create_item(item: Item):
    """
    สร้างสินค้าใหม่ในระบบ
    - **name**: ชื่อสินค้า (Required)
    - **description**: รายละเอียดสินค้า (Optional)
    - **price**: ราคา (Required)
    - **tax**: ภาษี (Optional)
    """
    global item_id_counter
    item_id_counter += 1
    items_db[item_id_counter] = item
    return item

# PUT: อัปเดต item ที่มีอยู่
@app.put("/items/{item_id}", response_model=Item, summary="อัปเดตข้อมูลสินค้า")
async def update_item(item_id: int, item: Item):
    """
    อัปเดตข้อมูลของสินค้าที่ระบุด้วย `item_id`
    หากสินค้าไม่มีอยู่ จะทำการสร้างใหม่
    """
    if item_id not in items_db:
        # หากไม่มี ให้สร้างใหม่
        items_db[item_id] = item
        return item
    items_db[item_id] = item
    return item

# DELETE: ลบ item
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT, summary="ลบสินค้า")
async def delete_item(item_id: int):
    """
    ลบสินค้าออกจากระบบด้วย `item_id`
    """
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    del items_db[item_id]
    return # No content for 204

จากตัวอย่างข้างต้น:

  • @app.get(): สำหรับการดึงข้อมูล
  • @app.post(): สำหรับการสร้างข้อมูลใหม่
  • @app.put(): สำหรับการอัปเดตข้อมูลทั้งหมดของ Resource
  • @app.delete(): สำหรับการลบข้อมูล

นอกจากนี้ยังมี @app.patch() สำหรับการอัปเดตข้อมูลบางส่วนของ Resource ด้วยครับ

เรายังได้ใช้ summary ใน Decorator และ Docstring ในฟังก์ชัน ซึ่งจะถูกนำไปแสดงในเอกสาร API โดยอัตโนมัติ ช่วยให้เอกสารของคุณมีรายละเอียดและเป็นประโยชน์มากขึ้นครับ

Path Parameters

Path Parameters คือค่าที่ถูกส่งมาใน URL Path มักใช้ระบุ Resource ที่ต้องการเข้าถึง เช่น /items/{item_id}.

# ... (โค้ดก่อนหน้า) ...

@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
    """
    ตัวอย่างการใช้หลาย Path Parameters
    """
    return {"user_id": user_id, "item_id": item_id}

FastAPI จะรับรู้ Path Parameters โดยอัตโนมัติ และจะทำการแปลง Type ให้ตาม Type Hint ที่คุณกำหนด (เช่น int, str, float, bool, UUID) หากข้อมูลที่ส่งมาไม่ตรง Type ก็จะเกิดข้อผิดพลาด 422 Unprocessable Entity โดยอัตโนมัติครับ

Query Parameters

Query Parameters คือพารามิเตอร์ที่ส่งมาใน URL ต่อจากเครื่องหมาย ? เช่น /items/?skip=0&limit=10 มักใช้สำหรับการกรอง, การแบ่งหน้า (Pagination) หรือการจัดเรียงข้อมูลครับ

# ... (โค้ดก่อนหน้า) ...

@app.get("/items_pagination/")
async def read_items_pagination(skip: int = 0, limit: int = 10):
    """
    ตัวอย่าง Query Parameters สำหรับ Pagination
    - **skip**: จำนวนรายการที่จะข้ามไป (ค่าเริ่มต้น 0)
    - **limit**: จำนวนรายการที่จะดึงมา (ค่าเริ่มต้น 10)
    """
    # ในที่นี้จะดึงข้อมูลจาก items_db จริงๆ
    all_items = list(items_db.values())
    return all_items[skip : skip + limit]

จากตัวอย่าง skip: int = 0 หมายความว่า skip เป็น Query Parameter ที่มี Type เป็น int และมีค่าเริ่มต้นคือ 0 ถ้าไม่มีการระบุมาใน Request ครับ

การรวม Path และ Query Parameters

คุณสามารถรวม Path Parameters และ Query Parameters ใน Path Operation เดียวกันได้ง่ายๆ ครับ

# ... (โค้ดก่อนหน้า) ...

@app.get("/mixed/{item_id}")
async def get_mixed_params(item_id: int, q: str | None = None, short: bool = False):
    """
    ตัวอย่างการรวม Path และ Query Parameters
    """
    item = {"item_id": item_id}
    if q:
        item.update({"q": q})
    if not short:
        item.update({"description": "This is an amazing item!"})
    return item

Optional Parameters

การกำหนดให้ Parameter เป็น Optional ทำได้โดยการกำหนดค่าเริ่มต้นเป็น None หรือใช้ Optional จากโมดูล typing (ใน Python 3.10+ สามารถใช้ | None ได้เลย)

# ... (โค้ดก่อนหน้า) ...

from typing import Optional # สำหรับ Python < 3.10

@app.get("/optional_item/")
async def read_optional_item(name: Optional[str] = None): # หรือ name: str | None = None สำหรับ Python 3.10+
    """
    ตัวอย่าง Optional Query Parameter
    """
    if name:
        return {"message": f"Hello {name}"}
    return {"message": "Hello Guest"}

การส่งข้อมูลด้วย Request Body (POST, PUT)

เมื่อคุณต้องการส่งข้อมูลจำนวนมากหรือข้อมูลที่มีโครงสร้างซับซ้อนไปยัง API (เช่น เมื่อสร้างหรืออัปเดต Resource) คุณจะใช้ Request Body ซึ่งมักจะอยู่ในรูปแบบ JSON ครับ FastAPI ทำให้การจัดการ Request Body ง่ายมากๆ ด้วยการใช้ Pydantic Models ครับ

Pydantic Models สำหรับ Request Body

Pydantic Models เป็นคลาส Python ที่สืบทอดมาจาก BaseModel ของ Pydantic ใช้สำหรับกำหนดโครงสร้างและ Type ของข้อมูลที่เราคาดหวังจาก Request Body ครับ

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool = False

จากตัวอย่างข้างต้น:

  • name: str: หมายถึงฟิลด์ name เป็น String และเป็น Required (ต้องมี)
  • description: str | None = None: หมายถึงฟิลด์ description เป็น String หรือ None และเป็น Optional (ไม่จำเป็นต้องมี)
  • price: float: ฟิลด์ price เป็น Floating Point Number และเป็น Required
  • disabled: bool = False: ฟิลด์ disabled เป็น Boolean และมีค่าเริ่มต้นเป็น False ถ้าไม่มีการส่งค่ามา

Pydantic จะทำการตรวจสอบข้อมูลที่ส่งเข้ามาใน Request Body โดยอัตโนมัติ หากข้อมูลไม่ตรงตาม Model ที่กำหนด ก็จะส่งข้อผิดพลาด 422 กลับไปให้ครับ

การใช้งาน Request Body ใน Path Operation

เพียงแค่คุณกำหนด Type Hint ของพารามิเตอร์ใน Path Operation Function เป็น Pydantic Model FastAPI ก็จะรู้ว่านี่คือ Request Body ครับ

# ... (โค้ดก่อนหน้า) ...

@app.post("/items_body/", response_model=Item)
async def create_item_with_body(item: Item):
    """
    สร้างสินค้าใหม่ด้วย Request Body โดยใช้ Pydantic Model
    """
    # ในโลกจริง คุณจะบันทึก item นี้ลงฐานข้อมูล
    # สำหรับตัวอย่าง เราแค่ส่งคืน item ที่ได้รับมา
    print(f"Received item: {item.dict()}")
    return item

เมื่อคุณส่ง POST Request ไปที่ /items_body/ ด้วย JSON Body เช่น:

{
    "name": "Laptop",
    "price": 1200.0,
    "description": "Powerful gaming laptop",
    "tax": 120.0
}

FastAPI จะรับข้อมูล JSON, Validate ด้วย Item Model, และแปลงเป็น Object Item ที่คุณสามารถใช้งานได้ในฟังก์ชันครับ

การ Nested Pydantic Models

บางครั้งข้อมูลของคุณอาจมีโครงสร้างที่ซับซ้อน มี Object ซ้อนกันอยู่ คุณสามารถใช้ Nested Pydantic Models ได้ครับ

from pydantic import BaseModel

class Image(BaseModel):
    url: str
    name: str

class ItemWithImages(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None
    tags: list[str] = [] # List ของ String
    images: list[Image] | None = None # List ของ Image Model

@app.post("/items_with_images/", response_model=ItemWithImages)
async def create_item_with_images(item: ItemWithImages):
    """
    สร้างสินค้าที่มีรูปภาพและ Tags โดยใช้ Nested Pydantic Models
    """
    print(f"Received item with images: {item.dict()}")
    return item

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

{
    "name": "Smartphone",
    "price": 800.0,
    "tags": ["mobile", "electronics"],
    "images": [
        {"url": "http://example.com/img1.jpg", "name": "Front view"},
        {"url": "http://example.com/img2.jpg", "name": "Back view"}
    ]
}

FastAPI และ Pydantic จะจัดการการ Validate และ Parsing ข้อมูลซ้อนกันนี้ให้คุณโดยอัตโนมัติครับ

การส่งข้อมูลหลายรูปแบบ (Query, Path, Body)

คุณสามารถใช้ Path Parameters, Query Parameters และ Request Body ใน Path Operation เดียวกันได้ครับ

# ... (โค้ดก่อนหน้า) ...

@app.put("/items_full/{item_id}")
async def update_item_full(
    item_id: int,
    item: Item,
    q: str | None = None,
    user_agent: str | None = None # ตัวอย่าง Header Parameter
):
    """
    อัปเดตสินค้าโดยใช้ Path, Body และ Query Parameters
    """
    results = {"item_id": item_id, "item": item.dict()}
    if q:
        results.update({"q": q})
    if user_agent:
        results.update({"user_agent": user_agent})
    return results

ในตัวอย่างนี้ item_id เป็น Path Parameter, item เป็น Request Body และ q เป็น Query Parameter ครับ FastAPI จะแยกแยะและจัดการข้อมูลเหล่านี้ให้คุณโดยอัตโนมัติอย่างชาญฉลาดครับ

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

การตรวจสอบข้อมูล (Data Validation) และการจัดการข้อผิดพลาด (Error Handling) เป็นส่วนสำคัญของ API ที่ดีครับ FastAPI มีเครื่องมือที่ยอดเยี่ยมในการช่วยคุณทำสิ่งเหล่านี้ครับ

Data Validation ด้วย Pydantic

Pydantic ไม่ได้เป็นเพียงแค่การกำหนด Type Hint เท่านั้น แต่ยังสามารถกำหนดเงื่อนไขการตรวจสอบข้อมูลที่ซับซ้อนมากขึ้นได้ด้วย Field Validation ครับ

from pydantic import BaseModel, Field

class Product(BaseModel):
    name: str = Field(..., min_length=3, max_length=50, description="ชื่อสินค้า")
    description: str | None = Field(None, max_length=300, description="รายละเอียดสินค้า")
    price: float = Field(..., gt=0, description="ราคาต้องมากกว่า 0") # gt = greater than
    discount_percentage: float = Field(0, ge=0, le=100, description="ส่วนลดเป็นเปอร์เซ็นต์ (0-100)") # ge = greater equal, le = less equal
    tags: list[str] = Field(default_factory=list, description="รายการแท็ก")

@app.post("/products/", response_model=Product)
async def create_product(product: Product):
    """
    สร้างสินค้าใหม่พร้อมการตรวจสอบข้อมูลที่เข้มงวด
    """
    return product

ประโยชน์ของการตรวจสอบข้อมูล:

  • ความถูกต้องของข้อมูล: มั่นใจได้ว่าข้อมูลที่เข้ามาใน API เป็นไปตามรูปแบบและเงื่อนไขที่กำหนด
  • ลดข้อผิดพลาด: ป้องกันข้อมูลที่ไม่ถูกต้องเข้าสู่ระบบฐานข้อมูลหรือการประมวลผล
  • ปรับปรุงประสบการณ์ผู้ใช้: ผู้ใช้จะได้รับข้อความผิดพลาดที่ชัดเจนและเข้าใจง่ายเมื่อส่งข้อมูลไม่ถูกต้อง
  • ความปลอดภัย: ช่วยป้องกันการโจมตีบางประเภท เช่น SQL Injection (แม้จะไม่ใช่การป้องกันหลักก็ตาม)

จากตัวอย่าง Field():

  • ...: หมายถึงฟิลด์นี้เป็น Required
  • min_length, max_length: กำหนดความยาวต่ำสุดและสูงสุดของ String
  • gt (greater than), ge (greater equal): กำหนดค่าตัวเลขที่มากกว่า/มากกว่าหรือเท่ากับ
  • lt (less than), le (less equal): กำหนดค่าตัวเลขที่น้อยกว่า/น้อยกว่าหรือเท่ากับ
  • default_factory=list: ใช้สำหรับกำหนดค่าเริ่มต้นที่เป็น Mutable Object เช่น List หรือ Dictionary เพื่อป้องกันปัญหาการแชร์ Instance เดียวกัน

หากคุณส่ง Request Body ที่ไม่ตรงตามเงื่อนไขเหล่านี้ FastAPI จะส่ง Response ที่มี Status Code 422 (Unprocessable Entity) พร้อมรายละเอียดข้อผิดพลาดที่ชัดเจนโดยอัตโนมัติครับ

การกำหนดค่าเริ่มต้นและ Required Fields

  • Required Fields: หากคุณกำหนด Type Hint ให้กับฟิลด์โดยไม่มีค่าเริ่มต้น (เช่น name: str) ฟิลด์นั้นจะถือว่าเป็น Required ครับ
  • Optional Fields: หากคุณกำหนดค่าเริ่มต้นเป็น None (description: str | None = None) หรือกำหนดค่าเริ่มต้นอื่นๆ (discount_percentage: float = 0) ฟิลด์นั้นจะถือว่าเป็น Optional ครับ

การจัดการข้อผิดพลาด HTTP (HTTPException)

บางครั้งข้อผิดพลาดไม่ได้เกิดจากการ Validate ข้อมูล แต่เป็น Business Logic หรือ Resource ที่ไม่พบ คุณสามารถใช้ HTTPException เพื่อส่งข้อผิดพลาด HTTP มาตรฐานกลับไปได้ครับ

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

# ... (Pydantic Models และ Path Operations อื่นๆ) ...

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """
    ดึงข้อมูลผู้ใช้ตาม ID
    """
    if user_id < 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="User ID cannot be negative"
        )
    if user_id not in [1, 2, 3]: # สมมติว่ามีแค่ user 1, 2, 3
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
            headers={"X-Error": "There goes my error"} # สามารถเพิ่ม Header ได้
        )
    return {"user_id": user_id, "name": f"User {user_id}"}

เมื่อคุณเรียก /users/5 คุณจะได้ Response ที่มี Status Code 404 และ JSON Body {"detail": "User not found"} ครับ

status_code ควรใช้ค่าจาก fastapi.status เพื่อความชัดเจนและป้องกันการพิมพ์ผิดครับ

Advanced Features (ฟีเจอร์ขั้นสูง)

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

Dependency Injection (การพึ่งพา)

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

ทำไมต้องใช้ Dependency Injection?

  • ลดการทำซ้ำ (DRY): โค้ดที่ใช้ร่วมกันสามารถเขียนครั้งเดียวและนำไปใช้ได้หลายที่ เช่น การเชื่อมต่อฐานข้อมูล, การตรวจสอบสิทธิ์.
  • ทดสอบง่ายขึ้น: สามารถ Mock Dependencies ได้ง่าย ทำให้การเขียน Unit Test มีประสิทธิภาพ.
  • จัดการง่าย: การเปลี่ยนแปลง Logic ใน Dependency จะส่งผลกระทบต่อที่เดียว.
  • ความยืดหยุ่น: สามารถสลับเปลี่ยน Implementation ของ Dependency ได้โดยไม่ต้องแก้ไข Path Operation.

ตัวอย่างการใช้ Dependency สำหรับ Query Parameters ทั่วไป:

from fastapi import FastAPI, Depends

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    """
    Dependency สำหรับ Query Parameters ที่ใช้บ่อย
    """
    return {"q": q, "skip": skip, "limit": limit}

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

ในตัวอย่างนี้ common_parameters เป็น Dependency Function ครับ เมื่อ read_items_di ถูกเรียกใช้ FastAPI จะเรียก common_parameters ก่อนและส่งค่าที่คืนกลับมาให้กับพารามิเตอร์ commons ครับ

ตัวอย่างการใช้ Dependency สำหรับการเชื่อมต่อฐานข้อมูล (จะเห็นประโยชน์ชัดเจนในส่วนฐานข้อมูลครับ):

# สมมติว่ามีฟังก์ชันสำหรับเชื่อมต่อฐานข้อมูล
def get_database_session():
    # โค้ดสำหรับการสร้าง session ฐานข้อมูล
    try:
        db = "Database Session Object" # ตัวอย่าง
        yield db
    finally:
        # โค้ดสำหรับการปิด session ฐานข้อมูล
        pass

@app.get("/data/")
async def get_data_from_db(db_session: str = Depends(get_database_session)):
    """
    Endpoint ที่ใช้ Dependency Injection เพื่อรับ Database Session
    """
    return {"message": f"Using {db_session} for data access"}

yield ใน get_database_session ทำให้ FastAPI สามารถจัดการ Lifecycle ของ Dependency ได้ครับ โดยจะรันโค้ดก่อน yield ก่อนเรียก Path Operation และรันโค้ดหลัง yield หลังจาก Path Operation ทำงานเสร็จ (เช่น การปิด Connection ฐานข้อมูล) ครับ

อ่านเพิ่มเติมเกี่ยวกับการจัดการ Dependency Injection

Security และ Authentication (OAuth2, JWT)

FastAPI มีเครื่องมือช่วยในการสร้างระบบ Authentication และ Authorization ที่แข็งแกร่งครับ โดยรองรับมาตรฐานต่างๆ เช่น OAuth2 พร้อม JSON Web Tokens (JWT).

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
from jose import JWTError, jwt
from datetime import datetime, timedelta

app = FastAPI()

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

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # "token" คือ endpoint สำหรับรับ token

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class UserInDB(BaseModel):
    username: str
    hashed_password: str

class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None

# Users database (สำหรับตัวอย่าง)
fake_users_db = {
    "john.doe": {
        "username": "john.doe",
        "email": "[email protected]",
        "full_name": "John Doe",
        "hashed_password": "supersecretpassword", # ในโลกจริงควรใช้ bcrypt
        "disabled": False,
    }
}

def verify_password(plain_password: str, hashed_password: str):
    return plain_password == hashed_password # ในโลกจริงควรใช้ bcrypt.checkpw

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

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

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user_from_db(token_data.username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordBearer = Depends(OAuth2PasswordBearer(tokenUrl="token"))):
    # This is a simplified example. In a real app, you'd use OAuth2PasswordRequestForm
    # to get username and password from form data.
    # For now, let's assume form_data is a dictionary with username and password
    username = "john.doe" # Hardcoded for example simplicity
    password = "supersecretpassword" # Hardcoded for example simplicity

    user = get_user_from_db(username)
    if not user or not verify_password(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=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_user)):
    """
    Endpoint ที่ต้องการการยืนยันตัวตน
    """
    return current_user

ในตัวอย่างนี้ OAuth2PasswordBearer และ Depends(get_current_user) ช่วยให้คุณสามารถปกป้อง Endpoint ต่างๆ ได้อย่างง่ายดายครับ

Middlewares

Middlewares คือฟังก์ชันที่สามารถทำงานก่อนหรือหลัง Request จะถูกประมวลผลโดย Path Operation Function ครับ มักใช้สำหรับงานทั่วๆ ไป เช่น Logging, CORS (Cross-Origin Resource Sharing), Authentication หรือการเพิ่ม Header ครับ

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# เพิ่ม CORS Middleware
origins = [
    "http://localhost",
    "http://localhost:8080",
    "https://www.siamlancard.com",
    # เพิ่มโดเมนอื่นๆ ที่คุณต้องการอนุญาต
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom Middleware สำหรับการจับเวลา Request
@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} took {process_time:.4f} seconds")
    return response

@app.get("/test_middleware/")
async def test_middleware():
    """
    Endpoint สำหรับทดสอบ Middleware
    """
    return {"message": "Middleware tested successfully!"}

CORSMiddleware ช่วยให้เบราว์เซอร์จากโดเมนอื่นสามารถเรียกใช้ API ของคุณได้ครับ และ Custom Middleware add_process_time_header จะเพิ่ม Header X-Process-Time ในทุก Response ครับ

Background Tasks

บางครั้งคุณอาจต้องการให้ API ส่ง Response กลับไปหา Client ทันที แต่มีบางอย่างที่ต้องทำต่อเบื้องหลัง เช่น ส่งอีเมลแจ้งเตือน หรือประมวลผลข้อมูลขนาดใหญ่ FastAPI มี 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, background_tasks: BackgroundTasks):
    """
    ส่ง Notification โดยใช้ Background Tasks
    """
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification sent in background"}

เมื่อเรียก /send_notification/, API จะส่ง Response กลับไปทันที ในขณะที่ฟังก์ชัน write_log จะถูกเรียกใช้ใน Background ครับ

Static Files

หาก API ของคุณต้องการ Serve ไฟล์ Static เช่น HTML, CSS, JavaScript หรือรูปภาพ คุณสามารถใช้ StaticFiles ได้ครับ

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# Mount the static files directory
app.mount("/static", StaticFiles(directory="static"), name="static")

@app.get("/")
async def read_root():
    return {"message": "Hello from FastAPI! See static content at /static/index.html"}

# สร้างโฟลเดอร์ 'static' และไฟล์ 'static/index.html'
# ในโฟลเดอร์เดียวกับ main.py
# ไฟล์ static/index.html อาจมีเนื้อหาดังนี้:
# <!DOCTYPE html>
# <html>
# <head>
#     <title>Static Page</title>
# </head>
# <body>
#     <h1>Welcome to Static Page!</h1>
#     <p>This content is served by FastAPI StaticFiles.</p>
# </body>
# </html>

เมื่อรันแอปพลิเคชันและเข้าถึง http://127.0.0.1:8000/static/index.html คุณจะเห็นเนื้อหาของไฟล์ index.html ครับ

การเชื่อมต่อกับฐานข้อมูล (ตัวอย่าง: SQLite + SQLAlchemy)

API ส่วนใหญ่จำเป็นต้องทำงานกับฐานข้อมูลครับ เราจะใช้ SQLAlchemy ซึ่งเป็น ORM (Object-Relational Mapper) ที่ได้รับความนิยมใน Python และเชื่อมต่อกับ SQLite ซึ่งเป็นฐานข้อมูลไฟล์ที่ใช้งานง่ายสำหรับตัวอย่างครับ

เราจะใช้แนวทาง Dependency Injection สำหรับ Database Session เพื่อให้โค้ด Modular และจัดการง่ายครับ

ติดตั้ง SQLAlchemy และ Alembic

Alembic เป็นเครื่องมือสำหรับ Database Migrations ซึ่งช่วยในการจัดการการเปลี่ยนแปลง Schema ของฐานข้อมูลครับ

pip install sqlalchemy alembic "psycopg2-binary>=2.8.5" # psycopg2-binary สำหรับ PostgreSQL ถ้าคุณต้องการใช้

ในตัวอย่างนี้เราจะใช้ SQLite ซึ่งไม่ต้องการ Driver เพิ่มเติมครับ

สร้าง Model และ Schema

สร้างไฟล์ database.py, models.py, schemas.py และ crud.py เพื่อจัดระเบียบโค้ดครับ

database.py: สำหรับการตั้งค่าฐานข้อมูลและการสร้าง Session

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# URL ของฐานข้อมูล SQLite
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"

# สร้าง Engine สำหรับเชื่อมต่อฐานข้อมูล
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} # สำหรับ SQLite เท่านั้น
)

# สร้าง SessionLocal class สำหรับสร้าง Database Session
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
    finally:
        db.close()

models.py: สำหรับกำหนดโครงสร้างตารางในฐานข้อมูล (SQLAlchemy Models)

from sqlalchemy import Column, Integer, String, Boolean
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, index=True)

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

schemas.py: สำหรับ Pydantic Models ที่ใช้ในการส่งและรับข้อมูลจาก API (Data Validation/Serialization)

from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: str | None = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True # เพื่อให้ Pydantic สามารถอ่านข้อมูลจาก SQLAlchemy model ได้

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = [] # List ของ Item

    class Config:
        orm_mode = True

crud.py: สำหรับฟังก์ชัน CRUD (Create, Read, Update, Delete) ที่ทำงานกับฐานข้อมูล

from sqlalchemy.orm import Session

import models, schemas

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed" # ควรใช้ hashing จริงๆ
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()

def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

การสร้าง Session สำหรับฐานข้อมูล

ใน database.py เราได้สร้าง get_db() ซึ่งเป็น Dependency Function ที่ใช้ yield เพื่อให้ FastAPI จัดการการเปิดและปิด Database Session ให้โดยอัตโนมัติครับ

CRUD Operations

เราจะนำฟังก์ชัน CRUD จาก crud.py มาใช้ใน Path Operations ของเราครับ

main.py (อัปเดต):

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session

import crud, models, schemas
from database import SessionLocal, engine, get_db

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

app = FastAPI()

# Endpoint สำหรับสร้าง User
@app.post("/users/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """
    สร้างผู้ใช้ใหม่
    """
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

# Endpoint สำหรับดึง User ทั้งหมด
@app.get("/users/", response_model=list[schemas.User])
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    ดึงรายการผู้ใช้ทั้งหมด
    """
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

# Endpoint สำหรับดึง User ตาม ID
@app.get("/users/{user_id}", response_model=schemas.User)
async def read_user(user_id: int, db: Session = Depends(get_db)):
    """
    ดึงข้อมูลผู้ใช้ตาม ID
    """
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# Endpoint สำหรับสร้าง Item ให้กับ User
@app.post("/users/{user_id}/items/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
async def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    """
    สร้างสินค้าใหม่ให้กับผู้ใช้ที่ระบุ
    """
    return crud.create_user_item(db=db, item=item, user_id=user_id)

# Endpoint สำหรับดึง Item ทั้งหมด
@app.get("/items/", response_model=list[schemas.Item])
async def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    """
    ดึงรายการสินค้าทั้งหมด
    """
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

# ... (สามารถเพิ่ม PUT/DELETE สำหรับ User และ Item ได้) ...

เมื่อรัน uvicorn main:app --reload และลองเข้าถึง /docs คุณจะเห็น Endpoint ใหม่เหล่านี้พร้อมให้ทดสอบได้ทันทีครับ

อ่านเพิ่มเติมเกี่ยวกับการเชื่อมต่อฐานข้อมูลใน FastAPI

การ Deploy FastAPI API

เมื่อพัฒนา API เสร็จแล้ว ขั้นตอนต่อไปคือการนำไป Deploy บน Production Server เพื่อให้สามารถเข้าถึงได้จริงครับ

Production Server (Gunicorn + Uvicorn Worker)

สำหรับการใช้งานจริง Uvicorn ตัวเดียวอาจไม่เพียงพอ เนื่องจากมันไม่ได้ถูกออกแบบมาเพื่อเป็น Production-ready Process Manager ครับ เรามักจะใช้ Gunicorn เป็น Process Manager และให้ Gunicorn รัน Uvicorn Worker หลายๆ ตัวครับ

1. ติดตั้ง Gunicorn:

pip install gunicorn

2. รันแอปพลิเคชันด้วย Gunicorn และ Uvicorn Workers:

gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

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