Asyncio support #610

Activity

1. ugoi

   added

   Request for enhancement

   on Aug 23, 2025
2. waketzheng commented on Sep 24, 2025

   

   waketzheng

   on Sep 24, 2025

   Last edited by waketzheng

   Contributor

   More actions

   @ugoi fastapi depends on anyio, you can do it like this:

   from io import BytesIO
   
   import anyio
   from docxtpl import DocxTemplate
   
   
   async def main() -> None:
       doc = DocxTemplate("my_word_template.docx")
       context = {"company_name": "World company"}
       doc.render(context)
       bio = BytesIO()
       doc.save(bio)
       filename = "generated_doc.docx"
       content = bio.getvalue()
       size = await anyio.Path(filename).write_bytes(content)
       print(f"Write to {filename} with {size=}")
   
   
   anyio.run(main)

   To control the concurrency number:

   import sys
   from io import BytesIO
   
   import anyio
   from asynctor import timeit  #  pip install asynctor
   from docxtpl import DocxTemplate
   
   LIMIT = 5
   
   
   async def new_docx(filepath: anyio.Path, limiter) -> None:
       doc = DocxTemplate("my_word_template.docx")
       context = {"company_name": "World company"}
       doc.render(context)
       bio = BytesIO()
       doc.save(bio)
       content = bio.getvalue()
       async with limiter:
           size = await filepath.write_bytes(content)
           await anyio.sleep(1)  # Sleep 1 second to verify the limiter result
       print(f"Write to {filepath.name} with {size=}")
   
   
   @timeit
   async def main() -> None:
       if sys.argv[1:] and sys.argv[1].isdigit():
           global LIMIT
           LIMIT = int(sys.argv[1])
           print(f"Changed LIMIT to {LIMIT}")
       filename = "generated_doc.docx"     
       async with anyio.create_task_group() as tg:
           f = anyio.Path(filename)
           limiter = anyio.CapacityLimiter(LIMIT)  # limiter must be defined outside the for loop
           for i in range(5):
               p = f.with_name(f.stem + f"_{i}" + f.suffix)
               tg.start_soon(new_docx, p, limiter)
   
   
   anyio.run(main)