src/backend/base/langflow/services/store/utils.py

from typing import TYPE_CHECKING

import httpx
from loguru import logger

if TYPE_CHECKING:
    from langflow.services.store.schema import ListComponentResponse
    from langflow.services.store.service import StoreService


def process_tags_for_post(component_dict):
    tags = component_dict.pop("tags", None)
    if tags and all(isinstance(tag, str) for tag in tags):
        component_dict["tags"] = [{"tags_id": tag} for tag in tags]
    return component_dict


async def update_components_with_user_data(
    components: list["ListComponentResponse"],
    store_service: "StoreService",
    store_api_key: str,
    *,
    liked: bool,
):
    """Updates the components with the user data (liked_by_user and in_users_collection)."""
    component_ids = [str(component.id) for component in components]
    if liked:
        # If liked is True, this means all we got were liked_by_user components
        # So we can set liked_by_user to True for all components
        liked_by_user_ids = component_ids
    else:
        liked_by_user_ids = await store_service.get_liked_by_user_components(
            component_ids=component_ids,
            api_key=store_api_key,
        )
    # Now we need to set the liked_by_user attribute
    for component in components:
        component.liked_by_user = str(component.id) in liked_by_user_ids

    return components


# Get the latest released version of langflow (https://pypi.org/project/langflow/)
async def get_lf_version_from_pypi():
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get("https://pypi.org/pypi/langflow/json")
        if response.status_code != httpx.codes.OK:
            return None
        return response.json()["info"]["version"]
    except Exception:  # noqa: BLE001
        logger.opt(exception=True).debug("Error getting the latest version of langflow from PyPI")
        return None


def process_component_data(nodes_list):
    names = [node["id"].split("-")[0] for node in nodes_list]
    metadata = {}
    for name in names:
        if name in metadata:
            metadata[name]["count"] += 1
        else:
            metadata[name] = {"count": 1}
    metadata["total"] = len(names)

    return metadata