()
| 1369 | |
| 1370 | def add_background_task(self, func: Callable, *args: Any, **kwargs: Any) -> None: |
| 1371 | async def _wrapper() -> None: |
| 1372 | try: |
| 1373 | async with self.app_context(): |
| 1374 | await self.ensure_async(func)(*args, **kwargs) |
| 1375 | except Exception as error: |
| 1376 | await self.handle_background_exception(error) |
| 1377 | |
| 1378 | task = asyncio.get_event_loop().create_task(_wrapper()) |
| 1379 | self.background_tasks.add(task) |
nothing calls this directly
no test coverage detected