Skip to content

Instantly share code, notes, and snippets.

@nicksonthc

nicksonthc/fastapi-redis.py

Created April 11, 2024 16:12
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save nicksonthc/525742d9a81d3950b443810e8899ee0e to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save nicksonthc/525742d9a81d3950b443810e8899ee0e to your computer and use it in GitHub Desktop.
Download ZIP
FastAPI with asyncio Redis and Lifespan Example
Raw
readme.md

Title

This is an example of a FastAPI application integrated with asyncio Redis using the FastAPI lifespan feature. The application demonstrates how to connect to a Redis database asynchronously during the lifespan of the FastAPI application and perform CRUD operations.

Tried Approach / Method

  1. launch coroutine redis with threading
  2. alternative depreciated fastapi on_start event

References

  1. https://fastapi.tiangolo.com/advanced/events/
  2. https://stackoverflow.com/questions/76322463/how-to-initialise-a-global-object-or-variable-and-reuse-it-in-every-fastapi-endp

Run

python fastapi-redis.py

Swagger

  1. get
  2. post

The Redis class provides methods to connect to Redis, insert a string value, and query a key from Redis. These methods use the redis.asyncio library for asynchronous Redis operations.

The run_redis coroutine function connects to Redis, inserts a string value ('fastapi') with the key 'stackoverflow', queries the value from Redis, and returns the Redis instance.

The lifespan async context manager initializes the Redis connection during the startup of the FastAPI application and cleans up the connection during shutdown. It stores the Redis instance in the FastAPI app state for easy access by request handlers.

Two endpoints are defined: /redis for retrieving the value of the 'stackoverflow' key and /redis POST method for inserting a new value for the 'stackoverflow' key. These endpoints use the Redis instance stored in the app state to interact with Redis.

To run the application, execute the main module using Uvicorn with the specified host and port. The FastAPI application will be served with asynchronous Redis integration enabled.

Raw
fastapi-redis.py
from contextlib import asynccontextmanager
from datetime import datetime
from fastapi import FastAPI, Request
import fastapi
from fastapi.datastructures import State
from fastapi.responses import JSONResponse
import redis.asyncio as redis
class Redis:
redis_client: redis.Redis = None
@classmethod
async def connect(
cls, host: str = "localhost", port: int = 6379, username=None, password=None
):
try:
# Connect to Redis server
cls.redis_client = redis.Redis(
host=host, port=port, username=username, password=password
)
except redis.RedisError as e:
print(f"Failed to connect to Redis: {e}")
raise
await cls.redis_client
@classmethod
async def close(cls):
if cls.redis_client is not None:
await cls.redis_client.aclose()
@classmethod
async def insert_string(cls, key: str, value: str):
await cls.redis_client.set(key, value)
@classmethod
async def query_key(cls, key: str):
value = await cls.redis_client.get(key)
value = value.decode('utf-8')
return value
async def run_redis():
await Redis.connect()
await Redis.insert_string("stackoverflow", 'fastapi')
value = await Redis.query_key("stackoverflow")
print(f"Retrieved value: {value}")
return Redis
@asynccontextmanager
async def lifespan(app: FastAPI):
print("enter lifespan")
# app.state is instance of fastapi.datastructures.State which it base is from starlette.datastructures.State
print(isinstance(app.state,State))
app.state.n_client = await run_redis()
yield
print("exit lifespan")
app = FastAPI(lifespan=lifespan)
@app.get('/redis')
async def get_redis(request: Request):
n_client = request.app.state.n_client
n_client : Redis
result = await n_client.query_key('stackoverflow')
return JSONResponse(dict(data=result,dt=datetime.now().isoformat()))
@app.post('/redis')
async def post_redis(request: Request , data):
n_client = request.app.state.n_client
n_client : Redis
await n_client.insert_string('stackoverflow',data)
result = await n_client.query_key('stackoverflow')
return JSONResponse(result)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment