🎨⚡️🔥 Reusable Utilities for FastAPI
Features:
- Repeat At / Every for scheduling cron jobs
- TTL LRU Cache
- Timing Middleware that logs you the time taken by a request
- Session Middleware
- CLI tool to generate skeleton
Source Code: https://github.com/priyanshu-panwar/fastapi-utilities
Youtube Link: Click Here
Inspired From: dmontagu/fastapi-utils
With the latest FastAPI version, on_event
lifespan functions are depreceated. Here is the official doc.
We need to make use of asynccontextmanager
with the latest fastapi.
Here is an example how to use lifespan (Repeated Tasks) functions with latest fastapi:
from fastapi import FastAPI
from contextlib import asynccontextmanager
from fastapi_utilities 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")
Only difference is to call our tasks from lifespan function instead of using on_event
function.
We have introduced ttl_lru_cache
now in our library.
from fastapi_utilities import ttl_lru_cache
@ttl_lru_cache(ttl=2, max_size=128)
def sum(a: int, b: int) -> int:
return a + b
sum(1, 3)
sum(1, 3)
With our CLI Tool you can get a skeleton project built to get you started with the code.
- Using
poetry
:poetry run cli init
- Using
pip
:python3 -m cli init
This package includes a number of utilities to help reduce boilerplate and reuse common functionality across projects:
- 🕒Repeated Tasks: Easily trigger periodic tasks on server startup using repeat_every.
from fastapi_utilities import repeat_every
@router.on_event('startup')
@repeat_every(seconds=3)
async def print_hello():
print("hello")
- 👷Cron Jobs: Easily trigger cron jobs on server startup using repeat_at by providing a cron expression.
from fastapi_utilities import repeat_at
@router.on_event("startup")
@repeat_at(cron="*/2 * * * *") #every 2nd minute
async def hey():
print("hey")
- 🕒Timer Middleware: Add a middleware to the FastAPI app that logs the time taken to process a request. Optionally, also logs the average response time.The average response time is reset after every (reset_after)100,000 requests.
import asyncio
from fastapi import FastAPI, Request
from fastapi_utilities import add_timer_middleware
app = FastAPI()
add_timer_middleware(app, show_avg=True)
@app.get("/")
def read_root():
return {"message": "Hello, World!"}
Response Logs:
INFO: (fastapi-utilities) "GET - /" :: Time Taken :: 0.97 ms
INFO: :: Average Response Time :: 0.97 ms
- Cached Sessions: Now use cached sessions along with context manager instead of
get_db
.
from fastapi import FastAPI
from .db import Base, engine
from fastapi_utilities import FastAPISessionMaker, repeat_every
from .models import User
import random
app = FastAPI()
Base.metadata.create_all(bind=engine)
session_maker = FastAPISessionMaker("sqlite:///db.sqlite3")
@app.on_event("startup")
@repeat_every(seconds=5, raise_exceptions=True)
async def startup():
print("Starting up...")
with session_maker.context_session() as session:
x = User(id=random.randint(0, 10000))
session.add(x)
print("Startup complete!")
This package is intended for use with any recent version of FastAPI and Python 3.7+.
pip install fastapi-utilities
This project is licensed under the terms of the MIT license.