Spaces:
Running
Running
import httpx | |
from langflow.logging.logger import logger | |
def _compute_non_prerelease_version(prerelease_version: str) -> str: | |
prerelease_keywords = ["a", "b", "rc", "dev", "post"] | |
for keyword in prerelease_keywords: | |
if keyword in prerelease_version: | |
return prerelease_version.split(keyword)[0][:-1] | |
return prerelease_version | |
def _get_version_info(): | |
"""Retrieves the version of the package from a possible list of package names. | |
This accounts for after package names are updated for -nightly builds. | |
Returns: | |
str: The version of the package | |
Raises: | |
ValueError: If the package is not found from the list of package names. | |
""" | |
from importlib import metadata | |
package_options = [ | |
("langflow", "Langflow"), | |
("langflow-base", "Langflow Base"), | |
("langflow-nightly", "Langflow Nightly"), | |
("langflow-base-nightly", "Langflow Base Nightly"), | |
] | |
__version__ = None | |
for pkg_name, display_name in package_options: | |
try: | |
__version__ = metadata.version(pkg_name) | |
prerelease_version = __version__ | |
version = _compute_non_prerelease_version(prerelease_version) | |
except (ImportError, metadata.PackageNotFoundError): | |
pass | |
else: | |
return { | |
"version": prerelease_version, | |
"main_version": version, | |
"package": display_name, | |
} | |
if __version__ is None: | |
msg = f"Package not found from options {package_options}" | |
raise ValueError(msg) | |
return None | |
VERSION_INFO = _get_version_info() | |
def is_pre_release(v: str) -> bool: | |
"""Whether the version is a pre-release version. | |
Returns a boolean indicating whether the version is a pre-release version, | |
as per the definition of a pre-release segment from PEP 440. | |
""" | |
return any(label in v for label in ["a", "b", "rc"]) | |
def is_nightly(v: str) -> bool: | |
"""Whether the version is a dev (nightly) version. | |
Returns a boolean indicating whether the version is a dev (nightly) version, | |
as per the definition of a dev segment from PEP 440. | |
""" | |
return "dev" in v | |
def fetch_latest_version(package_name: str, *, include_prerelease: bool) -> str | None: | |
from packaging import version as pkg_version | |
package_name = package_name.replace(" ", "-").lower() | |
try: | |
response = httpx.get(f"https://pypi.org/pypi/{package_name}/json") | |
versions = response.json()["releases"].keys() | |
valid_versions = [v for v in versions if include_prerelease or not is_pre_release(v)] | |
if not valid_versions: | |
return None # Handle case where no valid versions are found | |
return max(valid_versions, key=pkg_version.parse) | |
except Exception: # noqa: BLE001 | |
logger.exception("Error fetching latest version") | |
return None | |
def get_version_info(): | |
return VERSION_INFO | |