Coverage for docs_src/background_tasks/tutorial002_an.py: 100%

1from typing import Union 7 ctx1hijklmn

2 

3from fastapi import BackgroundTasks, Depends, FastAPI 7 ctx1hijklmn

4from typing_extensions import Annotated 7 ctx1hijklmn

5 

6app = FastAPI() 7 ctx1hijklmn

7 

8 

9def write_log(message: str): 7 ctx1hijklmn

10 with open("log.txt", mode="a") as log: 7 ctx1abcdefg

11 log.write(message) 7 ctx1abcdefg

12 

13 

14def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None): 7 ctx1hijklmn

15 if q: 7 ctx1abcdefg

16 message = f"found query: {q}\n" 7 ctx1abcdefg

17 background_tasks.add_task(write_log, message) 7 ctx1abcdefg

18 return q 7 ctx1abcdefg

19 

20 

21@app.post("/send-notification/{email}") 7 ctx1hijklmn

22async def send_notification( 7 ctx1hijklmn

23 email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)] 

24): 

25 message = f"message to {email}\n" 7 ctx1abcdefg

26 background_tasks.add_task(write_log, message) 7 ctx1abcdefg

27 return {"message": "Message sent"} 7 ctx1abcdefg