Create a file called main.py with the example code from the README:
from fastapi import FastAPI
from contextlib import asynccontextmanager
from fastapi_utilities.repeat import repeat_every, repeat_at
@asynccontextmanager
async def lifespan(app: FastAPI):
# --- startup ---
await test()
test2()
yield
# --- shutdown ---
app = FastAPI(lifespan=lifespan)
# Repeat Every Example
@repeat_every(seconds=2)
async def test():
print("test")
# Repeat At Example
@repeat_at(cron="* * * * *")
def test2():
print("test2")
Run the test app:
uvicorn main:app --reload
You will get the following error:
INFO: Will watch for changes in these directories: ['/some/path/test_dir']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [12607] using WatchFiles
Process SpawnProcess-1:
Traceback (most recent call last):
File "/some/path/.pyenv/versions/3.12.1/lib/python3.12/multiprocessing/process.py", line 314, in _bootstrap
self.run()
File "/some/path/.pyenv/versions/3.12.1/lib/python3.12/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/_subprocess.py", line 78, in subprocess_started
target(sockets=sockets)
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/server.py", line 65, in run
return asyncio.run(self.serve(sockets=sockets))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/some/path/.pyenv/versions/3.12.1/lib/python3.12/asyncio/runners.py", line 194, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "/some/path/.pyenv/versions/3.12.1/lib/python3.12/asyncio/runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "uvloop/loop.pyx", line 1517, in uvloop.loop.Loop.run_until_complete
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/server.py", line 69, in serve
await self._serve(sockets)
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/server.py", line 76, in _serve
config.load()
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/config.py", line 433, in load
self.loaded_app = import_from_string(self.app)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/some/path/test_dir/.venv/lib/python3.12/site-packages/uvicorn/importer.py", line 19, in import_from_string
module = importlib.import_module(module_str)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/some/path/.pyenv/versions/3.12.1/lib/python3.12/importlib/__init__.py", line 90, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "<frozen importlib._bootstrap>", line 1387, in _gcd_import
File "<frozen importlib._bootstrap>", line 1360, in _find_and_load
File "<frozen importlib._bootstrap>", line 1331, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 935, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 994, in exec_module
File "<frozen importlib._bootstrap>", line 488, in _call_with_frames_removed
File "/some/path/test_dir/main.py", line 16, in <module>
@repeat_every(seconds=2)
^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'module' object is not callable
Workaround
I was able to get the code to work by adding an extra level to the import statements, so I think this may be an issue with how the repeat module is structured or something.
This code works:
from fastapi import FastAPI
from contextlib import asynccontextmanager
# from fastapi_utilities.repeat import repeat_every, repeat_at # <-- Commented out
from fastapi_utilities.repeat.repeat_at import repeat_at # <-- Changed this
from fastapi_utilities.repeat.repeat_every import repeat_every # <-- Changed this
@asynccontextmanager
async def lifespan(app: FastAPI):
# --- startup ---
await test()
test2()
yield
# --- shutdown ---
app = FastAPI(lifespan=lifespan)
# Repeat Every Example
@repeat_every(seconds=2)
async def test():
print("test")
# Repeat At Example
@repeat_at(cron="* * * * *")
def test2():
print("test2")
The examples on how to use the repeat annotations seem to be broken, as
repeat.repeat_every
andrepeat.repeat_at
are modules, not objects.Steps to replicate
Set up a new venv for testing:
Create a file called
main.py
with the example code from the README:Run the test app:
You will get the following error:
Workaround
I was able to get the code to work by adding an extra level to the import statements, so I think this may be an issue with how the repeat module is structured or something.
This code works: