Closed Korolev-Oleg closed 2 years ago
solution for synchronous tasks:
using a lambda fundus, you can pass an application object into a recurring task bypassing the limit of fastapi_utils.tasks.NoArgsNoReturnFuncT
import fastapi_utils.tasks
from fastapi import FastAPI
import main_tasks
app = FastAPI()
app.state.db = 'database pool'
app.on_event('startup')(
fastapi_utils.tasks.repeat_every(seconds=5)(lambda: main_tasks.task(app))
)
solution asynchronous tasks:
you need recreate and modify fastapi_utils.tasks.repeat_every.py
https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/tasks.py#L47
def decorator(
func: Union[NoArgsNoReturnAsyncFuncT, NoArgsNoReturnFuncT],
*args,
**kwargs
) -> NoArgsNoReturnAsyncFuncT:
https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/tasks.py#L64
await func(*args, **kwargs) # type: ignore
I need to get
app.state
to retrieve the Postgres pool driver in my periodic task. Because task repeating every 5 sec.How can I do this with the fastapi routing case (when
app.state
is available only inrequest
object)?In this example, I'm forced to create gspread and asyncpg connections every 5 seconds this is a terrible decision.