priyanshu-panwar/fastapi-utilities

🎨⚡️🔥 Reusable Utilities for FastAPI

Welcome to my GitHub profile!

I'm Priyanshu, a software engineer passionate about building innovative solutions that make a difference. Currently, I'm part of the dynamic team at Blink Health, where we're revolutionizing healthcare technology to improve access and affordability for all.

🔧 Technologies & Tools

• Languages: C++, Java, Python, JavaScript
• Frameworks: FastAPI, React, React Native, Next.js, Django, Spring Boot

🚀 Experience

• Blink Health (Current)
• Samsung
• Amazon

📫 Let's Connect

• LinkedIn: My Profile
• Email: [email protected]
• Youtube: My Channel

fastapi-utilities's Issues

Not compatible with Pydantic v2 (2.3.0)

I am using pydantic 2.3.0 and it seems it is not compatible with fastapi-utilities 0.2.0.

.on_event('startup') is deprecated

See https://fastapi.tiangolo.com/advanced/events/#alternative-events-deprecated

Can you please provide the current way to add repeat_every() functions?

TypeError: 'module' object is not callable for repeat methods

The examples on how to use the repeat annotations seem to be broken, as repeat.repeat_every and repeat.repeat_at are modules, not objects.

Steps to replicate

1. Set up a new venv for testing:

mkdir -p test_dir; cd test_dir/
python -m venv .venv
source .venv/bin/activate
pip install fastapi fastapi_utilities

1. 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")

1. Run the test app:

uvicorn main:app --reload

1. 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")