FastAPI: Tạo API chuẩn và cực nhanh trong Python

Từ khi biết tới OpenAPI (Swagger) là mình đã mê cách viết tài liệu API có khả năng tương tác. Cực kỳ rõ ràng, xúc tích, dễ hiểu, chỉ cần gởi đường link về tài liệu API là đối tác hay lập trình viên đều có thể tương tác với API, theo cách không thể tuyệt vời hơn. Và càng thích thú hơn nữa khi mình biết tới FastAPI, tạo API chuẩn, đầy đủ tính năng ngon và cực nhanh lại còn tạo ra tài liệu chuẩn OpenAPI. Bài viết này sẽ giới thiệu về FastAPI, framework tạo API trên Python thuộc nhóm nhanh nhất hiện nay.

Giới thiệu FastAPI

FastAPI là nền tảng thiết kế, lập trình xây dựng API cực kỳ nhanh trên cả 2 phương diện phát triển và thực thi trên Python 3.6+.

Trang chủ: FastAPI

Từ Python 3.6+ thì bạn đã có thể sử dụng cú pháp await/async để chạy code bất đồng bộ, vì lý do này các framework trên Python sẽ đạt được hiệu năng cao, FastAPI là một trong số Python framework nhanh nhất hiện nay.

Với một thời gian tìm hiểu kha khá, mình xác định dự án tới về API sẽ sử dụng FastAPI.

Các tính năng chính

• Nhanh: hiệu suất thực thi rất cao, có thể cạnh tranh được với Nodejs và Go
• Phát triển nhanh: nâng cao tốc độ lập trình từ 200% đến 300%
• Ít lỗi: giảm 40% lỗi do tích hợp ràng buộc dữ liệu khá tốt
• Trực quan: hỗ trợ bộ soạn thảo tốt với Completion bao phủ.
• Dễ dàng: được thiết kế để dễ dàng học và sử dụng
• Ngắn gọn: tối giản hoá sự trùng lặp code.
• Vững chãi: code đạt mức production, với khả năng tương tác API trên tài liệu
• Quy chuẩn: thiết kế hoàn toàn dựa trên OpenAPI và JSON schema

Trên đây là các yếu tố được quảng cáo bởi chính FastAPI, tuy nhiên mình cũng xác nhận rằng 60~70% là có thể tin được. Còn tốc độ lập trình và ít lỗi thì còn phụ thuộc vào khả năng code nữa.

Điểm hay nữa của FastAPI chính là tài liệu đầy đủ và ví dụ rõ ràng, dễ hiểu phủ toàn bộ về cách xử lý các thành phần trong request, cho đến xử lý form, tập tin, và bảo mật.

Thực hành viết API

Chúng ta sẽ thử viết api đơn giản thao tác với db sqlite.

Môi trường và thư viện đi kèm

python3.6 -m venv venv
source venv/bin/activate
pip install sqlalchemy uvicorn fastapi

Viết code

File db.py: mục đích là để tạo ra db SQLite

# db.py
from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base


SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"
# SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False}, echo=True
)

Base = declarative_base()

# Định nghĩa table Todo
class Todo(Base):
    __tablename__ = 'todos'
    id = Column('id', Integer, primary_key = True)
    title = Column('title', String(200))
    done = Column('done', Boolean, default=False)

# Tạo table Todo
Base.metadata.create_all(bind=engine)

Chạy python db.py, nếu không có lỗi gì hết thì file “test.db” sẽ được tạo ra.

Mở file “test.db” bằng DB Browser for SQLite:

File main.py

# main.py
from fastapi import Depends, FastAPI
from fastapi import FastAPI, Path, Body, Depends
from sqlalchemy.orm import Session, sessionmaker
from starlette.requests import Request
from db import Todo, engine

# Connect when a session class instance for DB connection is created
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Definition of data passed to API using Pydantic Addition of Validation and Documentation functions
class TodoIn(BaseModel):
    title: str
    done: bool


# Utility to get a single Todo
def get_todo(db_session: Session, todo_id: int):
    return db_session.query(Todo).filter(Todo.id == todo_id).first()


# Pass the session of DB connection to the function of each endpoint
def get_db(request: Request):
    return request.state.db


# List Todo
@app.get("/todos/")
def read_todos(db: Session = Depends(get_db)):
    todos = db.query(Todo).all()
    return todos


# Get a Todo by id
@app.get("/todos/{todo_id}")
def read_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    return todo


# Create a Todo
@app.post("/todos/")
async def create_todo(todo_in: TodoIn, db: Session = Depends(get_db)):
    todo = Todo(title=todo_in.title, done=False)
    db.add(todo)
    db.commit()
    todo = get_todo(db, todo.id)
    return todo


# Update a Todo
@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, todo_in: TodoIn, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    todo.title = todo_in.title
    todo.done = todo_in.done
    db.commit()
    todo = get_todo(db, todo_id)
    return todo


# Delete a Todo
@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int, db: Session = Depends(get_db)):
    todo = get_todo(db, todo_id)
    db.delete(todo)
    db.commit()


# Create a session instance for middleware DB connection which will be called for each request
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    request.state.db = SessionLocal()
    response = await call_next(request)
    request.state.db.close()
    return response

Chạy ứng dụng với lệnh uvicorn main:app --reload, sau đó bạn có thể sử dụng API ở http://127.0.0.1:8000/todos.

Điểm đặc biệt như đã nói ở trên, là bạn có thể xem và thao tác với API thông qua trang tài liệu của chính ứng dụng ở http://127.0.0.1:8000/docs, hoặc http://127.0.0.1:8000/redoc.

Fullstack Station Tips

Vài điểm cần lưu ý:

• Về Graphql: Graphql được dựa chính trên Starlette và Graphene, vì vậy nếu dự án thiên về Graphql thì FastAPI không mang lại nhiều ý nghĩa. Nếu dùng graphql trên python, bạn có thể sử dụng Graphene và kết hợp với bất kỳ nền tảng nào cũng được như Flask, Django.
• Như đã từng giới thiệu về Graphql (xem Graphql là gì), thì API Restful thực sự có nhiều điểm bất lợi hơn, bạn hãy cân nhắc kỹ ứng dụng của mình nên sử dụng Restful hay Graphql
• Để kích hoạt hiệu suất cao nhất, bạn cần sử dụng driver async tương ứng với DB sử dụng. Ví dụ với MongoDB thì nên dùng https://github.com/mongodb/motor.

Tham khảo:

https://www.toptal.com/python/build-high-performing-apps-with-the-python-fastapi-framework