How do I use this?

[dicolasi]

Coming from resilience4j, I have no idea how to use this nice library. Am I missing the docs?

[dicolasi]

dumb question -> https://hyx.readthedocs.io/en/latest/

[roma-glushko]

@dicolasi there is no dump question :D Thanks for showing some interest in Hyx 🙌

Yes, that's our main doc source https://hyx.readthedocs.io/en/latest/

Also, feel free to talk a bit about your usecase. Want to make sure Hyx is aligned with its users needs 👀

[dicolasi]

@roma-glushko I am using it in FastAPI app. I am interested in using retry for now. However, not convinced that this is the right way to use it:

import logging
from fastapi_cache.decorator import cache
from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport
from src.config import Config
from hyx.retry import retry
import httpx


_transport = RequestsHTTPTransport(url=Config.HASURA_URL)
client = Client(transport=_transport, fetch_schema_from_transport=True)


class CityDetailsClient:

    @staticmethod
    @cache(expire=Config.CACHE_EXPIRY)
    @retry(on=httpx.NetworkError, attempts=4)
    def get_details(city: str):
        query = gql("""
            query GetCityDetails($city: String) {
                uk_data_city_details(where: {city: {_eq: $city}}) {
                    city
                    description
                    h2
                    h3
                    imagelink
                    imgalt
                    shutterstockid
                }
            }
        """)
        variables = {"city": city.title()}
        try:
            data_dict = client.execute(query, variable_values=variables)
            return data_dict["uk_data_city_details"][0]
        except Exception as e:
            logging.error(f"Failed to fetch details for city {city}: {str(e)}")
            return None

should I use an event specific to FastAPI instead of httpx?

[dicolasi]

for instance, would it be possible to use a general master circuit breaker rather than annotate all the methods?

[roma-glushko]

@dicolasi,

I am using it in FastAPI app. I am interested in using retry for now. However, not convinced that this is the right way to use it

Based on the snippet, I could see you are trying to use for the client that hits some API. This is perfectly fine to decorate that get_details() method. However, in this specific case, the except Exception as e: block will intercept all exceptions giving no chance for Hyx's @retry come to play.

So I would remove that try-except block and let @retry() catch NetworkErrors. Then on one level upper where the get_details() method is actually called, I would add try-catch around the method call to handle cases when all 4 retries were failed like:

from hyx.retry.exceptions import AttemptsExceeded

try:
    details = client.get_details("London")
except AttemptsExceeded:
    # handle cases when the API is completely down
    ...

would it be possible to use a general master circuit breaker rather than annotate all the methods?

This sounds like a great idea to have this sort of manager in Hyx. The current circuit breaker implementation is rather suited for gatekeeping one operation/workflow, just like resiliency4j if I remember correctly. But having this higher level component would make a use of circuit breakers even easier.

I could imagine one initing circuit breaker manager somewhere in a dependency injection container and then pass it further to relevant places where breaker's functionality needed like:

from typing import Any

import httpx


class CurrencyRate:
    ...


class CurrencyRates:
    def __init__(self, breaker_manager: CircuitBreakerManager) -> None:
        self._breaker_manager = breaker_manager

    async def get_rates(self) -> list[CurrencyRate]:
        with self.breaker_manager("currency.rates", exceptions=(httpx.NetworkError,)):  # manager creates/retrieves and activates breaker
            async with httpx.AsyncClient() as client:
                response = await client.get("https://api.github.com/events")

                return self._map_rates(response.json())

    def _map_rates(self, data: list[dict[str, Any]]) -> list[CurrencyRate]:
        ...

Let me know if this is what you had on mind. I will create a task to bring this into Hyx.

[dicolasi]

Sorry for the late reply. Yes that is what I had in mind :)