Điều này hữu ích cho các hoạt động cần xảy ra sau một yêu cầu, nhưng máy khách không thực sự phải đợi hoạt động hoàn tất trước khi nhận được phản hồi
Điều này bao gồm, ví dụ
- Thông báo qua email được gửi sau khi thực hiện một hành động
- Vì kết nối với máy chủ email và gửi email có xu hướng "chậm" [vài giây], bạn có thể trả lời phản hồi ngay lập tức và gửi thông báo email trong nền
- Tài liệu đã qua xử lý
- Ví dụ: giả sử bạn nhận được một tệp phải trải qua một quá trình xử lý chậm, bạn có thể trả về phản hồi là "Đã chấp nhận" [HTTP 202] và xử lý tệp đó ở chế độ nền
Sử dụng from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
5¶
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Đầu tiên, nhập
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
FastAPI sẽ tạo đối tượng loại
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Tạo chức năng nhiệm vụ¶
Tạo một hàm để chạy dưới dạng tác vụ nền
Nó chỉ là một hàm tiêu chuẩn có thể nhận các tham số
Nó có thể là một hàm
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Trong trường hợp này, hàm tác vụ sẽ ghi vào một tệp [mô phỏng gửi email]
Và vì thao tác ghi không sử dụng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Thêm tác vụ nền¶
Bên trong hàm thao tác đường dẫn của bạn, chuyển hàm tác vụ của bạn tới đối tượng tác vụ nền bằng phương thức
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
- Một chức năng nhiệm vụ được chạy trong nền [______16]
- Bất kỳ chuỗi đối số nào sẽ được chuyển đến hàm tác vụ theo thứ tự [
7]from fastapi import BackgroundTasks, FastAPI app = FastAPI[] def write_notification[email: str, message=""]: with open["log.txt", mode="w"] as email_file: content = f"notification for {email}: {message}" 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"} - Bất kỳ đối số từ khóa nào nên được chuyển đến hàm tác vụ [
8]from fastapi import BackgroundTasks, FastAPI app = FastAPI[] def write_notification[email: str, message=""]: with open["log.txt", mode="w"] as email_file: content = f"notification for {email}: {message}" 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"}
Tiêm phụ thuộc¶
Sử dụng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
FastAPI biết phải làm gì trong từng trường hợp và cách sử dụng lại cùng một đối tượng để tất cả các tác vụ nền được hợp nhất với nhau và chạy trong nền sau đó
Trăn 3. 6 trở lênPython 3. 10 trở lên
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Trong ví dụ này, các tin nhắn sẽ được ghi vào tệp
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Nếu có một truy vấn trong yêu cầu, nó sẽ được ghi vào nhật ký trong một tác vụ nền
Và sau đó, một tác vụ nền khác được tạo tại chức năng vận hành đường dẫn sẽ viết một thông báo bằng cách sử dụng tham số đường dẫn
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Chi tiết kỹ thuật¶
Lớp
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Nó được nhập/bao gồm trực tiếp vào FastAPI để bạn có thể nhập nó từ
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Bằng cách chỉ sử dụng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Vẫn có thể sử dụng riêng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Bạn có thể xem thêm chi tiết trong tài liệu chính thức của Starlette về Tác vụ nền
báo trước¶
Nếu bạn cần thực hiện tính toán nền nặng và bạn không nhất thiết cần chạy nó theo cùng một quy trình [ví dụ: bạn không cần chia sẻ bộ nhớ, biến, v.v.], bạn có thể hưởng lợi từ việc sử dụng các công cụ lớn hơn khác như
Chúng có xu hướng yêu cầu cấu hình phức tạp hơn, trình quản lý hàng đợi tin nhắn/công việc, như RabbitMQ hoặc Redis, nhưng chúng cho phép bạn chạy các tác vụ nền trong nhiều quy trình và đặc biệt là trong nhiều máy chủ
Để xem ví dụ, hãy kiểm tra Trình tạo dự án, tất cả chúng đều bao gồm Celery đã được định cấu hình
Nhưng nếu bạn cần truy cập các biến và đối tượng từ cùng một ứng dụng FastAPI hoặc bạn cần thực hiện các tác vụ nền nhỏ [như gửi thông báo qua email], bạn chỉ cần sử dụng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Tóm tắt lại¶
Nhập và sử dụng
from fastapi import BackgroundTasks, FastAPI
app = FastAPI[]
def write_notification[email: str, message=""]:
with open["log.txt", mode="w"] as email_file:
content = f"notification for {email}: {message}"
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"}
Bạn có thể sử dụng async mà không cần chờ Python không?
Quy tắc. không-async-không chờ đợi
. Functions marked async must contain an await or return statement.
Python không đồng bộ hay đồng bộ?
Có hai loại phương thức cơ bản trong Parallels Python API. đồng bộ và không đồng bộ . Khi một phương thức đồng bộ được gọi, nó sẽ hoàn thành việc thực thi trước khi quay lại trình gọi. Một phương thức không đồng bộ bắt đầu một công việc ở chế độ nền và trả lại cho người gọi ngay lập tức.
Không đồng bộ cho vòng lặp Python là gì?
Biểu thức async for được dùng để duyệt một trình vòng lặp không đồng bộ . Nó là một câu lệnh vòng lặp for không đồng bộ. Một trình lặp không đồng bộ là một trình vòng lặp mang lại các giá trị có thể chờ đợi. trình lặp không đồng bộ. Một đối tượng triển khai các phương thức __aiter__[] và __anext__[].
Làm cách nào bạn có thể thực hiện tác vụ không đồng bộ trong Python?
Trong Python, chúng ta có thể làm điều này bằng cách sử dụng từ khóa async trước định nghĩa hàm . Việc thực hiện coroutine này dẫn đến một đối tượng coroutine.