如何部署FastAPI

环境：

Win10

FastAPI

问题描述：

如何部署FastAPI

解决方案：

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，使用 Python 3.6 及更高版本。它的设计目的是提供简单且易于使用的接口，同时利用 Python 的类型提示和异步特性来实现高效的性能。以下是 FastAPI 的一些关键特点和优势：

关键特点

高性能:
- FastAPI 是基于 Starlette（一个用于构建 Web 应用程序的高性能异步框架）构建的，同时也使用了 Pydantic 进行数据验证，因此其性能接近于 Node.js 和 Go 这样的语言构建的框架。
简单易用:
- FastAPI 提供了简单的接口和清晰的文档，使得即使是初学者也能快速上手。通过使用 Python 的类型提示，您可以更容易地定义 API 的输入和输出。
支持异步编程:
- FastAPI 支持异步请求处理，这使得它能够高效地处理大量并发请求，尤其适合 I/O 密集型的应用程序（例如，调用外部 API、数据库操作等）。
自动生成文档:
- FastAPI 自动生成 OpenAPI（以前称为 Swagger）文档，使得用户可以方便地查看 API 的接口文档。可以通过 /docs 访问交互式文档，或者通过 /redoc 访问另一种风格的文档。
强类型支持:
- FastAPI 利用 Python 的类型注释，支持请求体、查询参数和路径参数的类型验证。这有助于减少错误并使代码更加清晰。
依赖注入:
- FastAPI 内置了一套简单而强大的依赖注入系统，使得组织代码和管理复杂依赖关系变得简单。
易于集成和扩展:
- FastAPI 可以与其他 Python 库和框架无缝集成，例如 SQLAlchemy（数据库 ORM）、Redis，以及其他的第三方库。

使用示例

以下是一个简单的 FastAPI 示例，展示了如何定义一个 API，并处理 GET 和 POST 请求：

from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strprice: float@app.get("/")
async def read_root():return {"Hello": "World"}@app.post("/items/")
async def create_item(item: Item):return {"item_name": item.name, "item_price": item.price}

启动 FastAPI 应用

您可以使用 uvicorn 启动 FastAPI 应用：

uvicorn main:app --reload

其中 main 是包含 FastAPI 应用的 Python 文件名（不带扩展名），app 是 FastAPI 实例的名称。添加 --reload 标志以便在代码更改后自动重新加载服务器。

我这操作已火山api为例

1.安装库

pip install fastapi[all]pip install uvicorn

2.写个py文件

from fastapi import FastAPI, HTTPException
import websockets
import json
import gzip
import uuid
from io import BytesIO
from fastapi.responses import Responseapp = FastAPI()# 火山引擎的配置信息
appid = "9250999999713"  # 替换为您的实际appid
token = "y8suQ999999iQCNuduno"  # 替换为您的实际token
cluster = "volcano_tts"
voice_type = "zh_female_wanwanxiaohe_moon_bigtts"
host = "openspeech.bytedance.com"
api_url = f"wss://openspeech.bytedance.com/api/v1/tts/ws_binary"default_header = bytearray(b'\x11\x10\x11\x00')@app.post("/tts")
async def tts(text: str):if not text:raise HTTPException(status_code=400, detail="No text provided")submit_request_json = {"app": {"appid": appid,"token": token,"cluster": cluster},"user": {"uid": "388808087185088"},"audio": {"voice_type": voice_type,"encoding": "mp3","speed_ratio": 1.0,"volume_ratio": 1.0,"pitch_ratio": 1.0,},"request": {"reqid": str(uuid.uuid4()),"text": text,"text_type": "plain","operation": "submit"}}try:async with websockets.connect(api_url, extra_headers={"Authorization": f"Bearer {token}"}) as ws:payload_bytes = json.dumps(submit_request_json).encode()payload_bytes = gzip.compress(payload_bytes)full_client_request = default_header + len(payload_bytes).to_bytes(4, 'big') + payload_bytesawait ws.send(full_client_request)file_to_save = BytesIO()while True:res = await ws.recv()if res:done = parse_response(res, file_to_save)if done:breakelse:breakexcept websockets.exceptions.WebSocketException as e:raise HTTPException(status_code=500, detail=f"WebSocket connection error: {str(e)}")return Response(content=file_to_save.getvalue(), media_type='audio/mp3')def parse_response(res, file):# 解析响应数据的逻辑与官方示例代码相同# ...if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=5100)

3.运行脚本
在这里插入图片描述
4.打开web页面

http api接口火山

import os
import sys
from fastapi import FastAPI, HTTPException, Depends, Query, Depends
from fastapi.responses import StreamingResponse, JSONResponse
from pydantic import BaseModel
from typing import Optional
import io
import asyncio
import base64
import uuid
import logging
import httpx
import argparse
from pydub import AudioSegment# 设置日志配置
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)app = FastAPI()class TTSSettings:def __init__(self, token: str, app_id: str, cluster_id: str, user_id: str):self.token = tokenself.app_id = app_idself.cluster_id = cluster_idself.user_id = user_id# 初始化火山引擎 TTS 设置 (请根据实际情况调整这些值)
tts_settings = TTSSettings(token="y8uduno",  # 确保这里使用的是有效的令牌app_id="9713",cluster_id="volcano_tts",user_id="your_user_id"
)class TTSRequest(BaseModel):text: str  # 只需要提供文本字段async def request_tts(text: str) -> bytes:url = "https://openspeech.bytedance.com/api/v1/tts"headers = {"Authorization": f"Bearer;{tts_settings.token}",  # 注意这里的分号"Content-Type": "application/json"}data = {"app": {"appid": tts_settings.app_id,"token": tts_settings.token,"cluster": tts_settings.cluster_id},"user": {"uid": tts_settings.user_id},"audio": {"voice_type": "zh_female_wanwanxiaohe_moon_bigtts",  # 使用适当的音色类型"encoding": "mp3","speed_ratio": 1.0},"request": {"reqid": str(uuid.uuid4()),  # 每次调用都生成新的唯一 reqid"text": text,"operation": "query"}}logger.debug(f"Sending TTS request with text: {text}")async with httpx.AsyncClient() as client:try:response = await client.post(url, json=data, headers=headers)response.raise_for_status()  # 如果响应状态码不是 2xx，则抛出异常except httpx.HTTPStatusError as exc:logger.error(f"HTTP error occurred while requesting TTS: {exc.response.status_code}, {exc.response.text}")raise HTTPException(status_code=exc.response.status_code, detail=exc.response.text)result = response.json()code = result.get('code')message = result.get('message')logger.debug(f"TTS response received: {result}")if code == 3000:  # 请求成功audio_base64 = result.get('data')if audio_base64:logger.info("Successfully received MP3 audio data.")return base64.b64decode(audio_base64)else:raise Exception("No audio data in the response.")else:raise Exception(f"TTS request failed: {message}")def mp3_to_wav_stream(mp3_data: bytes) -> io.BytesIO:"""Convert MP3 data to WAV format."""logger.debug("Converting MP3 to WAV...")audio_segment = AudioSegment.from_file(io.BytesIO(mp3_data), format="mp3")wav_io = io.BytesIO()audio_segment.export(wav_io, format="wav")wav_io.seek(0)logger.debug("MP3 to WAV conversion completed.")return wav_ioasync def generate_tts_audio(text: str, media_type: str = "wav") -> io.BytesIO:try:mp3_data = await request_tts(text)if mp3_data is not None:output_io = io.BytesIO()if media_type.lower() != "mp3":# 使用 pydub 和 ffmpeg 转换为指定格式wav_stream = mp3_to_wav_stream(mp3_data)audio_segment = AudioSegment.from_file(wav_stream, format="wav")audio_segment.export(output_io, format=media_type)else:# 如果请求的是 MP3 格式，则直接返回原始数据output_io.write(mp3_data)output_io.seek(0)return output_ioelse:raise Exception("Failed to generate TTS audio.")except Exception as e:logger.error(f"Error during TTS generation: {e}")raise@app.post("/tts")
async def process_tts_request(request_data: TTSRequest):try:fixed_media_type = "wav"  # 固定其他参数audio_io = await generate_tts_audio(text=request_data.text,media_type=fixed_media_type)# 定义保存文件的路径和文件名output_dir = "output_audios"  # 输出目录if not os.path.exists(output_dir):os.makedirs(output_dir)  # 如果目录不存在，则创建# 使用 UUID 生成唯一的文件名，避免文件名冲突file_name = f"{uuid.uuid4()}.{fixed_media_type}"file_path = os.path.join(output_dir, file_name)# 将 BytesIO 内容写入文件with open(file_path, 'wb') as f:f.write(audio_io.getvalue())logger.info(f"Audio saved to {file_path}.")return StreamingResponse(audio_io, media_type=f"audio/{fixed_media_type}")except Exception as e:logger.error(f"TTS request failed: {e}")raise HTTPException(status_code=400, detail=str(e))@app.get("/")
async def read_root():"""根路径简单响应"""return {"message": "Welcome to the TTS API"}@app.get("/control")
async def control(command: str = Query(..., description="Command to execute")):if command == "restart":os.execl(sys.executable, sys.executable, *sys.argv)elif command == "exit":os.kill(os.getpid(), signal.SIGTERM)return JSONResponse(content={"message": "Service terminated"}, status_code=200)else:raise HTTPException(status_code=400, detail="Invalid command")if __name__ == "__main__":import uvicornparser = argparse.ArgumentParser(description="VolcanoTTS API")parser.add_argument("-a", "--bind_addr", type=str, default="127.0.0.1", help="Bind address, default: 127.0.0.1")parser.add_argument("-p", "--port", type=int, default=9881, help="Port number, default: 9881")args = parser.parse_args()uvicorn.run(app, host=args.bind_addr, port=args.port)