Trang chủ > Bài viết > Xây dựng API với FastAPI và Python

Xây dựng API với FastAPI và Python

5 phút đọc
Xây dựng API với FastAPI và Python
Tác giả

Tác giả

Chuyên gia lập trình

  <div class="prose prose-lg max-w-none">
    <h1>Xây dựng API với FastAPI và Python</h1>
    
    <p>
      FastAPI là một framework hiện đại, nhanh (hiệu suất cao), dễ học để xây dựng API với Python 3.6+ 
      dựa trên các tiêu chuẩn API mở. Trong bài viết này, chúng ta sẽ tìm hiểu cách xây dựng một RESTful API 
      đơn giản sử dụng FastAPI.
    </p>

    <h2>Tại sao chọn FastAPI?</h2>
    
    <p>FastAPI có nhiều ưu điểm nổi bật:</p>
    
    <ul>
      <li><strong>Nhanh</strong>: Hiệu suất rất cao, ngang với NodeJS và Go</li>
      <li><strong>Nhanh để code</strong>: Tăng tốc độ phát triển khoảng 200% đến 300%</li>
      <li><strong>Ít lỗi</strong>: Giảm khoảng 40% lỗi do con người</li>
      <li><strong>Trực quan</strong>: Hỗ trợ editor với autocompletion mạnh mẽ</li>
      <li><strong>Dễ học</strong>: Thiết kế để dễ sử dụng và học hỏi</li>
      <li><strong>Ngắn gọn</strong>: Giảm thiểu code trùng lặp</li>
      <li><strong>Mạnh mẽ</strong>: Có tính năng tự động tạo tài liệu API</li>
      <li><strong>Dựa trên tiêu chuẩn</strong>: Dựa trên và hoàn toàn tương thích với OpenAPI và JSON Schema</li>
    </ul>

    <div class="my-8">
      <img 
        src="/fastapi-python-banner.png" 
        alt="FastAPI Python Banner" 
        class="rounded-lg shadow-md mx-auto"
      />
      <p class="text-center text-sm text-gray-500 mt-2">FastAPI - Framework hiện đại cho Python</p>
    </div>

    <h2>Cài đặt FastAPI</h2>
    
    <p>Để bắt đầu, chúng ta cần cài đặt FastAPI và Uvicorn (ASGI server):</p>
    
    <pre><code class="language-bash">pip install fastapi uvicorn</code></pre>
    
    <h2>Tạo API đầu tiên</h2>
    
    <p>Hãy tạo một file <code>main.py</code> với nội dung sau:</p>
    
    <pre><code class="language-python">from fastapi import FastAPI

app = FastAPI()

@app.get("/") def read_root(): return {"Hello": "World"}

@app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}</code></pre>

    <p>Để chạy ứng dụng, sử dụng lệnh:</p>
    
    <pre><code class="language-bash">uvicorn main:app --reload</code></pre>
    
    <p>Bây giờ, API của bạn đã chạy tại <code>http://127.0.0.1:8000</code>. Bạn có thể truy cập:</p>
    
    <ul>
      <li><a href="http://127.0.0.1:8000" target="_blank">http://127.0.0.1:8000</a> - Endpoint chính</li>
      <li><a href="http://127.0.0.1:8000/items/5?q=test" target="_blank">http://127.0.0.1:8000/items/5?q=test</a> - Endpoint với path parameter và query parameter</li>
      <li><a href="http://127.0.0.1:8000/docs" target="_blank">http://127.0.0.1:8000/docs</a> - Tài liệu API tự động (Swagger UI)</li>
      <li><a href="http://127.0.0.1:8000/redoc" target="_blank">http://127.0.0.1:8000/redoc</a> - Tài liệu API thay thế (ReDoc)</li>
    </ul>

    <h2>Tạo mô hình dữ liệu với Pydantic</h2>
    
    <p>FastAPI sử dụng Pydantic để xác thực dữ liệu, serialize và deserialize. Hãy tạo một mô hình Pydantic:</p>
    
    <pre><code class="language-python">from fastapi import FastAPI

from pydantic import BaseModel from typing import Optional

app = FastAPI()

class Item(BaseModel): name: str price: float is_offer: Optional[bool] = None

@app.get("/") def read_root(): return {"Hello": "World"}

@app.get("/items/{item_id}") def read_item(item_id: int, q: Optional[str] = None): return {"item_id": item_id, "q": q}

@app.put("/items/{item_id}") def update_item(item_id: int, item: Item): return {"item_name": item.name, "item_id": item_id}</code></pre>

    <h2>Xử lý HTTP Methods</h2>
    
    <p>FastAPI hỗ trợ tất cả các HTTP methods phổ biến:</p>
    
    <pre><code class="language-python">@app.post("/items/")

def create_item(item: Item): return item

@app.get("/items/") def read_items(): return [{"name": "Item 1"}, {"name": "Item 2"}]

@app.put("/items/{item_id}") def update_item(item_id: int, item: Item): return {"item_name": item.name, "item_id": item_id}

@app.delete("/items/{item_id}") def delete_item(item_id: int): return {"deleted": item_id}</code></pre>

    <h2>Xác thực và phân quyền</h2>
    
    <p>FastAPI cung cấp nhiều cách để xác thực người dùng. Dưới đây là một ví dụ đơn giản về xác thực với OAuth2 và JWT:</p>
    
    <pre><code class="language-python">from fastapi import Depends, FastAPI, HTTPException, status

from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from pydantic import BaseModel from typing import Optional import jwt from datetime import datetime, timedelta

Cài đặt bảo mật

SECRET_KEY = "your-secret-key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30

Mô hình dữ liệu

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

class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None

class UserInDB(User): hashed_password: str

Giả lập database người dùng

fake_users_db = { "johndoe": { "username": "johndoe", "full_name": "John Doe", "email": "johndoe@example.com", "hashed_password": "fakehashedsecret", "disabled": False, } }

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

Hàm tiện ích

def verify_password(plain_password, hashed_password): return plain_password + "notreallyhashed" == hashed_password

def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str): user = get_user(fake_db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user

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: 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 except jwt.PyJWTError: raise credentials_exception user = get_user(fake_users_db, username=username) if user is None: raise credentials_exception return user

async def get_current_active_user(current_user: User = Depends(get_current_user)): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user

Endpoints

@app.post("/token", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: 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_active_user)): return current_user</code></pre>

    <h2>Kết nối với cơ sở dữ liệu</h2>
    
    <p>FastAPI không có ORM tích hợp, nhưng bạn có thể sử dụng bất kỳ thư viện Python nào như SQLAlchemy, Tortoise ORM, hoặc thậm chí là MongoDB với Motor. Dưới đây là ví dụ với SQLAlchemy:</p>
    
    <pre><code class="language-python">from fastapi import Depends, FastAPI, HTTPException

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session

Cấu hình SQLAlchemy

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

Mô hình SQLAlchemy

class ItemDB(Base): tablename = "items"

id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String)
price = Column(Integer)

Tạo bảng

Base.metadata.create_all(bind=engine)

Mô hình Pydantic

from pydantic import BaseModel

class ItemBase(BaseModel): name: str description: str = None price: float

class ItemCreate(ItemBase): pass

class Item(ItemBase): id: int

class Config:
    orm_mode = True

Dependency

def get_db(): db = SessionLocal() try: yield db finally: db.close()

app = FastAPI()

@app.post("/items/", response_model=Item) def create_item(item: ItemCreate, db: Session = Depends(get_db)): db_item = ItemDB(**item.dict()) db.add(db_item) db.commit() db.refresh(db_item) return db_item

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

@app.get("/items/{item_id}", response_model=Item) def read_item(item_id: int, db: Session = Depends(get_db)): db_item = db.query(ItemDB).filter(ItemDB.id == item_id).first() if db_item is None: raise HTTPException(status_code=404, detail="Item not found") return db_item</code></pre>

    <h2>Xử lý tệp tin</h2>
    
    <p>FastAPI cũng hỗ trợ tải lên tệp tin:</p>
    
    <pre><code class="language-python">from fastapi import FastAPI, File, UploadFile

from fastapi.responses import HTMLResponse import shutil from typing import List

app = FastAPI()

@app.post("/uploadfile/") async def create_upload_file(file: UploadFile): # Lưu tệp tin tải lên with open(f"uploaded_{file.filename}", "wb") as buffer: shutil.copyfileobj(file.file, buffer)

return {"filename": file.filename}

@app.post("/uploadfiles/") async def create_upload_files(files: List[UploadFile] = File(...)): return {"filenames": [file.filename for file in files]}

@app.get("/") async def main(): content = """ <body> <form action="/uploadfile/" enctype="multipart/form-data" method="post"> <input name="file" type="file"> <input type="submit"> </form> <form action="/uploadfiles/" enctype="multipart/form-data" method="post"> <input name="files" type="file" multiple> <input type="submit"> </form> </body> """ return HTMLResponse(content=content)</code></pre>

    <h2>Xử lý lỗi và ngoại lệ</h2>
    
    <p>FastAPI cho phép bạn xử lý lỗi và ngoại lệ một cách dễ dàng:</p>
    
    <pre><code class="language-python">from fastapi import FastAPI, HTTPException, Request

from fastapi.responses import JSONResponse

app = FastAPI()

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. There goes a rainbow..."}, )

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

@app.get("/items/{item_id}") async def read_item(item_id: int): if item_id == 42: raise HTTPException(status_code=404, detail="Item not found") return {"item_id": item_id}</code></pre>

    <h2>Middleware</h2>
    
    <p>FastAPI hỗ trợ middleware để xử lý các yêu cầu trước khi chúng đến endpoint:</p>
    
    <pre><code class="language-python">import time

from fastapi import FastAPI, Request

app = FastAPI()

@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) return response</code></pre>

    <h2>Tạo tài liệu API</h2>
    
    <p>FastAPI tự động tạo tài liệu API dựa trên OpenAPI. Bạn có thể tùy chỉnh tài liệu này:</p>
    
    <pre><code class="language-python">from fastapi import FastAPI

description = """ ChimichangApp API helps you do awesome stuff. 🚀

Items

You can read items.

Users

You will be able to:

  • Create users
  • Read users """

app = FastAPI( title="ChimichangApp", description=description, version="0.0.1", terms_of_service="http://example.com/terms/", contact={ "name": "Deadpoolio the Amazing", "url": "http://x-force.example.com/contact/", "email": "dp@x-force.example.com", }, license_info={ "name": "Apache 2.0", "url": "https://www.apache.org/licenses/LICENSE-2.0.html", }, )</code></pre>

    <h2>Triển khai FastAPI</h2>
    
    <p>Để triển khai ứng dụng FastAPI trong môi trường sản xuất, bạn có thể sử dụng Uvicorn với Gunicorn:</p>
    
    <pre><code class="language-bash">pip install gunicorn uvicorn</code></pre>
    
    <p>Và chạy với lệnh:</p>
    
    <pre><code class="language-bash">gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker</code></pre>
    
    <p>Hoặc bạn có thể đóng gói ứng dụng trong Docker:</p>
    
    <pre><code class="language-dockerfile">FROM python:3.9

WORKDIR /code

COPY ./requirements.txt /code/requirements.txt

RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt

COPY ./app /code/app

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]</code></pre>

    <h2>Kết luận</h2>
    
    <p>
      FastAPI là một framework mạnh mẽ và hiện đại để xây dựng API với Python. Với hiệu suất cao, 
      tài liệu tự động, và tính năng xác thực kiểu dữ liệu, FastAPI là một lựa chọn tuyệt vời cho 
      các dự án API mới.
    </p>
    
    <p>
      Trong bài viết này, chúng ta đã tìm hiểu các khái niệm cơ bản của FastAPI và cách xây dựng 
      một RESTful API đơn giản. Bạn có thể tìm hiểu thêm trong 
      <a href="https://fastapi.tiangolo.com/" target="_blank">tài liệu chính thức của FastAPI</a>.
    </p>
    
    <div class="bg-blue-50 p-4 rounded-lg my-6">
      <h3 class="text-blue-800 font-semibold">Tài nguyên học tập</h3>
      <ul class="list-disc pl-5 text-blue-700">
        <li><a href="https://fastapi.tiangolo.com/" target="_blank">Tài liệu chính thức FastAPI</a></li>
        <li><a href="https://github.com/tiangolo/fastapi" target="_blank">GitHub repository</a></li>
        <li><a href="https://testdriven.io/blog/topics/fastapi/" target="_blank">Các bài viết về FastAPI trên TestDriven.io</a></li>
      </ul>
    </div>
  </div>