In the fast-paced world of web development, performance is paramount. Efficient caching mechanisms can significantly enhance the responsiveness of your API by reducing redundant computations and database queries. In this article, we’ll explore how to integrate the py-cachify library into a FastAPI application using SQLModel and Redis to implement caching and concurrency control.
Caching is a powerful technique to improve the performance of web applications by storing the results of expensive operations and serving them from a quick-access storage. With py-cachify, we can seamlessly add caching to our FastAPI applications, utilizing Redis for storage. Additionally, py-cachify provides tools for concurrency control, preventing race conditions during critical operations.
In this tutorial, we’ll walk through setting up the py-cachify library in a FastAPI application with SQLModel for ORM and Redis for caching.
Let’s start by setting up our project environment.
Start a new project via poetry:
# create new project poetry new --name app py-cachify-fastapi-demo # enter the directory cd py-cachify-fastapi-demo # point poetry to use python3.12 poetry env use python3.12 # add dependencies poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify
Before we can use py-cachify, we need to initialize it with our Redis clients. We’ll do this using FastAPI’s lifespan parameter.
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url
@asynccontextmanager
async def lifespan(_: FastAPI):
init_cachify(
# Replace with your redis url if it differs
async_client=from_url('redis://localhost:6379/0'),
)
yield
app = FastAPI(lifespan=lifespan)
Inside the lifespan, we:
We’ll create a simple User model to interact with our database.
# app/db.py
from sqlmodel import Field, SQLModel
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
email: str
Set up the database engine and create the tables in the lifespan function:
# app/db.py
# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)
# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel
from .db import engine
@asynccontextmanager
async def lifespan(_: FastAPI):
init_cachify(
async_client=from_url('redis://localhost:6379/0'),
)
# Create SQL Model tables
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
yield
Note: We’re using SQLite for simplicity, but you can use any database supported by SQLAlchemy.
Let’s create endpoints to interact with our User model.
# create new project poetry new --name app py-cachify-fastapi-demo # enter the directory cd py-cachify-fastapi-demo # point poetry to use python3.12 poetry env use python3.12 # add dependencies poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify
Now, let’s cache the results of the read_user endpoint to avoid unnecessary database queries.
The endpoint code will look like this:
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url
@asynccontextmanager
async def lifespan(_: FastAPI):
init_cachify(
# Replace with your redis url if it differs
async_client=from_url('redis://localhost:6379/0'),
)
yield
app = FastAPI(lifespan=lifespan)
With the @cached decorator:
When a user’s data is updated, we need to reset the cache to ensure clients receive the latest information. To make it happen, let’s modify the update_user endpoint.
# app/db.py
from sqlmodel import Field, SQLModel
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
email: str
By calling read_user.reset(user_id=user_id), we:
Underneath, the cached decorator dynamically wraps your function, adding the .reset method. This method mimics the function’s signature, and type, this way it will be either sync or async depending on the original function and will accept the same arguments.
The .reset method uses the same key generation logic defined in the cached decorator to identify which cached entry to invalidate. For example, if your caching key pattern is user-{user_id}, calling await read_user.reset(user_id=123) will specifically target and delete the cache entry for user_id=123.
To prevent race conditions during updates, we’ll use the once decorator to lock the execution of the update endpoint.
# app/db.py
# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)
# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel
from .db import engine
@asynccontextmanager
async def lifespan(_: FastAPI):
init_cachify(
async_client=from_url('redis://localhost:6379/0'),
)
# Create SQL Model tables
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
yield
With once:
Optionally, you can configure @once to raise an exception or return a specific value if the lock is already acquired.
Now it’s time to run and test our app!
1) Start the Redis Server:
Ensure your Redis server is running locally or is accessible remotely. You can start a local Redis server using Docker:
# app/main.py
# Adjust imports
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
from .db import User, engine, session_maker
# Database session dependency
async def get_session():
async with session_maker() as session:
yield session
app = FastAPI(lifespan=lifespan)
@app.post('/users/')
async def create_user(user: User, session: AsyncSession = Depends(get_session)) -> User:
session.add(user)
await session.commit()
await session.refresh(user)
return user
@app.get('/users/{user_id}')
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
return await session.get(User, user_id)
@app.put('/users/{user_id}')
async def update_user(user_id: int, new_user: User, session: AsyncSession = Depends(get_session)) -> User | None:
user = await session.get(User, user_id)
if not user:
return None
user.name = new_user.name
user.email = new_user.email
session.add(user)
await session.commit()
await session.refresh(user)
return user
2) Run the FastAPI Application:
With everything set up, you can launch your FastAPI application using Poetry. Navigate to the root directory of your project and execute the following command:
# app/main.py
# Add the import
from py_cachify import cached
@app.get('/users/{user_id}')
@cached('read_user-{user_id}', ttl=300) # New decorator
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
return await session.get(User, user_id)
3) Testing and Playing with Caching and Locking:
Caching: Add delay (e.g., using asyncio.sleep) in the read_user function to simulate a long-running computation. Observe how the response time drastically improves once the result is cached.
Example:
# create new project poetry new --name app py-cachify-fastapi-demo # enter the directory cd py-cachify-fastapi-demo # point poetry to use python3.12 poetry env use python3.12 # add dependencies poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify
Concurrency and Locking: Similarly, introduce a delay in the update_user function to observe the behavior of locks when concurrent update attempts are made.
Example:
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url
@asynccontextmanager
async def lifespan(_: FastAPI):
init_cachify(
# Replace with your redis url if it differs
async_client=from_url('redis://localhost:6379/0'),
)
yield
app = FastAPI(lifespan=lifespan)
These delays can help you see the effectiveness of caching and locking mechanisms in action, as subsequent reads should be faster due to caching, and concurrent writes to the same resource should be managed effectively through locking.
Now, you can test your endpoints using a tool like Postman or by going to http://127.0.0.1:8000/docs (when the app is running!), and observe the performance improvements and concurrency controls in action.
Enjoy experimenting with your enhanced FastAPI app!
By integrating py-cachify into our FastAPI application, we’ve unlocked a plethora of benefits that enhance both the performance and reliability of our API.
Let’s recap some of the key strengths:
For those eager to explore further, check out py-cachify’s GitHub repository and the official documentation for more in-depth guidance, tutorials, and examples.
You can access the full code for this tutorial on GitHub here. Feel free to clone the repository and play around with the implementation to suit your project’s needs.
If you find py-cachify beneficial, consider supporting the project by giving it a star on GitHub! Your support helps drive further improvements and new features.
Happy coding!
The above is the detailed content of Maximize Your FastAPI Efficiency: Blazingly Fast Implementation of Caching and Locking with py-cachify. For more information, please follow other related articles on the PHP Chinese website!
![[Web front-end] Node.js quick start](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
