Comment implémenter le basculement et les nouvelles tentatives de requêtes dans FastAPI

Comment implémenter la récupération après échec de demande et les nouvelles tentatives dans FastAPI

Introduction :
Lors du développement d'applications Web, nous avons souvent besoin de communiquer avec d'autres services. Cependant, ces services peuvent rencontrer des pannes, telles que des pannes temporaires de réseau ou des délais de réponse. Pour que nos applications restent fiables, nous devons nous remettre des échecs et réessayer si nécessaire. Dans cet article, nous apprendrons comment implémenter le basculement et les nouvelles tentatives de requêtes dans FastAPI.

FastAPI est un framework Web moderne basé sur Python, qui fournit des fonctions de traitement et de routage des requêtes simples et efficaces. Il utilise une méthode asynchrone en interne pour traiter les demandes, ce qui facilite la mise en œuvre de la reprise après incident et des nouvelles tentatives.

Idées d'implémentation de la récupération sur erreur et des nouvelles tentatives :
Avant d'implémenter les fonctions de récupération sur erreur et de nouvelle tentative, nous devons d'abord comprendre certains concepts de base.

1. Mode Disjoncteur :
   Le disjoncteur est un mode de conception qui peut empêcher la propagation des défauts. Lorsqu'un service tombe en panne, le disjoncteur ferme temporairement les demandes adressées au service pendant une certaine période et renvoie directement les demandes. Cela évite une charge supplémentaire sur le service défaillant et donne au service un certain temps pour récupérer. Lorsqu'aucun défaut ne se produit dans un certain laps de temps, le disjoncteur récupère automatiquement et transmet à nouveau la demande au service.
2. Mécanisme de nouvelle tentative :
   Le mécanisme de nouvelle tentative consiste à renvoyer la même demande lorsque la demande échoue, ce qui ajoute généralement un certain temps de retard. Le mécanisme de nouvelle tentative peut garantir la fiabilité de la demande et renvoyer la demande après une reprise après échec.

Sur la base des idées ci-dessus, nous pouvons réaliser une récupération après échec de la demande et réessayer en utilisant le mode disjoncteur et le mécanisme de nouvelle tentative.

Étapes d'implémentation :
L'exemple de code suivant montre comment implémenter le basculement et la nouvelle tentative de demande dans FastAPI :

from fastapi import FastAPI
import requests

app = FastAPI()

@app.get("/retry")
def retry():
    url = "http://example.com/api"  # 要请求的URL
    max_retries = 3  # 最大重试次数
    retry_interval = 1  # 重试间隔时间（秒）

    retries = 0
    while retries <= max_retries:
        try:
            response = requests.get(url)
            return response.json()
        except requests.exceptions.RequestException as error:
            print(f"Request failed: {error}")
            retries += 1
            time.sleep(retry_interval)

    return {"message": "Max retries exceeded"}

@app.get("/circuit-breaker")
def circuit_breaker():
    url = "http://example.com/api"  # 要请求的URL
    breaker = Breaker(url, max_failures=3, reset_timeout=10)

    try:
        response = breaker.execute(requests.get)
        return response.json()
    except BreakerOpenError:
        return {"message": "Service temporarily unavailable"}

class Breaker:
    def __init__(self, url, max_failures, reset_timeout):
        self.url = url
        self.max_failures = max_failures
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.last_failure = None

    def execute(self, func):
        if self.failures >= self.max_failures:
            if self.last_failure and time.time() - self.last_failure < self.reset_timeout:
                raise BreakerOpenError()
            else:
                self.reset()

        try:
            response = func(self.url)
            self.failures = 0
            return response
        except requests.exceptions.RequestException as error:
            print(f"Request failed: {error}")
            self.failures += 1
            self.last_failure = time.time()

    def reset(self):
        self.failures = 0
        self.last_failure = None

class BreakerOpenError(Exception):
    pass

Dans le code ci-dessus, nous utilisons une classe de disjoncteur personnalisée en utilisant l'exemple de retry和circuit_breaker两个路由示例来实现请求的故障恢复和重试。retry路由示例使用一个while循环来进行重试，当请求失败时，会等待一段时间后重新发送请求。circuit_breakerroutage, dans Après le nombre maximum d'échecs est atteint, une exception BreakerOpenError personnalisée sera levée.

Conclusion :
Dans cet article, nous avons appris comment implémenter le basculement et les nouvelles tentatives de requêtes dans FastAPI. En utilisant le modèle de disjoncteur et le mécanisme de nouvelle tentative, nous pouvons améliorer la fiabilité de l'application et récupérer des pannes. Dans le même temps, nous avons également démontré comment implémenter les fonctions de récupération après échec et de nouvelle tentative dans FastAPI via un exemple de code. J'espère que cet article vous sera utile, merci d'avoir lu !

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!