आज की डिजिटल दुनिया में, हर एक्शन - चाहे वह डेटिंग ऐप पर स्वाइप करे या पर्दे के पीछे कुशलता से काम करने वाले एपीआई पर खरीदारी पूरी कर रही हो। बैक-एंड डेवलपर्स के रूप में, हम जानते हैं कि हर मिलीसेकंड मायने रखता है। लेकिन हम एपीआई को तेजी से जवाब कैसे दे सकते हैं? उत्तर कैशिंग में निहित है।

] इसे अपने रसोई काउंटरटॉप पर कुंजी सामग्री (नमक, काली मिर्च, तेल) रखने के बजाय इसे हर बार पैंट्री से लाने के बजाय हर बार जब आप पकाने के लिए पेंट्री से लाने के बारे में सोचें - यह समय बचाता है और प्रक्रिया को अधिक कुशल बनाता है। इसी तरह, कैशिंग एपीआई प्रतिक्रिया समय को कम कर देता है, जो कि रेडिस की तरह एक तेज, सुलभ स्थान में आमतौर पर अनुरोधित डेटा को संग्रहीत करके होता है।

आवश्यक पुस्तकालयों को स्थापित करने के लिए

]

]

Pydantic डेटाबेस टेबल और संरचनाएं बनाने के लिए है। AIOCACHE कैश पर अतुल्यकालिक संचालन करेगा। Uvicorn सर्वर चलाने के लिए जिम्मेदार है।

pip install fastapi uvicorn aiocache pydantic

एक विंडोज सिस्टम में सीधे रेडिस सेट करना इस मोड़ पर संभव नहीं है। इसलिए, इसे लिनक्स के लिए विंडोज सबसिस्टम में सेटअप और चलाना होगा। WSL स्थापित करने के निर्देश नीचे दिए गए हैं

Redis सर्वर कनेक्टिविटी का परीक्षण करने के लिए, निम्न कमांड का उपयोग किया जाता है

redis-cli

sudo apt update
sudo apt install redis-server
sudo systemctl start redis

इस कमांड के बाद, यह पोर्ट 6379 के एक आभासी टर्मिनल में प्रवेश करेगा। उस टर्मिनल में, Redis कमांड टाइप और परीक्षण किया जा सकता है।

FASTAPI एप्लिकेशन सेट करना

] हम कैश्ड प्रतिक्रियाओं को संग्रहीत करने के लिए Redis का उपयोग करेंगे।

redis-cli

चरण 1: उपयोगकर्ता डेटा के लिए pydantic मॉडल को परिभाषित करें

हम अपने उपयोगकर्ता मॉडल को परिभाषित करने के लिए Pydantic का उपयोग करेंगे, जो API प्रतिक्रिया की संरचना का प्रतिनिधित्व करता है।

pydantic आयात basemodel से कक्षा उपयोगकर्ता (Basemodel): में नहीं नाम: str ईमेल: str आयु: int

चरण 2: एक कैशिंग डेकोरेटर बनाएं

] यह डेकोरेटर वास्तविक फ़ंक्शन को कॉल करने से पहले Redis से प्रतिक्रिया को पुनः प्राप्त करने का प्रयास करेगा।

आयात json फंक्शनल इम्पोर्ट रैप्स से Aiocache आयात कैश से Fastapi आयात httpexception से def cache_response (ttl: int = 60, namespace: str = "main"): "" "" Fastapi समापन बिंदु के लिए कैशिंग डेकोरेटर। TTL: सेकंड में कैश के लिए रहने का समय। नेमस्पेस: रेडिस में कैश कीज़ के लिए नेमस्पेस। "" "" डेफ डेकोरेटर (फंक): @Wraps (func) async def आवरण (*args, ** kwargs): user_id = kwargs.get ('user_id') या args [0] # उपयोगकर्ता आईडी मान लेना पहला तर्क है cache_key = f "{namespace}: उपयोगकर्ता: {user_id}" cache = cache.redis (एंडपॉइंट = "लोकलहोस्ट", पोर्ट = 6379, नेमस्पेस = नेमस्पेस) # कैश से डेटा पुनः प्राप्त करने का प्रयास करें cached_value = इंतजार cache.get (cache_key) यदि cached_value: JSON.Loads (Cached_value) # रिटर्न कैश्ड डेटा रिटर्न करें # यदि कैश हिट नहीं है, तो वास्तविक फ़ंक्शन को कॉल करें प्रतिक्रिया = प्रतीक्षा करें (*args, ** kwargs) कोशिश करना: # एक टीटीएल के साथ Redis में प्रतिक्रिया को स्टोर करें Cache.set (cache_key, json.dumps (प्रतिक्रिया), ttl = ttl) का इंतजार ई के रूप में अपवाद को छोड़कर: HTTPException (Status_code = 500, विस्तार = f "त्रुटि कैशिंग डेटा: {E}") बढ़ाएं वापसी प्रतिक्रिया वापसी आवरण लौटने वाला

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    email: str
    age: int

] प्रतिक्रिया को बाद के अनुरोधों में तेजी से पहुंच के लिए Redis का उपयोग करके कैश किया जाएगा।

Fastapi आयात Fastapi से app = fastapi () # एक डेटाबेस में उपयोगकर्ताओं का प्रतिनिधित्व करने वाले नमूना डेटा users_db = { 1: {"आईडी": 1, "नाम": "एलिस", "ईमेल": "[email protected]", "आयु": 25}, 2: {"आईडी": 2, "नाम": "बॉब", "ईमेल": "[email protected]", "आयु": 30}, 3: {"आईडी": 3, "नाम": "चार्ली", "ईमेल": "[email protected]", "आयु": 22}, } @app.get ("/उपयोगकर्ता/{user_id}") @cache_response (ttl = 120, namespace = "उपयोगकर्ता") async def get_user_details (user_id: int): # Users_db से डेटा प्राप्त करके एक डेटाबेस कॉल का अनुकरण करें उपयोगकर्ता = users_db.get (user_id) यदि उपयोगकर्ता नहीं: HTTPException (Status_code = 404, विस्तार = "उपयोगकर्ता नहीं मिला") बढ़ाएं लौटें उपयोगकर्ता

चरण 4: आवेदन चलाएं

import json
from functools import wraps
from aiocache import Cache
from fastapi import HTTPException

def cache_response(ttl: int = 60, namespace: str = "main"):
    """
    Caching decorator for FastAPI endpoints.

    ttl: Time to live for the cache in seconds.
    namespace: Namespace for cache keys in Redis.
    """
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            user_id = kwargs.get('user_id') or args[0]  # Assuming the user ID is the first argument
            cache_key = f"{namespace}:user:{user_id}"

            cache = Cache.REDIS(endpoint="localhost", port=6379, namespace=namespace)

            # Try to retrieve data from cache
            cached_value = await cache.get(cache_key)
            if cached_value:
                return json.loads(cached_value)  # Return cached data

            # Call the actual function if cache is not hit
            response = await func(*args, **kwargs)

            try:
                # Store the response in Redis with a TTL
                await cache.set(cache_key, json.dumps(response), ttl=ttl)
            except Exception as e:
                raise HTTPException(status_code=500, detail=f"Error caching data: {e}")

            return response
        return wrapper
    return decorator

]

अब, आप उपयोगकर्ता विवरण प्राप्त करके एपीआई का परीक्षण कर सकते हैं:

http://127.0.0.1:8000/users/1

sudo apt update
sudo apt install redis-server
sudo systemctl start redis

पहला अनुरोध उपयोगकर्ताओं से डेटा प्राप्त करेगा।

कैश का परीक्षण

आप Redis में संग्रहीत कुंजियों का निरीक्षण करके कैश को सत्यापित कर सकते हैं। Redis Cli खोलें:

uvicorn main:app --reload

]

इस उदाहरण में कैशिंग कैसे काम करता है

पहला अनुरोध

http://127.0.0.1:8000/users/1

बाद के अनुरोध:

]

TTL (लाइव करने का समय):

]

निष्कर्ष

इस ट्यूटोरियल में, हमने यह प्रदर्शित किया है कि एक साधारण उपयोगकर्ता विवरण उदाहरण का उपयोग करके FASTAPI एप्लिकेशन में Redis Caching को कैसे लागू किया जाए। एपीआई प्रतिक्रियाओं को कैशिंग करके, आप अपने एप्लिकेशन के प्रदर्शन में काफी सुधार कर सकते हैं, विशेष रूप से डेटा के लिए जो अक्सर नहीं बदलता है।

redis-cli
KEYS *