让我们举个例子，在启动时下载10GB的文件（https://speed.hetzner.de/10GB.bin）。
当应用程序启动时，它使用aiohttp触发一个异步下载任务，从https://speed.hetzner.de/10GB.bin获取一个文件并将其保存为downloaded_file。
下载是分块进行的，此后台进程允许应用程序启动其他任务并响应传入的请求，而无需等待下载完成。

import asyncio
from fastapi import FastAPI
import aiohttp

app = FastAPI()

async def download_large_file():
    async with aiohttp.ClientSession() as session:
        url = "https://speed.hetzner.de/10GB.bin"
        async with session.get(url) as response:
            if response.status == 200:
                with open('downloaded_file', 'wb') as file:
                    while True:
                        chunk = await response.content.read(1024)
                        if not chunk:
                            break
                        file.write(chunk)

@app.on_event("startup")
async def startup_event():
    loop = asyncio.get_event_loop()
    loop.create_task(download_large_file())

希望这段代码能有所帮助。

这个答案从以前回答的问题中派生代码和信息。因此，请查看以下答案以了解更多详细信息和解释：

1. How to initialise a global object or variable and reuse it in every FastAPI endpoint?
2. What is the proper way to make downstream Https requests inside of Uvicorn/FastAPI?
3. Is having a concurrent.futures.ThreadPoolExecutor call dangerous in a FastAPI endpoint?
4. FastAPI python: How to run a thread in the background?
5. Return File/Streaming response from online video URL in FastAPI
6. FastAPI UploadFile is slow compared to Flask
7. How to download a large file using FastAPI?
8. How to run another application within the same running event loop?
   下面提供的解决方案使用httpx库，它为Python提供了一个强大的HTTP客户端库，一个async API，并支持HTTP/1.1和HTTP/2。aiofiles库还用于处理asyncio应用程序中的文件操作（如将文件写入磁盘）。用于测试解决方案的公共视频（大文件）可以在here中找到。

解决方案一

如果您希望在应用程序中重用HTTP客户端，请使用此解决方案。

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
import asyncio
import aiofiles
import httpx

async def download_large_file(client: httpx.AsyncClient):
    large_file_url = 'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4'
    path = 'save_to/video.mp4'
    req = client.build_request('GET', large_file_url)
    r = await client.send(req, stream=True)
    async with aiofiles.open(path, 'wb') as f:
        async for chunk in r.aiter_raw():
            await f.write(chunk)
    await r.aclose()

    
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialise the Client on startup and add it to the state
    async with httpx.AsyncClient() as client:
        asyncio.create_task(download_large_file(client))
        yield {'client': client}
        # The Client closes on shutdown

app = FastAPI(lifespan=lifespan)

@app.get('/')
async def home():
    return 'Hello World!'

@app.get('/download')
async def download_some_file(request: Request):
    client = request.state.client  # reuse the HTTP client
    req = client.build_request('GET', 'https://www.example.com')
    r = await client.send(req, stream=True)
    return StreamingResponse(r.aiter_raw(), background=BackgroundTask(r.aclose))

方案二

如果您不需要重用HTTP客户端，而只需要在启动时使用它，请使用此解决方案。

from fastapi import FastAPI
from contextlib import asynccontextmanager
import asyncio
import aiofiles
import httpx

async def download_large_file():
    large_file_url = 'http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4'
    path = 'save_to/video.mp4'
    async with httpx.AsyncClient() as client:
        async with client.stream('GET', large_file_url) as r:
            async with aiofiles.open(path, 'wb') as f:
                async for chunk in r.aiter_raw():   
                    await f.write(chunk)

@asynccontextmanager
async def lifespan(app: FastAPI):
    asyncio.create_task(download_large_file())
    yield

app = FastAPI(lifespan=lifespan)

@app.get('/')
async def home():
    return 'Hello World!'