当前位置：首页 > news >正文

Python开发：使用FastAPI创建后端服务

news 2025/9/23 7:34:55

业务场景

API接收的参数是一个下载链接，服务端拿到这个链接后，下载文件，之后解析文件，解析成功后返回解析成功否则解析失败。

项目结构

app/
├── main.py          # 入口
├── api.py           # API 路由
├── services/├── downloader.py  # 下载逻辑└── parser.py      # 解析逻辑

安装依赖

在项目的虚拟环境中安装依赖：
如何创建虚拟环境参考之前的文章：
https://blog.csdn.net/zhang_jiamin/article/details/147442999?spm=1011.2415.3001.5331

pip3 install fastapi uvicorn aiohttp

fastapi → Web 框架
uvicorn → 启动 ASGI 服务器
aiohttp → 异步下载文件（比 requests 更适合异步）

定义 API 路由 (app/api.py)

from fastapi import APIRouter, HTTPException, Form
from fastapi.responses import JSONResponse
from app.services.downloader import download_file
from app.services.parser import parse_filerouter = APIRouter()@router.post("/app/fileParse")
async def parse_file_api(url: str = Form(...)):try:# 1. 下载文件local_path = await download_file(url)# 2. 解析文件apk_url = await parse_file(local_path)return {"apkUrl": apk_url}except Exception as e:raise HTTPException(status_code=500, detail=str(e))

下载逻辑 (app/services/downloader.py)

import aiohttp
import aiofiles
import tempfile
import os
import logginglogger = logging.getLogger(__name__)async def download_file(url: str) -> str:tmp_dir = tempfile.mkdtemp()local_path = os.path.join(tmp_dir, "test_file.file")async with aiohttp.ClientSession() as session:async with session.get(url) as resp:if resp.status != 200:raise Exception(f"下载失败: {resp.status}")logger.info("开始下载文件")async with aiofiles.open(local_path, 'wb') as f:while chunk := await resp.content.read(1024):await f.write(chunk)return local_path

解析逻辑 (app/services/parser.py)

import os
import subprocess
import logginglogger = logging.getLogger(__name__)async def parse_file(local_path: str) -> str:logger.info("开始解析文件")base_dir = os.path.dirname(os.path.abspath(__file__))bundletool_path = os.path.join(base_dir, "..", "..", "bundletool.jar")jks_path = os.path.join(base_dir, "..", "..", "debug.jks")dir_path = os.path.dirname(local_path)apk_dir_path = os.path.join(dir_path, "test_app.apks")cmd = ["java", "-jar", f"{bundletool_path}", "build-apks",f"--bundle={local_path}",f"--output={apk_dir_path}",f"--ks={jks_path}","--ks-pass=pass:1234567","--ks-key-alias=debug","--key-pass=pass:1234567"]result = subprocess.run(cmd, capture_output=True, text=True)if result.returncode == 0:logger.info(f"命令执行成功，returncode={result.returncode}")else:logger.error(f"命令执行失败，returncode={result.returncode}, stderr={result.stderr}")"""上传解析后的文件到s3，返回s3链接，或者直接返回解析后的文件和结果"""apk_url = "https://your-bucket.s3.amazonaws.com/fake-app.apk"return apk_url

FastAPI 入口 (app/main.py)

from fastapi import FastAPI
from app.api import routerapp = FastAPI(title="FILE Parser API")app.include_router(router, prefix="/v1")# 启动服务: uvicorn main:app --reload

启动服务

testmanzhang@loaclhost file_parse % source .venv/bin/activate    
(.venv) testmanzhang@loaclhost file_parse % 
(.venv) testmanzhang@loaclhost file_parse % 
(.venv) testmanzhang@loaclhost file_parse % uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['/Users/testmanzhang/PycharmProjects/file_parse']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [73816] using StatReload
INFO:     Started server process [73819]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:55102 - "GET /docs HTTP/1.1" 200 OK
INFO:     127.0.0.1:55102 - "GET /openapi.json HTTP/1.1" 200 OK