Client

pyhellofresh.HelloFreshClient

HelloFreshClient(
    *,
    session: ClientSession | None = None,
    access_token: str | None = None,
    refresh_token: str | None = None,
    guest_token: str | None = DEFAULT_GUEST_TOKEN,
    country: str = DEFAULT_COUNTRY,
    locale: str = DEFAULT_LOCALE,
    base_url: str = DEFAULT_BASE_URL,
    request_timeout: float = 10.0,
)

Async Client for accessing HelloFresh APIs.

Supports dependency injection of an external aiohttp.ClientSession in compliance with Home Assistant's inject-websession quality rule.

Initialize the HelloFresh Client.

Parameters:

Name	Type	Description	Default
session	ClientSession | None	Optional externally managed aiohttp.ClientSession.	None
access_token	str | None	Optional OAuth Bearer access token.	None
refresh_token	str | None	Optional OAuth refresh token.	None
guest_token	str | None	Optional guest token for unauthenticated gateway endpoints.	DEFAULT_GUEST_TOKEN
country	str	ISO country code (default "GB").	DEFAULT_COUNTRY
locale	str	Language locale identifier (default "en-GB").	DEFAULT_LOCALE
base_url	str	Base URL for HelloFresh gateway (default "https://www.hellofresh.co.uk").	DEFAULT_BASE_URL
request_timeout	float	Timeout in seconds for HTTP requests (default 10.0).	10.0

Source code in src/pyhellofresh/client.py

def __init__(
    self,
    *,
    session: aiohttp.ClientSession | None = None,
    access_token: str | None = None,
    refresh_token: str | None = None,
    guest_token: str | None = DEFAULT_GUEST_TOKEN,
    country: str = DEFAULT_COUNTRY,
    locale: str = DEFAULT_LOCALE,
    base_url: str = DEFAULT_BASE_URL,
    request_timeout: float = 10.0,
) -> None:
    """Initialize the HelloFresh Client.

    Args:
        session: Optional externally managed aiohttp.ClientSession.
        access_token: Optional OAuth Bearer access token.
        refresh_token: Optional OAuth refresh token.
        guest_token: Optional guest token for unauthenticated gateway endpoints.
        country: ISO country code (default "GB").
        locale: Language locale identifier (default "en-GB").
        base_url: Base URL for HelloFresh gateway (default "https://www.hellofresh.co.uk").
        request_timeout: Timeout in seconds for HTTP requests (default 10.0).
    """
    self._session = session
    self._owns_session = session is None
    self._access_token = access_token
    self._refresh_token = refresh_token
    self._guest_token = guest_token
    self._country = country
    self._locale = locale
    self._base_url = base_url.rstrip("/")
    self._request_timeout = request_timeout

access_token property writable

access_token: str | None

Get current access token.

refresh_token property writable

refresh_token: str | None

Get current refresh token.

country property

country: str

Get country code.

locale property

locale: str

Get locale identifier.

base_url property

base_url: str

Get base URL.

request_timeout property

request_timeout: float

Get default request timeout in seconds.

with_session async classmethod

with_session(
    *,
    access_token: str | None = None,
    refresh_token: str | None = None,
    guest_token: str | None = DEFAULT_GUEST_TOKEN,
    country: str = DEFAULT_COUNTRY,
    locale: str = DEFAULT_LOCALE,
    base_url: str = DEFAULT_BASE_URL,
    request_timeout: float = 10.0,
) -> Self

Create a client instance with an internally managed aiohttp ClientSession.

Parameters:

Name	Type	Description	Default
access_token	str | None	Optional OAuth Bearer access token.	None
refresh_token	str | None	Optional OAuth refresh token.	None
guest_token	str | None	Optional guest token for unauthenticated gateway endpoints.	DEFAULT_GUEST_TOKEN
country	str	ISO country code.	DEFAULT_COUNTRY
locale	str	Language locale.	DEFAULT_LOCALE
base_url	str	Base gateway URL.	DEFAULT_BASE_URL
request_timeout	float	Request timeout in seconds.	10.0

Returns:

Type	Description
Self	HelloFreshClient initialized with an active session.

Source code in src/pyhellofresh/client.py

@classmethod
async def with_session(
    cls,
    *,
    access_token: str | None = None,
    refresh_token: str | None = None,
    guest_token: str | None = DEFAULT_GUEST_TOKEN,
    country: str = DEFAULT_COUNTRY,
    locale: str = DEFAULT_LOCALE,
    base_url: str = DEFAULT_BASE_URL,
    request_timeout: float = 10.0,
) -> Self:
    """Create a client instance with an internally managed aiohttp ClientSession.

    Args:
        access_token: Optional OAuth Bearer access token.
        refresh_token: Optional OAuth refresh token.
        guest_token: Optional guest token for unauthenticated gateway endpoints.
        country: ISO country code.
        locale: Language locale.
        base_url: Base gateway URL.
        request_timeout: Request timeout in seconds.

    Returns:
        HelloFreshClient initialized with an active session.
    """
    session = aiohttp.ClientSession()
    return cls(
        session=session,
        access_token=access_token,
        refresh_token=refresh_token,
        guest_token=guest_token,
        country=country,
        locale=locale,
        base_url=base_url,
        request_timeout=request_timeout,
    )

__aenter__ async

__aenter__() -> Self

Async context manager entry point.

Source code in src/pyhellofresh/client.py

async def __aenter__(self) -> Self:
    """Async context manager entry point."""
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Async context manager exit point.

Source code in src/pyhellofresh/client.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: types.TracebackType | None,
) -> None:
    """Async context manager exit point."""
    await self.close()

close async

close() -> None

Close the internally owned aiohttp session.

Note: If the session was injected externally, it will NOT be closed.

Source code in src/pyhellofresh/client.py

async def close(self) -> None:
    """Close the internally owned aiohttp session.

    Note: If the session was injected externally, it will NOT be closed.
    """
    if self._owns_session and self._session and not self._session.closed:
        await self._session.close()

fetch_guest_token async

fetch_guest_token() -> str

Dynamically fetch a fresh guest token from HelloFresh SSR HTML.

Returns:

Type	Description
str	The freshly extracted guest JWT access token.

Raises:

Type	Description
HelloFreshConnectionError	On network issue or timeout.
HelloFreshAuthenticationError	If guest token cannot be parsed from SSR HTML.

Source code in src/pyhellofresh/client.py

async def fetch_guest_token(self) -> str:
    """Dynamically fetch a fresh guest token from HelloFresh SSR HTML.

    Returns:
        The freshly extracted guest JWT access token.

    Raises:
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshAuthenticationError: If guest token cannot be parsed from SSR HTML.
    """
    session = await self._get_session()
    url = f"{self._base_url}/login"
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }

    try:
        async with session.get(url, headers=headers) as resp:
            if resp.status >= 400:
                text = await resp.text()
                raise HelloFreshResponseError(resp.status, text)

            html = await resp.text()
            match = re.search(
                r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
                html,
            )
            if match:
                data = json.loads(match.group(1))
                server_auth = (
                    data.get("props", {})
                    .get("pageProps", {})
                    .get("ssrPayload", {})
                    .get("serverAuth", {})
                )
                guest_token = server_auth.get("access_token")
                if guest_token and isinstance(guest_token, str):
                    self._guest_token = guest_token
                    return guest_token

            raise HelloFreshAuthenticationError(
                "Unable to parse guest token from HelloFresh SSR HTML payload."
            )
    except TimeoutError as err:
        raise HelloFreshConnectionError("Request timed out") from err
    except aiohttp.ClientError as err:
        raise HelloFreshConnectionError(f"Connection error: {err}") from err

start_passwordless_login async

start_passwordless_login(
    email: str, redirect_url: str | None = None
) -> str

Start passwordless magic link login flow.

Parameters:

Name	Type	Description	Default
email	str	User email address.	required
redirect_url	str | None	Optional custom redirect URL.	None

Returns:

Type	Description
str	The generated public_id UUID string.

Raises:

Type	Description
HelloFreshConnectionError	On network issue or timeout.
HelloFreshResponseError	On unexpected status code.

Source code in src/pyhellofresh/client.py

async def start_passwordless_login(
    self, email: str, redirect_url: str | None = None
) -> str:
    """Start passwordless magic link login flow.

    Args:
        email: User email address.
        redirect_url: Optional custom redirect URL.

    Returns:
        The generated public_id UUID string.

    Raises:
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshResponseError: On unexpected status code.
    """
    public_id = str(uuid.uuid4())
    path = "/gw/v1/passwordless/start"
    params = {
        "country": self._country,
        "locale": self._locale,
    }
    payload = {
        "email": email,
        "channel": "email",
        "send": "link",
        "redirect_url": redirect_url
        or f"{self._base_url}/my-account/deliveries/menu",
        "public_id": public_id,
    }
    await self._request(
        "POST", path, params=params, json_data=payload, auth_required=False
    )
    return public_id

finish_passwordless_login_from_url async

finish_passwordless_login_from_url(
    url: str, public_id: str | None = None
) -> TokenResponse

Complete passwordless magic link login using a magic link URL or tracking link.

Resolves tracking redirects (e.g. click.link.hellofresh.co.uk), extracts the code, email, public_id, and redirect_url query parameters, and exchanges them for OAuth tokens.

Parameters:

Name	Type	Description	Default
url	str	The full magic link URL received in email.	required
public_id	str | None	Optional fallback public_id if missing from URL query parameters.	None

Returns:

Type	Description
TokenResponse	TokenResponse containing access_token and refresh_token.

Raises:

Type	Description
HelloFreshAuthenticationError	If required parameters cannot be extracted.
HelloFreshConnectionError	On network or redirection failure.

Source code in src/pyhellofresh/client.py

async def finish_passwordless_login_from_url(
    self,
    url: str,
    public_id: str | None = None,
) -> TokenResponse:
    """Complete passwordless magic link login using a magic link URL or tracking link.

    Resolves tracking redirects (e.g. click.link.hellofresh.co.uk), extracts
    the code, email, public_id, and redirect_url query parameters, and exchanges
    them for OAuth tokens.

    Args:
        url: The full magic link URL received in email.
        public_id: Optional fallback public_id if missing from URL query parameters.

    Returns:
        TokenResponse containing access_token and refresh_token.

    Raises:
        HelloFreshAuthenticationError: If required parameters cannot be extracted.
        HelloFreshConnectionError: On network or redirection failure.
    """
    parsed = urlparse(url)
    query = parse_qs(parsed.query)

    code_list = query.get("code")
    email_list = query.get("email")
    pub_id_list = query.get("public_id")
    redirect_list = query.get("redirect_url")

    code = code_list[0] if code_list else None
    email = email_list[0] if email_list else None
    resolved_pub_id = pub_id_list[0] if pub_id_list else public_id
    redirect_url = redirect_list[0] if redirect_list else None

    # If missing parameters and it's a tracking URL, resolve redirect using Location header
    if (not code or not email or not resolved_pub_id) and url.startswith(
        ("http://", "https://")
    ):
        session = await self._get_session()
        headers = {"User-Agent": USER_AGENT}
        try:
            async with session.get(
                url, headers=headers, allow_redirects=False
            ) as resp:
                if (
                    resp.status in (301, 302, 303, 307, 308)
                    and "Location" in resp.headers
                ):
                    final_url_str = resp.headers["Location"]
                else:
                    final_url_str = str(resp.url)

                parsed = urlparse(final_url_str)
                query = parse_qs(parsed.query)
                code = (query.get("code") or [None])[0]
                email = (query.get("email") or [None])[0]
                resolved_pub_id = (query.get("public_id") or [resolved_pub_id])[0]
                redirect_url = (query.get("redirect_url") or [redirect_url])[0]
        except aiohttp.ClientError as err:
            raise HelloFreshConnectionError(
                f"Failed to resolve magic link URL redirects: {err}"
            ) from err

    if not code or not email or not resolved_pub_id:
        raise HelloFreshAuthenticationError(
            "Invalid magic link URL. Missing required query parameters "
            f"(code, email, or public_id) in URL: {url}"
        )

    return await self.finish_passwordless_login(
        code=code,
        email=email,
        public_id=resolved_pub_id,
        redirect_url=redirect_url,
    )

finish_passwordless_login async

finish_passwordless_login(
    code: str,
    email: str,
    public_id: str,
    redirect_url: str | None = None,
) -> TokenResponse

Complete passwordless magic link login using the code from the magic link.

Parameters:

Name	Type	Description	Default
code	str	The magic link token code.	required
email	str	User email address.	required
public_id	str	The public_id returned by start_passwordless_login.	required
redirect_url	str | None	Optional custom redirect URL.	None

Returns:

Type	Description
TokenResponse	TokenResponse object containing access_token and refresh_token.

Raises:

Type	Description
HelloFreshConnectionError	On network issue or timeout.
HelloFreshResponseError	On unexpected status code.

Source code in src/pyhellofresh/client.py

async def finish_passwordless_login(
    self,
    code: str,
    email: str,
    public_id: str,
    redirect_url: str | None = None,
) -> TokenResponse:
    """Complete passwordless magic link login using the code from the magic link.

    Args:
        code: The magic link token code.
        email: User email address.
        public_id: The public_id returned by start_passwordless_login.
        redirect_url: Optional custom redirect URL.

    Returns:
        TokenResponse object containing access_token and refresh_token.

    Raises:
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshResponseError: On unexpected status code.
    """
    path = "/gw/v1/passwordless/magic-link/finish"
    params = {
        "channel": "email",
        "code": code,
        "country": self._country.lower(),
        "email": email,
        "public_id": public_id,
        "redirect_url": redirect_url
        or f"{self._base_url}/my-account/deliveries/menu",
    }
    data = await self._request("GET", path, params=params, auth_required=False)
    token_resp = TokenResponse.from_dict(data)
    if token_resp.access_token:
        self._access_token = token_resp.access_token
    if token_resp.refresh_token:
        self._refresh_token = token_resp.refresh_token
    return token_resp

refresh_access_token async

refresh_access_token(
    refresh_token: str | None = None,
) -> TokenResponse

Refresh the access token using a refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str | None	Optional refresh token. Uses stored refresh_token if omitted.	None

Returns:

Type	Description
TokenResponse	TokenResponse containing updated access_token and refresh_token.

Raises:

Type	Description
HelloFreshAuthenticationError	If no refresh token is available.
HelloFreshConnectionError	On network issue or timeout.
HelloFreshResponseError	On unexpected status code.

Source code in src/pyhellofresh/client.py

async def refresh_access_token(
    self, refresh_token: str | None = None
) -> TokenResponse:
    """Refresh the access token using a refresh token.

    Args:
        refresh_token: Optional refresh token. Uses stored refresh_token if omitted.

    Returns:
        TokenResponse containing updated access_token and refresh_token.

    Raises:
        HelloFreshAuthenticationError: If no refresh token is available.
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshResponseError: On unexpected status code.
    """
    token_to_use = refresh_token or self._refresh_token
    if not token_to_use:
        raise HelloFreshAuthenticationError("No refresh token available.")

    path = "/gw/refresh"
    params = {
        "country": self._country,
        "locale": self._locale,
    }
    payload = {
        "refresh_token": token_to_use,
    }
    data = await self._request(
        "POST", path, params=params, json_data=payload, auth_required=False
    )
    token_resp = TokenResponse.from_dict(data)
    if token_resp.access_token:
        self._access_token = token_resp.access_token
    if token_resp.refresh_token:
        self._refresh_token = token_resp.refresh_token
    return token_resp

get_profile async

get_profile() -> Profile

Fetch the current customer profile.

Returns:

Type	Description
Profile	Profile object containing household and dietary preferences.

Raises:

Type	Description
HelloFreshAuthenticationError	If not authenticated or token invalid.
HelloFreshConnectionError	On network issue or timeout.
HelloFreshResponseError	On unexpected status code.

Source code in src/pyhellofresh/client.py

async def get_profile(self) -> Profile:
    """Fetch the current customer profile.

    Returns:
        Profile object containing household and dietary preferences.

    Raises:
        HelloFreshAuthenticationError: If not authenticated or token invalid.
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshResponseError: On unexpected status code.
    """
    path = "/gw/profile-service/v2/customers/me/profile"
    params = {
        "brand": DEFAULT_BRAND,
        "regionCode": self._country,
    }
    data = await self._request("GET", path, params=params)
    return Profile.from_dict(data)

get_customer_info async

get_customer_info() -> dict[str, Any]

Fetch basic customer info including UUID, active subscription ID, and plan IDs.

Returns:

Type	Description
dict[str, Any]	Dictionary containing customer info.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_customer_info(self) -> dict[str, Any]:
    """Fetch basic customer info including UUID, active subscription ID, and plan IDs.

    Returns:
        Dictionary containing customer info.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    path = "/gw/api/customers/me/info"
    params = {
        "country": self._country,
        "locale": self._locale,
    }
    return await self._request("GET", path, params=params, auth_required=True)

get_subscriptions async

get_subscriptions() -> list[dict[str, Any]]

Fetch customer active subscriptions.

Returns:

Type	Description
list[dict[str, Any]]	List of subscription dictionaries.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_subscriptions(self) -> list[dict[str, Any]]:
    """Fetch customer active subscriptions.

    Returns:
        List of subscription dictionaries.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    path = "/gw/api/customers/me/subscriptions"
    params = {"country": self._country}
    data = await self._request("GET", path, params=params, auth_required=True)
    if isinstance(data, list):
        return data
    if isinstance(data, dict) and "items" in data:
        return data["items"]
    return []

get_balance async

get_balance(
    customer_id: str | None = None,
) -> AccountBalance

Fetch account balance.

Parameters:

Name	Type	Description	Default
customer_id	str | None	Optional customer UUID string. If None, fetches customer info UUID.	None

Returns:

Type	Description
AccountBalance	AccountBalance object.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_balance(self, customer_id: str | None = None) -> AccountBalance:
    """Fetch account balance.

    Args:
        customer_id: Optional customer UUID string. If None, fetches customer info UUID.

    Returns:
        AccountBalance object.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    if not customer_id or customer_id == "me":
        info = await self.get_customer_info()
        customer_id = info.get("uuid") or info.get("id")

    path = f"/gw/payments/customers/{customer_id}/balance"
    params = {
        "business_unit": self._country,
        "country": self._country,
    }
    data = await self._request("GET", path, params=params, auth_required=True)
    return AccountBalance.from_dict(data)

get_past_deliveries async

get_past_deliveries(
    range_start: str | None = None,
    range_end: str | None = None,
) -> PastDeliveries

Fetch customer deliveries schedule.

Parameters:

Name	Type	Description	Default
range_start	str | None	Optional ISO week string start (e.g. '2026-W30').	None
range_end	str | None	Optional ISO week string end (e.g. '2026-W40').	None

Returns:

Type	Description
PastDeliveries	PastDeliveries model containing past delivery items.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_past_deliveries(
    self,
    range_start: str | None = None,
    range_end: str | None = None,
) -> PastDeliveries:
    """Fetch customer deliveries schedule.

    Args:
        range_start: Optional ISO week string start (e.g. '2026-W30').
        range_end: Optional ISO week string end (e.g. '2026-W40').

    Returns:
        PastDeliveries model containing past delivery items.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    path = "/gw/api/customers/me/deliveries"
    params = {
        "country": self._country,
        "locale": self._locale,
    }
    if range_start:
        params["rangeStart"] = range_start
    if range_end:
        params["rangeEnd"] = range_end

    data = await self._request("GET", path, params=params, auth_required=True)
    return PastDeliveries.from_dict(data)

get_menu async

get_menu(
    week: str,
    subscription_id: str | int | None = None,
    product_sku: str | None = None,
) -> WeeklyMenu

Fetch weekly menu recipes.

Parameters:

Name	Type	Description	Default
week	str	Target delivery week identifier (e.g. '2026-W32').	required
subscription_id	str | int | None	Optional customer subscription ID.	None
product_sku	str | None	Optional product SKU (default auto-retrieved from customer info).	None

Returns:

Type	Description
WeeklyMenu	WeeklyMenu model.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_menu(
    self,
    week: str,
    subscription_id: str | int | None = None,
    product_sku: str | None = None,
) -> WeeklyMenu:
    """Fetch weekly menu recipes.

    Args:
        week: Target delivery week identifier (e.g. '2026-W32').
        subscription_id: Optional customer subscription ID.
        product_sku: Optional product SKU (default auto-retrieved from customer info).

    Returns:
        WeeklyMenu model.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    if not subscription_id or not product_sku:
        try:
            info = await self.get_customer_info()
            if not subscription_id:
                subscription_id = info.get("activeSubscriptionId")
            if not product_sku:
                product_sku = info.get("activeSubscriptionSkus")
        except HelloFreshError:
            pass

    path = "/gw/my-deliveries/menu"
    params: dict[str, Any] = {
        "week": week,
        "country": self._country.lower(),
        "locale": self._locale,
    }
    if product_sku:
        params["product-sku"] = str(product_sku)
    if subscription_id:
        params["subscription"] = str(subscription_id)

    data = await self._request("GET", path, params=params, auth_required=True)
    return WeeklyMenu.from_dict(data)

get_meals_for_week_offset async

get_meals_for_week_offset(offset: int = 0) -> list[Meal]

Return the meals selected for a delivery week.

Offset 0 is the latest locked delivery: the newest scheduled week whose cutoff has already passed. That is the box arriving now, rather than the following week that is still open for selection. Positive offsets move forward one ISO week at a time, and negative offsets move backward.

Parameters:

Name	Type	Description	Default
offset	int	Weeks after the latest locked delivery. 1 is the following week's selection.	0

Returns:

Type	Description
list[Meal]	Meals whose menu selection quantity is greater than zero.

Raises:

Type	Description
HelloFreshError	If the delivery schedule has no usable week.
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_meals_for_week_offset(self, offset: int = 0) -> list[Meal]:
    """Return the meals selected for a delivery week.

    Offset ``0`` is the latest locked delivery: the newest scheduled week
    whose cutoff has already passed. That is the box arriving now, rather
    than the following week that is still open for selection. Positive
    offsets move forward one ISO week at a time, and negative offsets move
    backward.

    Args:
        offset: Weeks after the latest locked delivery. ``1`` is the
            following week's selection.

    Returns:
        Meals whose menu selection quantity is greater than zero.

    Raises:
        HelloFreshError: If the delivery schedule has no usable week.
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    now = _utcnow()
    anchor_week = current_iso_week(now)
    deliveries = await self.get_past_deliveries(
        shift_iso_week(anchor_week, -8),
        shift_iso_week(anchor_week, 6),
    )
    latest_week = select_latest_delivery_week(deliveries.weeks, now)
    menu = await self.get_menu(shift_iso_week(latest_week, offset))
    return [meal for meal in menu.meals if meal.selected]

get_recipe async

get_recipe(recipe_id: str) -> Recipe

Fetch details for a specific recipe.

Parameters:

Name	Type	Description	Default
recipe_id	str	Recipe identifier string.	required

Returns:

Type	Description
Recipe	Recipe model object.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_recipe(self, recipe_id: str) -> Recipe:
    """Fetch details for a specific recipe.

    Args:
        recipe_id: Recipe identifier string.

    Returns:
        Recipe model object.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    path = f"/gw/recipes/recipes/{recipe_id}"
    params = {
        "country": self._country,
        "locale": self._locale,
    }
    data = await self._request("GET", path, params=params, auth_required=True)
    return Recipe.from_dict(data)

search_recipes async

search_recipes(
    query: str, take: int = 20, skip: int = 0
) -> list[Recipe]

Search HelloFresh recipe catalog by keyword query.

Parameters:

Name	Type	Description	Default
query	str	Keyword search string (e.g. "pasta", "chicken", "tacos").	required
take	int	Maximum number of recipes to return (default 20).	20
skip	int	Pagination offset index (default 0).	0

Returns:

Type	Description
list[Recipe]	List of Recipe models matching search query.

Raises:

Type	Description
HelloFreshConnectionError	On network issue or timeout.
HelloFreshResponseError	On unexpected API error.

Source code in src/pyhellofresh/client.py

async def search_recipes(
    self,
    query: str,
    take: int = 20,
    skip: int = 0,
) -> list[Recipe]:
    """Search HelloFresh recipe catalog by keyword query.

    Args:
        query: Keyword search string (e.g. "pasta", "chicken", "tacos").
        take: Maximum number of recipes to return (default 20).
        skip: Pagination offset index (default 0).

    Returns:
        List of Recipe models matching search query.

    Raises:
        HelloFreshConnectionError: On network issue or timeout.
        HelloFreshResponseError: On unexpected API error.
    """
    path = "/gw/api/recipes/search"
    params = {
        "country": self._country,
        "locale": self._locale,
        "q": query,
        "take": take,
        "skip": skip,
    }
    data = await self._request("GET", path, params=params, auth_required=False)
    items = data.get("items", []) if isinstance(data, dict) else []
    return [Recipe.from_dict(item) for item in items if isinstance(item, dict)]

get_cart_price async

get_cart_price(
    week: str,
    box_size: int = 2,
    subscription_id: str | int | None = None,
    product_sku: str | None = None,
    products: list[dict[str, Any]] | None = None,
) -> CartPrice

Calculate cart price for a specific week.

Parameters:

Name	Type	Description	Default
week	str	Target delivery week identifier (e.g. '2026-W32').	required
box_size	int	Number of meals / box size (default 2).	2
subscription_id	str | int | None	Optional subscription ID.	None
product_sku	str | None	Optional product SKU string.	None
products	list[dict[str, Any]] | None	Optional list of product dictionaries.	None

Returns:

Type	Description
CartPrice	CartPrice model.

Raises:

Type	Description
HelloFreshAuthenticationError	If access_token is missing or expired.
HelloFreshConnectionError	On network issue or timeout.

Source code in src/pyhellofresh/client.py

async def get_cart_price(
    self,
    week: str,
    box_size: int = 2,
    subscription_id: str | int | None = None,
    product_sku: str | None = None,
    products: list[dict[str, Any]] | None = None,
) -> CartPrice:
    """Calculate cart price for a specific week.

    Args:
        week: Target delivery week identifier (e.g. '2026-W32').
        box_size: Number of meals / box size (default 2).
        subscription_id: Optional subscription ID.
        product_sku: Optional product SKU string.
        products: Optional list of product dictionaries.

    Returns:
        CartPrice model.

    Raises:
        HelloFreshAuthenticationError: If access_token is missing or expired.
        HelloFreshConnectionError: On network issue or timeout.
    """
    customer_id_val: int | None = None
    plan_id_val: str | None = None

    try:
        info = await self.get_customer_info()
        if not subscription_id:
            subscription_id = info.get("activeSubscriptionId")
        if not product_sku:
            product_sku = info.get("activeSubscriptionSkus")
        if info.get("id"):
            try:
                customer_id_val = int(info["id"])
            except (ValueError, TypeError):
                pass
        if info.get("customerPlanIds") and isinstance(
            info["customerPlanIds"], list
        ):
            plan_id_val = info["customerPlanIds"][0]
    except HelloFreshError:
        pass

    if not products:
        sku_handle = product_sku or "GB-CBU-2-2-0"
        products = [
            {
                "handle": sku_handle,
                "hfWeek": week,
            }
        ]

    path = f"/gw/v1/carts/{week}/price"
    payload: dict[str, Any] = {
        "country": self._country,
        "locale": self._locale,
        "boxSize": box_size,
        "isFirstOrder": False,
        "isRecurring": True,
        "products": products,
    }
    if customer_id_val is not None:
        payload["customerID"] = customer_id_val
    if subscription_id is not None:
        if isinstance(subscription_id, int):
            payload["subscriptionID"] = subscription_id
        elif str(subscription_id).isdigit():
            payload["subscriptionID"] = int(subscription_id)
        else:
            payload["subscriptionID"] = subscription_id
    if plan_id_val:
        payload["planID"] = plan_id_val

    data = await self._request("POST", path, json_data=payload, auth_required=True)
    return CartPrice.from_dict(data)