When you use FastAPI, you might be tempted to create sync (def
) dependencies, on which you actually don't perform thread blocking operations.
The thing is that FastAPI will always run your sync dependencies in a thread pool, which is not always necessary.
The goal of this package is to make explicit if you want to run a dependency in a thread pool.
The package is available on PyPI:
pip install fastapi-dependency
This package is really small and contains simple functions:
Signature:
Depends(dependency: Callable[..., Any] | None = None, *, use_cache: bool = True, use_thread_pool: bool | None = None)
This function is a drop-in replacement for fastapi.Depends
and it has the same signature.
The only difference is that it has an extra parameter: use_thread_pool
.
If you want to run a dependency in a thread pool, you can set use_thread_pool
to True
.
from fastapi import FastAPI
from fastapi_dependency import Depends
app = FastAPI()
def dependency():
return "Hello World!"
@app.get("/")
def index(message: str = Depends(dependency, use_thread_pool=True)):
return {"message": message}
If you don't set use_thread_pool
on sync dependencies, it will raise a RuntimeError
.
Signature:
ThreadDepends(dependency: Callable[..., Any] | None = None, *, use_cache: bool = True)
This function is a drop-in replacement for fastapi.Depends
and it has the same signature.
The only difference is that it will always run the dependency in a thread pool.
from fastapi import FastAPI
from fastapi_dependency import ThreadDepends
app = FastAPI()
def dependency():
return "Hello World!"
@app.get("/")
def index(message: str = ThreadDepends(dependency)):
return {"message": message}
Signature:
ThreadlessDepends(dependency: Callable[..., Any] | None = None, *, use_cache: bool = True)
This function is a drop-in replacement for fastapi.Depends
and it has the same signature.
The only difference is that it will never run the dependency in a thread pool.
from fastapi import FastAPI
from fastapi_dependency import ThreadlessDepends
app = FastAPI()
def dependency():
return "Hello World!"
@app.get("/")
def index(message: str = ThreadlessDepends(dependency)):
return {"message": message}
The analogous functions for fastapi.Security
are:
Security
ThreadSecurity
ThreadlessSecurity
This project is licensed under the terms of the MIT license.