异步API¶
本页面介绍如何使用异步API。
基本用法¶
异步创建工作簿¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
async def create_workbook():
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("异步示例")
await sheet.set_cell_value("A1", "Hello Async")
await wb.save("async_output.xlsx")
asyncio.run(create_workbook())
异步操作¶
异步设置值¶
async def fill_data():
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("数据")
for row in range(1, 1001):
await sheet.set_cell_value(f"A{row}", f"Row {row}")
await wb.save("large_data.xlsx")
asyncio.run(fill_data())
并发操作¶
并发创建多个文件¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
async def create_report(name, data):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet(name)
for idx, value in enumerate(data, start=1):
await sheet.set_cell_value(f"A{idx}", value)
await wb.save(f"{name}.xlsx")
async def main():
reports = [
("report1", [1, 2, 3]),
("report2", [4, 5, 6]),
("report3", [7, 8, 9]),
]
tasks = [create_report(name, data) for name, data in reports]
await asyncio.gather(*tasks)
asyncio.run(main())
完整示例¶
示例: 并发生成报表¶
import asyncio
from symphra_excel.async_support import AsyncWorkbook
from symphra_excel.styles import PredefinedStyles
async def generate_sales_report(department, sales_data):
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet(department)
await sheet.set_cell_value("A1", f"{department}销售报表")
await sheet.apply_style("A1", PredefinedStyles.title_style())
headers = ["日期", "销售额", "利润"]
for col_idx, header in enumerate(headers, start=1):
cell = f"{chr(64+col_idx)}2"
await sheet.set_cell_value(cell, header)
await sheet.apply_style(cell, PredefinedStyles.header_style())
for row_idx, data in enumerate(sales_data, start=3):
await sheet.set_cell_value(f"A{row_idx}", data["date"])
await sheet.set_cell_value(f"B{row_idx}", data["sales"])
await sheet.set_cell_value(f"C{row_idx}", data["profit"])
await wb.save(f"{department}_report.xlsx")
async def main():
departments = {
"销售部": [
{"date": "2024-01-01", "sales": 1000, "profit": 300},
{"date": "2024-01-02", "sales": 1500, "profit": 450},
],
"技术部": [
{"date": "2024-01-01", "sales": 800, "profit": 240},
{"date": "2024-01-02", "sales": 900, "profit": 270},
],
}
tasks = [
generate_sales_report(dept, data)
for dept, data in departments.items()
]
await asyncio.gather(*tasks)
asyncio.run(main())
最佳实践¶
✅ 推荐做法¶
async def process_data():
async with AsyncWorkbook() as wb:
sheet = await wb.create_worksheet("数据")
await sheet.set_cell_value("A1", "Hello")
await wb.save("output.xlsx")
tasks = [process_data() for _ in range(5)]
await asyncio.gather(*tasks)