Sử dụng tính năng tiêm phụ thuộc để truy cập DB và bảo vệ điểm cuối

Chúng tôi đã thiết lập điểm cuối đăng nhập cung cấp JWT cho người dùng khi đăng nhập thành công. Người dùng có thể lưu mã thông báo này trong bộ nhớ cục bộ và hiển thị nó cho giao diện người dùng của chúng tôi dưới dạng tiêu đề Ủy quyền. Các điểm cuối chỉ mong đợi quyền truy cập từ người dùng đã đăng nhập có thể giải mã mã thông báo và tìm ra người yêu cầu. Loại công việc này không bị ràng buộc với một điểm cuối cụ thể, thay vào đó nó được chia sẻ logic được sử dụng trong tất cả các điểm cuối được bảo vệ. Tốt nhất bạn nên thiết lập logic giải mã mã thông báo như một phần phụ thuộc có thể được sử dụng trong bất kỳ trình xử lý yêu cầu nào.

Trong FastAPI-speak, các chức năng hoạt động đường dẫn của chúng tôi (trình xử lý yêu cầu) sau đó sẽ phụ thuộc vào get_current_user. Phần get_current_userphụ thuộc cần có kết nối với cơ sở dữ liệu và kết nối với OAuth2PasswordBearerlogic của FastAPI để lấy mã thông báo. Chúng tôi sẽ giải quyết vấn đề này bằng cách get_current_userphụ thuộc vào các chức năng khác. Bằng cách này, chúng ta có thể định nghĩa các chuỗi phụ thuộc, đây là một khái niệm rất mạnh mẽ.

def get_db():
   """provide db session to path operation functions"""
   try:
    db = SessionLocal()
    yield db
   finally:
    db.close()
def get_current_user(db: Session = Depends(get_db),
                 token: str = Depends(oauth2_scheme)):
   return decode_access_token(db, token)
@app.get("/api/me", response_model=schemas.User)
def read_logged_in_user(current_user: models.User = Depends(get_current_user)):
   """return user settings for current user"""
   return current_user

Người dùng đã đăng nhập CÓ THỂ LÀM VIỆC CẦN LÀM

Trước khi chúng tôi viết các hàm hoạt động đường dẫn cho TODO Tạo, Đọc, Cập nhật, Xóa (CRUD), chúng tôi xác định các hàm trợ giúp sau để thực hiện CRUD thực tế trên db.

def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate):
   todo = models.TODO(text=todo_data.text,
                    completed=todo_data.completed)
   todo.owner = current_user
   db.add(todo)
   db.commit()
   db.refresh(todo)
   return todo
def update_todo(db: Session, todo_data: schemas.TODOUpdate):
   todo = db.query(models.TODO).filter(models.TODO.id == id).first()
   todo.text = todo_data.text
   todo.completed = todo.completed
   db.commit()
   db.refresh(todo)
   return todo
def delete_todo(db: Session, id: int):
   todo = db.query(models.TODO).filter(models.TODO.id == id).first()
   db.delete(todo)
   db.commit()
 
def get_user_todos(db: Session, userid: int):
   return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

Các hàm cấp db này sẽ được sử dụng trong các điểm cuối REST sau:

@app.get("/api/mytodos", response_model=List[schemas.TODO])
def get_own_todos(current_user: models.User = Depends(get_current_user),
              db: Session = Depends(get_db)):
   """return a list of TODOs owned by current user"""
   todos = crud.get_user_todos(db, current_user.id)
   return todos
@app.post("/api/todos", response_model=schemas.TODO)
def add_a_todo(todo_data: schemas.TODOCreate,
           current_user: models.User = Depends(get_current_user),
           db: Session = Depends(get_db)):
   """add a TODO"""
   todo = crud.create_meal(db, current_user, meal_data)
   return todo
@app.put("/api/todos/{todo_id}", response_model=schemas.TODO)
def update_a_todo(todo_id: int,
              todo_data: schemas.TODOUpdate,
              current_user: models.User = Depends(get_current_user),
              db: Session = Depends(get_db)):
   """update and return TODO for given id"""
   todo = crud.get_todo(db, todo_id)
   updated_todo = crud.update_todo(db, todo_id, todo_data)
   return updated_todo
@app.delete("/api/todos/{todo_id}")
def delete_a_meal(todo_id: int,
              current_user: models.User = Depends(get_current_user),
              db: Session = Depends(get_db)):
   """delete TODO of given id"""
   crud.delete_meal(db, todo_id)
   return {"detail": "TODO Deleted"}

Viết thử nghiệm

Hãy viết một vài bài kiểm tra cho API TODO của chúng tôi. FastAPI cung cấp một TestClientlớp dựa trên thư viện Yêu cầu phổ biến và chúng tôi có thể chạy thử nghiệm với Pytest.

Để đảm bảo chỉ những người dùng đã đăng nhập mới có thể tạo VIỆC CẦN LÀM, chúng ta có thể viết một cái gì đó như sau:

from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_unauthenticated_user_cant_create_todos():   todo=dict(text="run a mile", completed=False)
response = client.post("/api/todos", data=todo)
assert response.status_code == 401

Thử nghiệm sau đây kiểm tra điểm cuối đăng nhập của chúng tôi và tạo JWT nếu được xuất trình bằng thông tin đăng nhập hợp lệ.

def test_user_can_obtain_auth_token():
  response = client.post("/api/token", data=good_credentials)
  assert response.status_code == 200
  assert 'access_token' in response.json()
  assert 'token_type' in response.json()

Tổng kết

Chúng tôi đã thực hiện xong một ứng dụng TODO rất đơn giản bằng FastAPI. Bây giờ, bạn đã thấy sức mạnh của các gợi ý kiểu được sử dụng hiệu quả trong việc xác định hình dạng của dữ liệu đến và đi thông qua giao diện REST của chúng tôi. Chúng tôi xác định các lược đồ tại một nơi và giao nó cho FastAPI để áp dụng xác thực và chuyển đổi dữ liệu. Tính năng đáng chú ý khác là tiêm phụ thuộc. Chúng tôi đã sử dụng khái niệm này để đóng gói logic được chia sẻ của việc lấy kết nối cơ sở dữ liệu, giải mã JWT để có được người dùng đang đăng nhập và triển khai OAuth2 đơn giản với mật khẩu và người mang. Chúng tôi cũng đã thấy cách các phụ thuộc có thể được xâu chuỗi với nhau.

Chúng ta có thể dễ dàng áp dụng khái niệm này để thêm các tính năng như truy cập dựa trên vai trò. Bên cạnh đó, chúng tôi đang viết mã ngắn gọn và mạnh mẽ mà không cần tìm hiểu các đặc thù của một khuôn khổ. Nói một cách dễ hiểu, FastAPI là một tập hợp các công cụ mạnh mẽ mà bạn không cần phải học vì chúng chỉ là Python hiện đại. Chúc vui vẻ.