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

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

March 16, 2026
1 View

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

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

สารบัญ

ทำความรู้จักกับ REST API และทำไมต้อง FastAPI?

REST API คืออะไร?

REST (Representational State Transfer) API คือชุดของหลักการออกแบบสถาปัตยกรรมซอฟต์แวร์สำหรับการสื่อสารระหว่างระบบคอมพิวเตอร์ผ่านเครือข่าย โดยส่วนใหญ่แล้วจะใช้โปรโตคอล HTTP ในการส่งข้อมูลครับ

ลองนึกภาพว่าคุณต้องการสั่งอาหารออนไลน์ แอปพลิเคชันบนมือถือของคุณ (Client) จะต้องสื่อสารกับระบบร้านอาหาร (Server) เพื่อดูเมนู สั่งอาหาร หรือตรวจสอบสถานะ REST API คือ “ภาษา” หรือ “ข้อตกลง” ที่ช่วยให้การสื่อสารนี้เป็นไปอย่างราบรื่นและเข้าใจกันได้ โดยมีหลักการสำคัญดังนี้ครับ:

  • Client-Server: แยก Client และ Server ออกจากกันอย่างชัดเจน ทำให้แต่ละส่วนพัฒนาและปรับขนาดได้อย่างอิสระ
  • Stateless: Server จะไม่เก็บสถานะของ Client ระหว่าง Request แต่ละครั้ง ทำให้ Server สามารถประมวลผล Request ใดๆ ก็ได้โดยไม่ต้องจำ Request ก่อนหน้า
  • Cacheable: Response จาก Server สามารถถูก Cache ได้ เพื่อปรับปรุงประสิทธิภาพการทำงาน
  • Layered System: Client อาจไม่รู้ว่ากำลังเชื่อมต่อกับ Server โดยตรงหรือผ่านตัวกลาง เช่น Load Balancer หรือ Proxy
  • Uniform Interface: เป็นชุดของข้อจำกัดที่ทำให้ Client และ Server สื่อสารกันได้ง่ายขึ้น เช่น การใช้ HTTP Methods (GET, POST, PUT, DELETE) และการใช้ URI ในการระบุทรัพยากร

REST API เป็นมาตรฐานที่ได้รับความนิยมอย่างสูงในการสร้าง Web Service เนื่องจากความยืดหยุ่น ความง่ายในการทำความเข้าใจ และความเข้ากันได้กับเทคโนโลยีหลากหลายรูปแบบครับ

ทำไมถึงเลือกใช้ FastAPI?

ในโลกของ Python มีเฟรมเวิร์กสำหรับสร้าง API มากมาย เช่น Flask, Django REST Framework แต่ FastAPI ได้ก้าวขึ้นมาเป็นตัวเลือกที่โดดเด่นด้วยเหตุผลหลายประการ ดังนี้ครับ:

  • ความเร็วที่เหนือกว่า: FastAPI สร้างขึ้นบน Starlette (สำหรับเว็บ) และ Pydantic (สำหรับการจัดการข้อมูล) ทำให้ได้ประสิทธิภาพที่เทียบเท่ากับ Node.js หรือ Go รวมถึงยังรองรับ Asynchronous (async/await) ทำให้สามารถจัดการ Request ได้พร้อมกันหลายตัวอย่างมีประสิทธิภาพ
  • ใช้งานง่ายและโค้ดน้อย: ด้วยความสามารถของ Pydantic ทำให้การกำหนดโครงสร้างข้อมูล การตรวจสอบความถูกต้องของข้อมูล (data validation) และการแปลงข้อมูล (serialization) เป็นเรื่องง่ายและใช้โค้ดเพียงไม่กี่บรรทัด
  • เอกสาร API อัตโนมัติ: FastAPI สร้างเอกสาร API แบบอินเทอร์แอคทีฟ (Interactive API Docs) ให้โดยอัตโนมัติ ไม่ว่าจะเป็น Swagger UI หรือ ReDoc ซึ่งช่วยให้นักพัฒนาสามารถทดสอบ API และทำความเข้าใจโครงสร้าง API ได้อย่างรวดเร็วโดยไม่ต้องเขียนเอกสารเองเลยครับ
  • ความปลอดภัยและการจัดการข้อผิดพลาด: มีฟีเจอร์สำหรับการจัดการ Dependencies, Authentication และ Error Handling ที่ช่วยให้การสร้าง API ที่แข็งแกร่งและปลอดภัยเป็นเรื่องง่าย
  • Type Hints: FastAPI ใช้ Type Hints ของ Python อย่างเต็มที่ ทำให้ IDE สามารถให้คำแนะนำ (autocompletion) และตรวจสอบข้อผิดพลาด (type checking) ได้ดีขึ้น ช่วยลดข้อผิดพลาดและเพิ่มความเร็วในการพัฒนา
  • ชุมชนที่เติบโตเร็ว: แม้จะเป็นเฟรมเวิร์กที่ค่อนข้างใหม่ แต่ FastAPI มีชุมชนผู้ใช้งานที่เติบโตอย่างรวดเร็ว มีเอกสารประกอบที่ดี และมีตัวอย่างโค้ดให้ศึกษามากมายครับ

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

สิ่งที่คุณต้องมีก่อนเริ่ม

เพื่อให้บทความนี้เป็นประโยชน์สูงสุด ผมแนะนำให้คุณมีสิ่งเหล่านี้ติดตัวไว้ก่อนนะครับ:

  • ความรู้พื้นฐาน Python: เข้าใจเรื่องตัวแปร, ฟังก์ชัน, คลาส, list, dict และการนำเข้าโมดูล
  • ความรู้พื้นฐาน HTTP/REST: เข้าใจว่า HTTP Method (GET, POST ฯลฯ) คืออะไร, Status Codes และแนวคิดของ REST API
  • Terminal/Command Prompt: สามารถใช้งาน Command Line Interface (CLI) ได้
  • Code Editor: เช่น VS Code, PyCharm หรือ Sublime Text ที่มีส่วนเสริมสำหรับ Python

เริ่มต้นกับ FastAPI: การติดตั้งและ Hello World

การตั้งค่าสภาพแวดล้อม (Virtual Environment)

ก่อนที่เราจะเริ่มเขียนโค้ด การตั้งค่า Virtual Environment เป็นสิ่งสำคัญมากครับ เพื่อแยกโปรเจกต์ของเราออกจากโปรเจกต์อื่นๆ และป้องกันปัญหาเรื่อง Dependency conflict ครับ

เปิด Terminal หรือ Command Prompt ขึ้นมา แล้วทำตามขั้นตอนเหล่านี้ครับ:

# 1. สร้างโฟลเดอร์สำหรับโปรเจกต์ของคุณ
mkdir my-fastapi-app
cd my-fastapi-app

# 2. สร้าง Virtual Environment
python3 -m venv venv

# 3. เปิดใช้งาน Virtual Environment
# บน macOS/Linux:
source venv/bin/activate

# บน Windows:
# .\venv\Scripts\activate

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

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

ต่อไป เราจะติดตั้ง FastAPI และ Uvicorn ซึ่งเป็น ASGI server ที่จะใช้รัน FastAPI application ของเราครับ

pip install fastapi "uvicorn[standard]"

อธิบาย:

  • fastapi: ตัวเฟรมเวิร์กหลัก
  • uvicorn[standard]: Uvicorn พร้อมกับส่วนเสริมมาตรฐานที่จำเป็นสำหรับการทำงาน เช่น python-dotenv และ watchgod สำหรับการรันแบบ reload เมื่อมีการเปลี่ยนแปลงโค้ด

สร้าง API แรกของคุณ: “Hello, FastAPI!”

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

# main.py
from fastapi import FastAPI

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

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

# คุณสามารถเพิ่ม Path Operation อื่นๆ ได้ที่นี่
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    """
    Endpoint ตัวอย่างสำหรับ Path Parameter และ Query Parameter
    """
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

ในโค้ดนี้:

  • เรานำเข้า FastAPI จากไลบรารี fastapi
  • สร้าง app instance จาก FastAPI()
  • ใช้ decorator @app.get("/") เพื่อกำหนด Path Operation สำหรับ HTTP GET request ที่ URL path /
  • ฟังก์ชัน read_root() เป็น Asynchronous function (ใช้ async def) ที่จะถูกเรียกเมื่อมี GET request มาที่ / และจะส่ง Python dictionary กลับไป ซึ่ง FastAPI จะแปลงเป็น JSON response โดยอัตโนมัติครับ
  • ตัวอย่าง /items/{item_id} แสดงการใช้ Path Parameter item_id (ระบุ type hint เป็น int) และ Query Parameter q (ระบุ type hint เป็น str และค่า default เป็น None)

รัน FastAPI Application

หลังจากบันทึกไฟล์ main.py แล้ว กลับไปที่ Terminal (ที่ยังคงอยู่ใน Virtual Environment) และรันคำสั่งนี้ครับ:

uvicorn main:app --reload

อธิบาย:

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

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

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 แล้วครับ!

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

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

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

หัวใจหลักของการสร้าง API ด้วย FastAPI

Path Operations: การกำหนดเส้นทางและ HTTP Method

Path Operation คือการกำหนดว่าเมื่อ Client ส่ง Request มาที่ URL Path ใด ด้วย HTTP Method ใด จะให้รันโค้ดส่วนไหน FastAPI มี decorator สำหรับ HTTP Method หลักๆ ให้ใช้งานครับ:

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

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

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

@app.post("/items/")
async def create_item(item: dict):
    """
    สร้าง item ใหม่
    """
    return {"message": "Item created successfully", "item": item}

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: dict):
    """
    อัปเดต item ที่มีอยู่
    """
    return {"message": f"Item {item_id} updated", "item": item}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    """
    ลบ item
    """
    return {"message": f"Item {item_id} deleted"}

Path Parameters: การส่งข้อมูลผ่าน URL

Path Parameters คือตัวแปรที่ฝังอยู่ใน URL Path เพื่อระบุทรัพยากรที่ต้องการ เช่น /items/1 เมื่อต้องการ Item ID เป็น 1

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

@app.get("/users/{user_id}/orders/{order_id}")
async def get_user_order(user_id: int, order_id: str):
    """
    ดึงข้อมูลคำสั่งซื้อของ User
    """
    return {"user_id": user_id, "order_id": order_id}

FastAPI จะใช้ Type Hints (user_id: int) ในการตรวจสอบความถูกต้องของข้อมูล (validation) และแปลงประเภทข้อมูลให้โดยอัตโนมัติ หากส่ง user_id ที่ไม่ใช่ตัวเลขเข้ามา FastAPI จะตอบกลับด้วย HTTP 422 Unprocessable Entity โดยอัตโนมัติครับ

Query Parameters: พารามิเตอร์เสริมในการค้นหา

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

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

from typing import Optional

@app.get("/products/")
async def read_products(skip: int = 0, limit: int = 10, search: Optional[str] = None):
    """
    ดึงรายการสินค้าพร้อมการแบ่งหน้าและการค้นหา
    - skip: จำนวนรายการที่จะข้ามไป
    - limit: จำนวนรายการที่จะดึงมา
    - search: คำค้นหา (ไม่บังคับ)
    """
    results = {"skip": skip, "limit": limit}
    if search:
        results["search"] = search
    # ในความเป็นจริงตรงนี้จะมีการ query จาก database
    return results

ในตัวอย่างนี้:

  • skip: int = 0 และ limit: int = 10 เป็น Query Parameters ที่มีค่า default หาก Client ไม่ได้ส่งมา
  • search: Optional[str] = None หมายความว่า search เป็น Query Parameter ที่เป็น String และเป็น Optional (ไม่บังคับ) หากมีค่าจะถูกส่งมา แต่ถ้าไม่มีก็จะเป็น None ครับ

Request Body: การส่งข้อมูลผ่าน Pydantic Model

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

มาสร้างไฟล์ models.py เพื่อเก็บ Pydantic Models ของเราครับ

# models.py
from typing import Optional
from pydantic import BaseModel, Field

class ItemBase(BaseModel):
    name: str = Field(..., example="Latest Gadget")
    description: Optional[str] = Field(None, example="A very cool gadget.")
    price: float = Field(..., gt=0, example=99.99)
    tax: Optional[float] = Field(None, example=8.0)

class ItemCreate(ItemBase):
    pass # อาจจะมี field เพิ่มเติมสำหรับการสร้าง

class Item(ItemBase):
    id: int = Field(..., example=1)
    is_offer: Optional[bool] = Field(None, example=True)

    class Config:
        orm_mode = True # เพื่อให้สามารถใช้กับ SQLAlchemy ORM ได้

แล้วแก้ไข main.py เพื่อใช้ Item model ครับ:

# main.py (ต่อจากโค้ดเดิม)
from fastapi import FastAPI, HTTPException
from typing import List
from models import Item, ItemCreate # นำเข้าจาก models.py

app = FastAPI()

# สมมติฐานว่ามีฐานข้อมูล Item อยู่
db_items = {} # ใช้ dictionary จำลองเป็น database
next_item_id = 1

@app.post("/items/", response_model=Item)
async def create_new_item(item: ItemCreate):
    """
    สร้าง Item ใหม่ในระบบ
    """
    global next_item_id
    db_item = Item(id=next_item_id, **item.dict())
    db_items[next_item_id] = db_item
    next_item_id += 1
    return db_item

@app.get("/items/", response_model=List[Item])
async def read_all_items():
    """
    ดึงรายการ Item ทั้งหมด
    """
    return list(db_items.values())

@app.get("/items/{item_id}", response_model=Item)
async def read_single_item(item_id: int):
    """
    ดึงข้อมูล Item เฉพาะ
    """
    item = db_items.get(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

ในตัวอย่างนี้:

  • เราสร้าง Pydantic Model ItemCreate สำหรับ Request Body ที่จะส่งเข้ามาเมื่อสร้าง Item และ Item สำหรับ Response Body ที่จะส่งกลับไป
  • item: ItemCreate ในฟังก์ชัน create_new_item จะบอก FastAPI ว่า Request Body คาดหวังข้อมูลตามโครงสร้างของ ItemCreate
  • FastAPI จะตรวจสอบข้อมูลให้โดยอัตโนมัติ หากข้อมูลไม่ตรงตาม Model (เช่น price ไม่ใช่ตัวเลข หรือเป็นค่าติดลบ) จะตอบกลับด้วย HTTP 422 โดยอัตโนมัติครับ
  • Field(..., gt=0) คือการกำหนดเงื่อนไขเพิ่มเติม (validator) ว่า price ต้องมากกว่า 0
  • example="Latest Gadget" จะช่วยให้เอกสาร API (Swagger UI) แสดงตัวอย่าง Request Body ที่ชัดเจนขึ้นครับ

นี่คือความมหัศจรรย์ของ Pydantic และ FastAPI ที่ทำงานร่วมกันอย่างราบรื่นครับ!

Response Model: กำหนดรูปแบบข้อมูลขาออก

การใช้ response_model ใน decorator เช่น @app.post("/items/", response_model=Item) มีประโยชน์มากครับ มันจะบอก FastAPI ว่าข้อมูลที่ฟังก์ชันนี้ส่งกลับไปควรมีโครงสร้างอย่างไร

  • FastAPI จะตรวจสอบว่าข้อมูลที่ส่งกลับมาตรงตาม response_model หรือไม่
  • จะกรองข้อมูลที่ไม่จำเป็นออกไป หากข้อมูลในฟังก์ชันมี Field เกินกว่าที่ response_model กำหนด
  • ช่วยให้เอกสาร API ชัดเจนขึ้นว่า Response ที่คาดหวังคืออะไร

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

HTTP Status Codes: การตอบกลับสถานะที่ถูกต้อง

การใช้ HTTP Status Codes ที่ถูกต้องเป็นสิ่งสำคัญมากในการสร้าง REST API ที่ดี เพื่อบอกสถานะของ Request ให้กับ Client ได้อย่างชัดเจน

  • 200 OK: Request สำเร็จ
  • 201 Created: ทรัพยากรใหม่ถูกสร้างขึ้นสำเร็จ (นิยมใช้กับ POST)
  • 204 No Content: Request สำเร็จแต่ไม่มีข้อมูลกลับมา (นิยมใช้กับ DELETE)
  • 400 Bad Request: Request ไม่ถูกต้อง (เช่น ข้อมูลไม่ครบถ้วน)
  • 401 Unauthorized: ไม่ได้รับอนุญาตให้เข้าถึง (ไม่มี Token หรือ Token ไม่ถูกต้อง)
  • 403 Forbidden: ได้รับอนุญาตแต่ไม่มีสิทธิ์ในการเข้าถึงทรัพยากรนั้น
  • 404 Not Found: ไม่พบทรัพยากรที่ร้องขอ
  • 422 Unprocessable Entity: Request Body ไม่ถูกต้องตามเงื่อนไข (Pydantic Validation Error)
  • 500 Internal Server Error: ข้อผิดพลาดที่ Server

คุณสามารถกำหนด Status Code ได้ใน Path Operation decorator หรือใช้ HTTPException ครับ

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

@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: dict):
    """
    สร้างผู้ใช้ใหม่
    จะตอบกลับด้วยสถานะ 201 Created
    """
    return {"message": "User created successfully", "user": user}

@app.get("/files/{file_id}")
async def get_file(file_id: str):
    """
    จำลองการดึงไฟล์
    """
    if file_id == "nonexistent":
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
    return {"file_id": file_id, "content": "This is file content."}

การใช้ status.HTTP_201_CREATED จาก fastapi.status ทำให้โค้ดอ่านง่ายขึ้นและป้องกันความผิดพลาดในการพิมพ์ตัวเลขครับ

ฟีเจอร์ขั้นสูงที่ทำให้ FastAPI ทรงพลังยิ่งขึ้น

Dependencies: การจัดการโค้ดซ้ำซ้อนและการฉีดการพึ่งพา

Dependencies Injection เป็นรูปแบบการออกแบบที่ช่วยให้โค้ดของคุณเป็นระเบียบ, สามารถนำกลับมาใช้ใหม่ได้ และทดสอบได้ง่ายขึ้น FastAPI มีระบบ Dependencies ที่ทรงพลังมากครับ

ลองนึกภาพว่าคุณมีฟังก์ชันที่ต้องตรวจสอบสิทธิ์ผู้ใช้ก่อนที่จะดำเนินการต่อ หรือฟังก์ชันที่ต้องเชื่อมต่อฐานข้อมูลในทุกๆ Request แทนที่จะเขียนโค้ดซ้ำๆ ในทุก Path Operation คุณสามารถสร้าง Dependency Function ขึ้นมา และให้ FastAPI จัดการการเรียกใช้ให้เองครับ

ตัวอย่าง: การจำลองการตรวจสอบ API Key

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

async def verify_api_key(x_api_key: str = Header(...)):
    """
    Dependency เพื่อตรวจสอบ API Key
    """
    if x_api_key != "SECRET_API_KEY":
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return x_api_key

@app.get("/protected-route/", dependencies=[Depends(verify_api_key)])
async def get_protected_data():
    """
    เส้นทางที่ต้องใช้ API Key ในการเข้าถึง
    """
    return {"message": "คุณเข้าถึงข้อมูลที่ถูกป้องกันได้แล้ว!"}

@app.get("/another-protected-route/")
async def get_another_protected_data(api_key: str = Depends(verify_api_key)):
    """
    อีกเส้นทางที่ต้องใช้ API Key และเรายังสามารถรับค่าจาก dependency ได้ด้วย
    """
    return {"message": f"คุณเข้าถึงข้อมูลที่ถูกป้องกันได้ด้วย key: {api_key}"}

ในตัวอย่างนี้ verify_api_key คือ Dependency Function ที่จะถูกเรียกก่อน get_protected_data หรือ get_another_protected_data หาก x_api_key ไม่ถูกต้อง ก็จะเกิด HTTPException และ Request จะไม่ไปถึงฟังก์ชันหลักครับ

Dependencies ยังสามารถมี Dependencies อื่นๆ ได้อีกด้วย (sub-dependencies) ทำให้คุณสามารถสร้างโครงสร้างการพึ่งพาที่ซับซ้อนได้อย่างเป็นระเบียบครับ

อ่านเพิ่มเติมเกี่ยวกับ Dependencies

Security & Authentication: การรักษาความปลอดภัย API

FastAPI มีเครื่องมือและแนวทางสำหรับการจัดการ Security และ Authentication ที่หลากหลาย โดยใช้ประโยชน์จากระบบ Dependencies

ตัวอย่าง: OAuth2 with JWT (JSON Web Tokens) – แนวคิดเบื้องต้น

FastAPI มีโมดูล fastapi.security สำหรับจัดการ Authentication Schemes ต่างๆ เช่น OAuth2, HTTP Basic, API Key

# main.py (ต่อจากโค้ดเดิม)
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
from typing import Annotated

# กำหนด secret key และ algorithm สำหรับ JWT
SECRET_KEY = "your-secret-key" # ควรเก็บใน environment variable
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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=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: Annotated[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
        # ในความเป็นจริงควรดึง user จาก database
        user_data = {"username": username, "full_name": "Test User"}
        return user_data
    except JWTError:
        raise credentials_exception

@app.post("/token")
async def login_for_access_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    # จำลองการตรวจสอบ username/password
    if form_data.username != "testuser" or form_data.password != "password":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Incorrect username or password"
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me/")
async def read_users_me(current_user: Annotated[dict, Depends(get_current_user)]):
    """
    แสดงข้อมูลผู้ใช้ที่ล็อกอินอยู่
    """
    return current_user

โค้ดนี้แสดงแนวคิดพื้นฐานของการทำ OAuth2 with JWT:

  • oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") กำหนดว่า Client จะต้องส่ง Bearer Token มาใน Header
  • login_for_access_token เป็น Endpoint สำหรับ User login เพื่อรับ Access Token
  • get_current_user เป็น Dependency Function ที่จะถอดรหัส JWT และตรวจสอบความถูกต้องของ Token
  • read_users_me เป็น Protected Route ที่ใช้ get_current_user เพื่อยืนยันตัวตนก่อนเข้าถึง

การนำไปใช้งานจริงจะต้องมีการจัดการ User ในฐานข้อมูล และใช้ไลบรารีสำหรับแฮชรหัสผ่าน เช่น passlib ครับ

การเชื่อมต่อฐานข้อมูล: SQLAlchemy และ Pydantic

FastAPI ไม่ได้กำหนดว่าต้องใช้ ORM (Object-Relational Mapper) หรือฐานข้อมูลชนิดใด คุณสามารถใช้ SQLAlchemy, Tortoise ORM, GINO หรือจะใช้ Raw SQL ก็ได้ครับ ในที่นี้เราจะใช้ SQLAlchemy ซึ่งเป็น ORM ยอดนิยมร่วมกับ Pydantic และ SQLite เพื่อความง่ายในการสาธิตครับ

ตั้งค่าฐานข้อมูล (SQLite)

สร้างไฟล์ database.py:

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

# สร้าง SQLite database file
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" # จะสร้างไฟล์ sql_app.db ในโฟลเดอร์เดียวกัน

# สำหรับ SQLite, check_same_thread=False จำเป็นสำหรับการทำงานแบบ multi-thread
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()

สร้างไฟล์ schemas.py สำหรับ Pydantic Models (สามารถรวมกับ models.py เดิมได้ แต่แยกเพื่อความชัดเจน):

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

class ItemBase(BaseModel):
    title: str
    description: Optional[str] = None

class ItemCreate(ItemBase):
    pass

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

    class Config:
        orm_mode = True # สำคัญมากสำหรับ SQLAlchemy ORM

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = [] # User สามารถมีหลาย Items

    class Config:
        orm_mode = True

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

# crud.py
from sqlalchemy.orm import Session
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
from sqlalchemy.orm import relationship

from database import Base # นำเข้า Base จาก database.py
import schemas # นำเข้า schemas.py

# กำหนด SQLAlchemy Models
class DBUser(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("DBItem", back_populates="owner")

class DBItem(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("DBUser", back_populates="items")

# ฟังก์ชัน CRUD สำหรับ User
def get_user(db: Session, user_id: int):
    return db.query(DBUser).filter(DBUser.id == user_id).first()

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

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

def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed" # ควรใช้ passlib ในการแฮช
    db_user = DBUser(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# ฟังก์ชัน CRUD สำหรับ Item
def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(DBItem).offset(skip).limit(limit).all()

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

แก้ไข main.py เพื่อรวมการทำงานของฐานข้อมูล:

# main.py (อัปเดตใหม่ทั้งหมดในส่วนนี้)
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

import crud, models, schemas # นำเข้า crud, models, schemas
from database import SessionLocal, engine, get_db # นำเข้า database components

# สร้างตารางในฐานข้อมูล (ครั้งแรก)
models.Base.metadata.create_all(bind=engine)

app = FastAPI()

@app.post("/users/", response_model=schemas.User)
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)

@app.get("/users/", response_model=List[schemas.User])
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

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    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

@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)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)

@app.get("/items/", response_model=List[schemas.Item])
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

# Path Operation เดิม (Hello World) สามารถลบทิ้งได้หากไม่ต้องการ
@app.get("/")
async def read_root():
    return {"message": "Welcome to FastAPI with SQLAlchemy!"}

ในโค้ดนี้:

  • models.Base.metadata.create_all(bind=engine) จะสร้างตารางในฐานข้อมูล SQLite (sql_app.db) ให้โดยอัตโนมัติเมื่อแอปพลิเคชันเริ่มทำงาน
  • get_db เป็น Dependency Function ที่จะเปิด Database Session และปิดเมื่อ Request จบลง
  • เราใช้ schemas.py สำหรับ Pydantic Models และ crud.py สำหรับ Logic การโต้ตอบกับฐานข้อมูล
  • response_model ถูกใช้เพื่อรับประกันว่าข้อมูลที่ส่งออกไปเป็นไปตาม Pydantic Model ที่กำหนดไว้

นี่คือโครงสร้างพื้นฐานสำหรับการเชื่อมต่อฐานข้อมูลใน FastAPI ที่เป็นระเบียบและ Scalable ครับ

Error Handling: การจัดการข้อผิดพลาดที่กำหนดเอง

FastAPI จัดการข้อผิดพลาดมาตรฐานได้ดีอยู่แล้ว แต่คุณอาจต้องการจัดการข้อผิดพลาดในลักษณะเฉพาะของคุณเอง

Custom Exception Handlers

# main.py (ต่อจากโค้ดเดิม)
from starlette.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

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

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

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

# การจัดการ HTTPException มาตรฐาน (ถ้าต้องการ override)
@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail, "custom_message": "An error occurred!"}
    )

คุณสามารถสร้าง Custom Exception ของคุณเอง แล้วใช้ @app.exception_handler() เพื่อกำหนดว่าเมื่อเกิด Exception นั้นๆ จะให้ Response กลับไปอย่างไรครับ

APIRouter: การจัดโครงสร้างโปรเจกต์ขนาดใหญ่

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

สร้างไฟล์ routers/items.py:

# routers/items.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

import crud, schemas
from database import get_db

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

@router.post("/", response_model=schemas.Item)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    return crud.create_user_item(db=db, item=item, user_id=1) # สมมติ user_id=1

@router.

ผมขอโทษด้วยครับ โค้ดถูกตัดไปเล็กน้อยเนื่องจากความยาวที่เพิ่มขึ้นอย่างรวดเร็ว ผมจะแก้ไขและต่อเติมส่วนที่ขาดหายไปเพื่อให้สมบูรณ์ และควบคุมความยาวของแต่ละส่วนให้เหมาะสมกับเป้าหมาย 3,000-4,000 คำครับ

มาต่อกันที่ไฟล์ routers/items.py และ routers/users.py:

# routers/items.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

import crud, schemas
from database import get_db

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

@router.post("/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
def create_item(item: schemas.ItemCreate, db: Session = Depends(get_db)):
    # ในความเป็นจริง ควรตรวจสอบ User ที่สร้าง Item ด้วย
    # ตอนนี้สมมติว่าสร้างโดย User ID 1
    return crud.create_user_item(db=db, item=item, user_id=1)

@router.get("/", response_model=List[schemas.Item])
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

@router.get("/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
    item = crud.get_item(db, item_id=item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    success = crud.delete_item(db, item_id=item_id)
    if not success:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"message": "Item deleted successfully"}

# เพิ่มฟังก์ชัน get_item และ delete_item ใน crud.py
# crud.py (เพิ่มฟังก์ชันเหล่านี้)
# def get_item(db: Session, item_id: int):
#     return db.query(DBItem).filter(DBItem.id == item_id).first()

# def delete_item(db: Session, item_id: int):
#     item = db.query(DBItem).filter(DBItem.id == item_id).first()
#     if item:
#         db.delete(item)
#         db.commit()
#         return True
#     return False

สร้างไฟล์ routers/users.py:

# routers/users.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session

import crud, schemas
from database import get_db

router = APIRouter(
    prefix="/users",
    tags=["Users"],
)

@router.post("/", response_model=schemas.User, status_code=status.HTTP_201_CREATED)
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)

@router.get("/", response_model=List[schemas.User])
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

@router.get("/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    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

@router.post("/{user_id}/items/", response_model=schemas.Item, status_code=status.HTTP_201_CREATED)
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)

สุดท้าย อัปเดต main.py เพื่อลงทะเบียน Router เหล่านี้:

# main.py (อัปเดตส่วนการลงทะเบียน Router)
from fastapi import FastAPI
import models
from database import engine
from routers import users, items # นำเข้า routers

models.Base.metadata.create_all(bind=engine)

app = FastAPI(
    title="SiamLancard FastAPI Tutorial",
    description="A comprehensive guide to building REST APIs with FastAPI.",
    version="0.0.1",
    docs_url="/documentation", # เปลี่ยน URL สำหรับ Swagger UI
    redoc_url=None, # ปิด ReDoc หากไม่ต้องการ
)

# ลงทะเบียน Routers
app.include_router(users.router)
app.include_router(items.router)

@app.get("/")
async def read_root():
    return {"message": "Welcome to FastAPI with SQLAlchemy! Check /documentation for API docs."}

ตอนนี้คุณมีโครงสร้างโปรเจกต์ที่ Scalable และเป็นระเบียบแล้วครับ การจัดระเบียบด้วย APIRouter ช่วยให้แต่ละส่วนของ API พัฒนาได้โดยอิสระ และจัดการ Dependencies ได้ง่ายขึ้นครับ

Background Tasks: ทำงานเบื้องหลังหลังการตอบกลับ

บางครั้งคุณอาจต้องการให้ API ตอบกลับ Client ทันที แต่มีงานบางอย่างที่ต้องทำต่อในเบื้องหลัง เช่น ส่งอีเมลแจ้งเตือน หรือประมวลผลข้อมูลขนาดใหญ่ FastAPI มีระบบ Background Tasks สำหรับสิ่งนี้ครับ

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

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

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}

ในตัวอย่างนี้ ฟังก์ชัน write_notification จะถูกเรียกใช้เป็น Background Task หลังจากที่ send_notification คืนค่า Response กลับไปให้ Client แล้วครับ

เจาะลึก Pydantic: สุดยอดเครื่องมือ Data Validation

Pydantic เป็นไลบรารีที่ FastAPI ใช้ในการจัดการข้อมูลทั้งหมด ไม่ว่าจะเป็นการตรวจสอบความถูกต้องของข้อมูลขาเข้า (Request Body, Path Parameters, Query Parameters) หรือการแปลงข้อมูลขาออก (Response Body) มาดูความสามารถของมันกันครับ

ประเภทข้อมูลพื้นฐานและ Optional Fields

Pydantic ใช้ Python Type Hints ในการกำหนดโครงสร้างข้อมูลอย่างง่ายดาย

# schemas.py (ตัวอย่างเพิ่มเติม)
from typing import Optional, List
from pydantic import BaseModel, Field, EmailStr, HttpUrl

class UserProfile(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    age: int = Field(..., gt=0, le=120) # ต้องมากกว่า 0 และน้อยกว่าหรือเท่ากับ 120
    email: EmailStr # ตรวจสอบรูปแบบอีเมลอัตโนมัติ
    website: Optional[HttpUrl] = None # Optional และต้องเป็น URL ที่ถูกต้อง

class Product(BaseModel):
    product_id: str
    name: str
    description: Optional[str] = None
    price: float
    tags: List[str] = [] # List ของ String, ค่าเริ่มต้นเป็น List ว่าง

ใน Pydantic:

  • Field(...) หมายถึง Field นั้นๆ เป็น Required (ต้องมี)
  • Field(None) หรือ Optional[...] = None หมายถึง Field นั้นเป็น Optional (ไม่บังคับ)
  • สามารถกำหนดเงื่อนไขการตรวจสอบเพิ่มเติมได้ เช่น min_length, max_length, gt (greater than), le (less than or equal)
  • มี Type พิเศษอย่าง EmailStr, HttpUrl ที่ Pydantic จะตรวจสอบรูปแบบให้โดยอัตโนมัติ

Nested Models: การจัดการข้อมูลซับซ้อน

Pydantic สามารถจัดการข้อมูลที่มีโครงสร้างซ้อนกันได้อย่างง่ายดาย

# schemas.py (ตัวอย่างเพิ่มเติม)

class Address(BaseModel):
    street: str
    city: str
    zip_code: str

class Order(BaseModel):
    order_id: str
    customer_name: str
    delivery_address: Address # Field นี้เป็น Pydantic Model อีกตัว
    products: List[Product] # List ของ Pydantic Model Product
    total_amount: float
    is_delivered: bool = False

เมื่อคุณใช้ Order เป็น Request Body FastAPI จะตรวจสอบความถูกต้องของข้อมูล delivery_address และแต่ละ product ใน products list ให้โดยอัตโนมัติทั้งหมดครับ

Custom Validators: สร้างเงื่อนไขการตรวจสอบเอง

ในบางกรณี คุณอาจต้องการสร้างเงื่อนไขการตรวจสอบที่ซับซ้อนกว่า Type Hints หรือ Field Validators ทั่วไป คุณสามารถใช้ @validator decorator ของ Pydantic ได้ครับ

# schemas.py (ตัวอย่างเพิ่มเติม)
from pydantic import BaseModel, ValidationError, validator

class UserRegistration(BaseModel):
    username: str
    password: str
    password_confirm: str

    @validator('password_confirm')
    def passwords_match(cls, v, values, **kwargs):
        if 'password' in values and v != values['password']:
            raise ValueError('passwords do not match')
        return v

# ลองทดสอบ
try:
    UserRegistration(username='test', password='abc', password_confirm='def')
except ValidationError as e:
    print(e.errors())
# Output: [{'loc': ('password_confirm',), 'msg': 'passwords do not match', 'type': 'value_error'}]

try:
    UserRegistration(username='test', password='abc', password_confirm='abc')
except ValidationError as e:
    print(e.errors()) # จะไม่เกิด Error

ในตัวอย่างนี้ @validator('password_confirm') จะตรวจสอบว่า password_confirm ตรงกับ password หรือไม่ การใช้ Pydantic ช่วยให้คุณมั่นใจได้ว่าข้อมูลที่เข้าสู่ระบบของคุณนั้นถูกต้องและเป็นไปตามข้อกำหนดเสมอครับ

การ Deploy FastAPI Application

เมื่อพัฒนา API เสร็จแล้ว ขั้นตอนต่อไปคือการนำไปใช้งานจริง (Deployment) มีหลายวิธีในการ Deploy FastAPI application ครับ

Uvicorn Workers และ Gunicorn

สำหรับการใช้งานจริง คุณไม่ควรใช้ uvicorn main:app --reload เพราะ --reload ใช้สำหรับพัฒนาเท่านั้น บน Production เรามักจะใช้ Uvicorn ร่วมกับ Gunicorn (สำหรับ Linux/macOS) หรือ Hypercorn (สำหรับ Windows) เพื่อจัดการ Worker Processes ครับ

Gunicorn เป็น WSGI HTTP Server ที่มีฟีเจอร์สำหรับจัดการ Worker Processes, Load Balancing และความทนทานต่อข้อผิดพลาด (fault tolerance) เมื่อใช้กับ Uvicorn จะเรียกว่า ASGI Server ครับ

การติดตั้ง:

pip install gunicorn

การรัน:

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

อธิบาย:

  • main:app: ระบุไฟล์และ FastAPI instance
  • --workers 4: กำหนดให้รัน 4 Worker Processes (จำนวน Worker ควรขึ้นอยู่กับจำนวน Core CPU ของ Server)
  • --worker-class uvicorn.workers.UvicornWorker: บอก Gunicorn ให้ใช้ Uvicorn Worker Class เพื่อรัน ASGI application
  • --bind 0.0.0.0:8000: ให้ Server ฟัง Request จากทุก IP Address ที่ Port 8000

วิธีนี้เป็นมาตรฐานในการ Deploy Python ASGI application บน Production ครับ

Dockerization: การบรรจุแอปพลิเคชัน

Docker เป็นเครื่องมือยอดนิยมสำหรับการบรรจุ (packaging) แอปพลิเคชันและ Dependencies ทั้งหมดลงใน Container ทำให้แอปพลิเคชันของคุณสามารถทำงานได้อย่างสม่ำเสมอในทุกสภาพแวดล้อมครับ

สร้างไฟล์ Dockerfile ในโฟลเดอร์โปรเจกต์ของคุณ:

# Dockerfile
FROM python:3.9-slim-buster

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# ใช้ Gunicorn + Uvicorn เพื่อรันแอปพลิเคชัน
CMD ["gunicorn", "main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

สร้างไฟล์ requirements.txt (หากยังไม่มี) โดยใช้คำสั่ง pip freeze > requirements.txt ใน Virtual Environment ที่คุณติดตั้งไลบรารีไว้แล้ว

# requirements.txt (ตัวอย่าง)
fastapi==0.104.1
uvicorn==0.23.2
gunicorn==21.2.0
sqlalchemy==2.0.22
pydantic==2.5.0
pydantic-settings==2.0.3
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4

การสร้างและรัน Docker Image:

docker build -t my-fastapi-app .
docker run -p 8000:8000 my-fastapi-app

ตอนนี้แอปพลิเคชัน FastAPI ของคุณจะรันอยู่ใน Docker Container และสามารถเข้าถึงได้ที่ http://localhost:8000 ครับ

Nginx Reverse Proxy

บน Production คุณมักจะใช้ Nginx หรือ Apache เป็น Reverse Proxy อยู่ข้างหน้า FastAPI application ของคุณครับ Nginx จะช่วยจัดการ HTTPS, Load Balancing, การแคช, การบีบอัดข้อมูล และการให้บริการไฟล์ Static

แนวคิด:

<

บทความที่เกี่ยวข้อง

iCafeForex.com — EA Forex และเครื่องมือเทรด · SiamCafe.net — ชุมชน 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