diff --git a/src/lfx/src/lfx/__main__.py b/src/lfx/src/lfx/__main__.py index 9b3e7930d5..e9e8a0cd16 100644 --- a/src/lfx/src/lfx/__main__.py +++ b/src/lfx/src/lfx/__main__.py @@ -147,6 +147,67 @@ def run_command_wrapper( ) +@app.command(name="requirements", help="Generate requirements.txt for a flow", no_args_is_help=True) +def requirements_command_wrapper( + flow_path: str = typer.Argument(help="Path to the Langflow flow JSON file"), + output: str | None = typer.Option( + None, + "--output", + "-o", + help="Output file path (default: stdout)", + ), + lfx_package: str = typer.Option( + "lfx", + "--lfx-package", + help="Name of the LFX package (default: lfx)", + ), + *, + no_lfx: bool = typer.Option( + False, # noqa: FBT003 + "--no-lfx", + help="Exclude the LFX package from output", + ), + no_pin: bool = typer.Option( + False, # noqa: FBT003 + "--no-pin", + help="Do not pin package versions (default: pin to currently installed versions)", + ), +) -> None: + """Generate requirements.txt from a Langflow flow JSON (lazy-loaded).""" + import json + from pathlib import Path + + from lfx.utils.flow_requirements import generate_requirements_txt + + path = Path(flow_path) + if not path.is_file(): + typer.echo(f"Error: File not found: {path}", err=True) + raise typer.Exit(1) + + try: + flow = json.loads(path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError) as e: + typer.echo(f"Error: Could not read flow JSON: {e}", err=True) + raise typer.Exit(1) from e + + content = generate_requirements_txt( + flow, + lfx_package=lfx_package, + include_lfx=not no_lfx, + pin_versions=not no_pin, + ) + + if output: + try: + Path(output).write_text(content, encoding="utf-8") + except OSError as e: + typer.echo(f"Error: Could not write to {output}: {e}", err=True) + raise typer.Exit(1) from e + typer.echo(f"Requirements written to {output}") + else: + typer.echo(content, nl=False) + + def main(): """Main entry point for the LFX CLI.""" app() diff --git a/src/lfx/src/lfx/base/models/unified_models.py b/src/lfx/src/lfx/base/models/unified_models.py index 8067b57d84..6178593d35 100644 --- a/src/lfx/src/lfx/base/models/unified_models.py +++ b/src/lfx/src/lfx/base/models/unified_models.py @@ -26,21 +26,37 @@ from lfx.log.logger import logger from lfx.services.deps import get_variable_service, session_scope from lfx.utils.async_helpers import run_until_complete -# Mapping from class name to (module_path, attribute_name). +# Mapping from class name to (module_path, attribute_name, install_hint | None). # Only the provider package that is actually needed gets imported at runtime. -_MODEL_CLASS_IMPORTS: dict[str, tuple[str, str]] = { - "ChatOpenAI": ("langchain_openai", "ChatOpenAI"), - "ChatAnthropic": ("langchain_anthropic", "ChatAnthropic"), - "ChatGoogleGenerativeAIFixed": ("lfx.base.models.google_generative_ai_model", "ChatGoogleGenerativeAIFixed"), - "ChatOllama": ("langchain_ollama", "ChatOllama"), - "ChatWatsonx": ("langchain_ibm", "ChatWatsonx"), +# install_hint overrides the auto-derived pip name for internal module paths. +_MODEL_CLASS_IMPORTS: dict[str, tuple[str, str, str | None]] = { + "ChatOpenAI": ("langchain_openai", "ChatOpenAI", None), + "ChatAnthropic": ("langchain_anthropic", "ChatAnthropic", None), + "ChatGoogleGenerativeAIFixed": ( + "lfx.base.models.google_generative_ai_model", + "ChatGoogleGenerativeAIFixed", + "langchain-google-genai", + ), + "ChatOllama": ("langchain_ollama", "ChatOllama", None), + "ChatWatsonx": ("langchain_ibm", "ChatWatsonx", None), } -_EMBEDDING_CLASS_IMPORTS: dict[str, tuple[str, str]] = { - "OpenAIEmbeddings": ("langchain_openai", "OpenAIEmbeddings"), - "GoogleGenerativeAIEmbeddings": ("langchain_google_genai", "GoogleGenerativeAIEmbeddings"), - "OllamaEmbeddings": ("langchain_ollama", "OllamaEmbeddings"), - "WatsonxEmbeddings": ("langchain_ibm", "WatsonxEmbeddings"), +_EMBEDDING_CLASS_IMPORTS: dict[str, tuple[str, str, str | None]] = { + "OpenAIEmbeddings": ("langchain_openai", "OpenAIEmbeddings", None), + "GoogleGenerativeAIEmbeddings": ("langchain_google_genai", "GoogleGenerativeAIEmbeddings", None), + "OllamaEmbeddings": ("langchain_ollama", "OllamaEmbeddings", None), + "WatsonxEmbeddings": ("langchain_ibm", "WatsonxEmbeddings", None), +} + +# Canonical mapping of provider name → embedding class name. +# Used by EmbeddingModelComponent and by flow_requirements to resolve +# which PyPI package a given embedding provider needs at runtime. +EMBEDDING_PROVIDER_CLASS_MAPPING: dict[str, str] = { + "OpenAI": "OpenAIEmbeddings", + "Google Generative AI": "GoogleGenerativeAIEmbeddings", + "Ollama": "OllamaEmbeddings", + "IBM WatsonX": "WatsonxEmbeddings", + "IBM watsonx.ai": "WatsonxEmbeddings", # Alias used by MODEL_PROVIDERS_DICT } _model_class_cache: dict[str, type] = {} @@ -60,16 +76,24 @@ def get_model_class(class_name: str) -> type: msg = f"Unknown model class: {class_name}" raise ValueError(msg) - module_path, attr_name = import_info + module_path, attr_name, install_hint = import_info + pkg_hint = install_hint or module_path.split(".")[0].replace("_", "-") try: module = importlib.import_module(module_path) except ImportError as exc: msg = ( f"Could not import '{module_path}' for model class '{class_name}'. " - f"Install the missing package (e.g. uv pip install {module_path.replace('.', '-')})." + f"Install the missing package (e.g. uv pip install {pkg_hint})." ) raise ImportError(msg) from exc - cls = getattr(module, attr_name) + try: + cls = getattr(module, attr_name) + except AttributeError as exc: + msg = ( + f"Module '{module_path}' was imported but does not have attribute '{attr_name}'. " + f"This may indicate a version mismatch. " + ) + raise AttributeError(msg) from exc _model_class_cache[class_name] = cls return cls @@ -87,16 +111,24 @@ def get_embedding_class(class_name: str) -> type: msg = f"Unknown embedding class: {class_name}" raise ValueError(msg) - module_path, attr_name = import_info + module_path, attr_name, install_hint = import_info + pkg_hint = install_hint or module_path.split(".")[0].replace("_", "-") try: module = importlib.import_module(module_path) except ImportError as exc: msg = ( f"Could not import '{module_path}' for embedding class '{class_name}'. " - f"Install the missing package (e.g. uv pip install {module_path.replace('.', '-')})." + f"Install the missing package (e.g. uv pip install {pkg_hint})." ) raise ImportError(msg) from exc - cls = getattr(module, attr_name) + try: + cls = getattr(module, attr_name) + except AttributeError as exc: + msg = ( + f"Module '{module_path}' was imported but does not have attribute '{attr_name}'. " + f"This may indicate a version mismatch. " + ) + raise AttributeError(msg) from exc _embedding_class_cache[class_name] = cls return cls @@ -1085,12 +1117,6 @@ def get_embedding_model_options(user_id: UUID | str | None = None) -> list[dict[ replace_with_live_models(all_models, user_id, enabled_providers, "embeddings", model_provider_metadata) options = [] - embedding_class_mapping = { - "OpenAI": "OpenAIEmbeddings", - "Google Generative AI": "GoogleGenerativeAIEmbeddings", - "Ollama": "OllamaEmbeddings", - "IBM WatsonX": "WatsonxEmbeddings", - } # Provider-specific param mappings param_mappings = { @@ -1168,7 +1194,7 @@ def get_embedding_model_options(user_id: UUID | str | None = None) -> list[dict[ "category": provider, "provider": provider, "metadata": { - "embedding_class": embedding_class_mapping.get(provider, "OpenAIEmbeddings"), + "embedding_class": EMBEDDING_PROVIDER_CLASS_MAPPING.get(provider, "OpenAIEmbeddings"), "param_mapping": param_mappings.get(provider, param_mappings["OpenAI"]), "model_type": "embeddings", # Mark as embedding model }, @@ -1179,7 +1205,7 @@ def get_embedding_model_options(user_id: UUID | str | None = None) -> list[dict[ # Add disabled providers (providers that exist in metadata but have no enabled models) if user_id: for provider, metadata in model_provider_metadata.items(): - if provider not in providers_with_models and provider in embedding_class_mapping: + if provider not in providers_with_models and provider in EMBEDDING_PROVIDER_CLASS_MAPPING: # This provider has no enabled models and supports embeddings, add it as a disabled provider entry options.append( { diff --git a/src/lfx/src/lfx/utils/__init__.py b/src/lfx/src/lfx/utils/__init__.py index 193ad4db08..69d55a6844 100644 --- a/src/lfx/src/lfx/utils/__init__.py +++ b/src/lfx/src/lfx/utils/__init__.py @@ -1 +1,27 @@ """Utilities for lfx package.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from lfx.utils.flow_requirements import ( + generate_requirements_from_file, + generate_requirements_from_flow, + generate_requirements_txt, + ) + +__all__ = [ + "generate_requirements_from_file", + "generate_requirements_from_flow", + "generate_requirements_txt", +] + + +def __getattr__(name: str): + if name in __all__: + from lfx.utils import flow_requirements + + return getattr(flow_requirements, name) + msg = f"module {__name__!r} has no attribute {name!r}" + raise AttributeError(msg) diff --git a/src/lfx/src/lfx/utils/flow_requirements.py b/src/lfx/src/lfx/utils/flow_requirements.py new file mode 100644 index 0000000000..23ddd4289b --- /dev/null +++ b/src/lfx/src/lfx/utils/flow_requirements.py @@ -0,0 +1,599 @@ +"""Generate requirements.txt from a Langflow flow JSON. + +Analyzes a flow's component code and configuration to determine the minimal +set of PyPI packages needed to run that flow on a standalone LFX runner. + +Uses ``importlib.metadata`` to dynamically resolve import names to PyPI +distribution names and to compute the transitive dependency tree of ``lfx``, +eliminating the need for static mapping tables. + +Known limitations +----------------- +* **String-based dynamic imports** — ``importlib.import_module(variable)``, + ``exec()``, and ``__import__()`` are invisible to AST analysis. If a custom + component loads a package this way, it will not appear in the output. +* **PythonREPLTool ``global_imports`` field** — The ``PythonREPLTool`` + component accepts a comma-separated string of module names in a template + field. These are imported at runtime via ``importlib.import_module()`` and + are not detected. +* **Cross-platform versions** — Versions are pinned from the *current* + environment. Packages pinned on macOS may lack Linux wheels (or vice versa) + and a pin from Python 3.12 may not install on 3.10. +* **System-level dependencies** — Native libraries required by Python packages + (e.g. ``libpq-dev`` for ``psycopg2``) cannot be expressed in + ``requirements.txt``. +* **``--lfx-package`` and transitive filtering** — The ``lfx_package`` parameter + controls only the output name (e.g. ``lfx-nightly``). The "already provided + by lfx" filter always resolves against the ``lfx`` distribution installed in + the current environment. If the alternative distribution has different + transitive dependencies, the output may include extra or missing packages. +""" + +from __future__ import annotations + +import ast +import importlib.metadata as md +import inspect +import json +import re +import sys +import warnings +from functools import lru_cache +from pathlib import Path +from types import MappingProxyType + +# --------------------------------------------------------------------------- +# Standard-library module names (3.10+) +# --------------------------------------------------------------------------- +try: + STDLIB_MODULES: frozenset[str] = frozenset(sys.stdlib_module_names) +except AttributeError: + STDLIB_MODULES = frozenset(sys.builtin_module_names) + +# --------------------------------------------------------------------------- +# Import name → PyPI name overrides for packages where the import name +# is completely different from the PyPI name and can't be guessed by the +# underscore-to-hyphen fallback. This is only needed when the package +# is not installed (so packages_distributions() can't resolve it). +# --------------------------------------------------------------------------- +IMPORT_NAME_OVERRIDES: dict[str, str] = { + "bs4": "beautifulsoup4", + "cv2": "opencv-python", + "googleapiclient": "google-api-python-client", + "mem0": "mem0ai", + "sklearn": "scikit-learn", + "attr": "attrs", + "gi": "PyGObject", + "serial": "pyserial", +} + +# --------------------------------------------------------------------------- +# Additional runtime deps that certain imports pull in but are not visible +# in the component's own import statements. +# --------------------------------------------------------------------------- +MODULE_EXTRA_DEPS: dict[str, list[str]] = { + "bs4": ["lxml", "tabulate"], +} + +# Import names that are internal to the lfx/langflow runtime and should +# never appear as separate requirements. +_INTERNAL_IMPORT_NAMES: frozenset[str] = frozenset({"lfx", "langflow", "langflow_base"}) + +# Fields in a component template that may contain provider selection info +# NOTE: Look back into how the dynamic components (LanguageModel, EmbeddingModel) are handled. +# Currently, these two make dependency extraction more complex by requiring +# this "guesswork" on what models are being used. +_MODEL_FIELDS = {"model", "agent_llm", "embeddings_model", "embedding_model"} + +# Fallback provider → package mapping for providers whose component class may +# not be importable in every environment (e.g. Azure OpenAI shares +# langchain-openai with the regular OpenAI provider). +_PROVIDER_PACKAGE_FALLBACKS: dict[str, set[str]] = { + "Azure OpenAI": {"langchain-openai"}, +} + + +# =================================================================== +# Dynamic resolution via importlib.metadata +# =================================================================== + + +@lru_cache(maxsize=1) +def _get_import_to_dist_map() -> MappingProxyType[str, list[str]]: + """Return the mapping of importable names → distribution names. + + Uses ``importlib.metadata.packages_distributions()`` which reverse-maps + every importable top-level name to the distribution(s) that provide it. + For example: ``{'PIL': ['pillow'], 'yaml': ['PyYAML'], ...}``. + + Returns a read-only ``MappingProxyType`` so that callers cannot + accidentally mutate the cached result. + """ + try: + return MappingProxyType(md.packages_distributions()) + except AttributeError: + warnings.warn( + "importlib.metadata.packages_distributions() not available. " + "Package resolution will use heuristic fallbacks.", + stacklevel=2, + ) + return MappingProxyType({}) + except (OSError, ValueError) as exc: + warnings.warn( + f"Failed to read package metadata: {exc}. Package resolution will use heuristic fallbacks.", + stacklevel=2, + ) + return MappingProxyType({}) + + +def _normalize_dist(name: str) -> str: + """Normalize a distribution name for comparison (PEP 503).""" + return re.sub(r"[-_.]+", "-", name).lower() + + +def _pin_version(package_name: str) -> str: + """Return ``package_name==X.Y.Z`` if the package is installed, else bare name.""" + try: + version = md.version(package_name) + except md.PackageNotFoundError: + warnings.warn( + f"Could not determine installed version for '{package_name}'. It will be included without a version pin.", + stacklevel=2, + ) + return package_name + return f"{package_name}=={version}" + + +@lru_cache(maxsize=1) +def _get_lfx_transitive_dists() -> frozenset[str]: + """Compute the full transitive closure of distributions provided by lfx. + + Recursively walks ``importlib.metadata.requires()`` to build the set of + all distribution names (normalized) that are already satisfied by + installing the ``lfx`` package. + """ + + def _collect(dist_name: str, seen: set[str]) -> None: + norm = _normalize_dist(dist_name) + if norm in seen: + return + seen.add(norm) + try: + reqs = md.requires(dist_name) or [] + except md.PackageNotFoundError: + return + for req in reqs: + if "extra ==" in req: + continue # skip optional/extra dependencies + child = re.split(r"[<>=~!\[; ]", req)[0].strip() + if child: + _collect(child, seen) + + seen: set[str] = set() + _collect("lfx", seen) + return frozenset(seen) + + +@lru_cache(maxsize=1) +def _get_lfx_provided_imports() -> frozenset[str]: + """Build the set of import names transitively provided by lfx. + + Combines ``packages_distributions()`` with the transitive dependency tree + to determine which import names are already available after + ``pip install lfx``. + """ + lfx_dists = _get_lfx_transitive_dists() + import_map = _get_import_to_dist_map() + + provided: set[str] = set() + for import_name, dist_names in import_map.items(): + for dist in dist_names: + if _normalize_dist(dist) in lfx_dists: + provided.add(import_name) + break + return frozenset(provided) + + +def _import_to_package(import_name: str) -> str: + """Map a Python import name to its PyPI distribution name. + + Resolution order: + 1. ``importlib.metadata.packages_distributions()`` (authoritative, live) + 2. ``IMPORT_NAME_OVERRIDES`` (non-guessable names for packages that may + not be installed in the current environment) + 3. Underscore-to-hyphen convention (covers most remaining cases) + """ + import_map = _get_import_to_dist_map() + dist_names = import_map.get(import_name) + if dist_names: + return dist_names[0] # first (primary) distribution + # Check the override table for non-guessable names + if import_name in IMPORT_NAME_OVERRIDES: + return IMPORT_NAME_OVERRIDES[import_name] + # Fallback: replace underscores with hyphens (covers most packages) + return import_name.replace("_", "-") + + +# =================================================================== +# AST-based import extraction +# =================================================================== + + +def _extract_imports(source: str) -> set[str]: + """Extract top-level package names from all imports in Python source via AST. + + Walks the entire AST (including function bodies and try/except blocks) so + that lazy imports inside ``build_model()`` etc. are captured. Returns only + the first segment of each dotted import (e.g. ``foo`` from ``import foo.bar``). + """ + try: + tree = ast.parse(source) + except SyntaxError as exc: + warnings.warn( + f"Could not parse component source (SyntaxError: {exc}). " + "Imports from this component will not be included in requirements.", + stacklevel=2, + ) + return set() + + imports: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + imports.add(alias.name.split(".")[0]) + elif isinstance(node, ast.ImportFrom): + if node.level > 0: + # Relative import - skip (internal to the component) + continue + if node.module: + imports.add(node.module.split(".")[0]) + return imports + + +# =================================================================== +# Template / provider detection +# =================================================================== + + +def _resolve_provider_packages(provider_name: str) -> set[str]: + """Dynamically resolve PyPI packages needed for a model provider. + + Uses ``MODEL_PROVIDERS_DICT`` to look up the provider's component instance, + then inspects its class's source module to extract import statements. This + avoids maintaining a static provider→package mapping table. + + This is specifically necessary because the ``LanguageModelComponent`` delegates + to provider-specific components (e.g. ``OpenAIModelComponent``) that dynamically + import the actual model class at runtime. + + Note: only the component's own module is inspected, not parent classes. + Parent classes (e.g. ``LCModelComponent``) are all part of lfx, so any + imports they introduce are already in lfx's transitive dependency tree + and would be filtered out regardless. + """ + try: + from lfx.base.models.model_input_constants import MODEL_PROVIDERS_DICT + except ImportError: + warnings.warn( + f"Could not import MODEL_PROVIDERS_DICT. Provider '{provider_name}' packages will not be resolved.", + stacklevel=2, + ) + return set() + + provider_info = MODEL_PROVIDERS_DICT.get(provider_name) + if not provider_info: + fallback = _PROVIDER_PACKAGE_FALLBACKS.get(provider_name) + if fallback: + return set(fallback) + warnings.warn( + f"Provider '{provider_name}' was detected in the flow but is not " + "registered in MODEL_PROVIDERS_DICT (its package may not be installed). " + "Its dependencies will not be included in requirements.", + stacklevel=2, + ) + return set() + + component_instance = provider_info.get("component_class") + if component_instance is None: + warnings.warn( + f"Provider '{provider_name}' has no component instance in MODEL_PROVIDERS_DICT. " + "Its dependencies will not be included in requirements.", + stacklevel=2, + ) + return set() + + try: + module = inspect.getmodule(type(component_instance)) + if module is None: + warnings.warn( + f"Could not locate source module for provider '{provider_name}'. " + "Its dependencies will not be included in requirements.", + stacklevel=2, + ) + return set() + source = inspect.getsource(module) + except (OSError, TypeError) as exc: + warnings.warn( + f"Could not inspect source for provider '{provider_name}': {exc}. " + "Its dependencies will not be included in requirements.", + stacklevel=2, + ) + return set() + + imports = _extract_imports(source) + lfx_provided = _get_lfx_provided_imports() + packages: set[str] = set() + for imp in imports: + if imp in STDLIB_MODULES or imp in _INTERNAL_IMPORT_NAMES: + continue + if imp in MODULE_EXTRA_DEPS: + for extra in MODULE_EXTRA_DEPS[imp]: + packages.add(extra) + if imp in lfx_provided: + continue + packages.add(_import_to_package(imp)) + return packages + + +def _resolve_embedding_provider_packages(provider_name: str) -> set[str]: + """Resolve PyPI packages needed for an embedding model provider. + + The ``EmbeddingModelComponent`` follows the same dynamic-import pattern as + the ``LanguageModelComponent``: its code field only imports from ``lfx`` + internals, while the actual provider package (e.g. ``langchain-openai``) is + imported at runtime via ``get_embedding_class()``. + + This function bridges that gap by chaining two registries from + ``unified_models.py``: + + 1. ``EMBEDDING_PROVIDER_CLASS_MAPPING``: provider name → embedding class name + 2. ``_EMBEDDING_CLASS_IMPORTS``: class name → (module_path, attr, install_hint) + + Because both registries live in ``unified_models.py``, adding a new + embedding provider there automatically makes it visible here — no + separate mapping to maintain. + """ + try: + from lfx.base.models.unified_models import ( + _EMBEDDING_CLASS_IMPORTS, + EMBEDDING_PROVIDER_CLASS_MAPPING, + ) + except ImportError: + warnings.warn( + "Could not import embedding registries from unified_models. " + f"Embedding packages for provider '{provider_name}' will not be resolved.", + stacklevel=2, + ) + return set() + + class_name = EMBEDDING_PROVIDER_CLASS_MAPPING.get(provider_name) + if not class_name: + # This provider has no embedding support (e.g. Anthropic, Groq). + # This is expected — not a warning — since this function is called + # for every detected provider, including language-model-only ones. + return set() + + import_info = _EMBEDDING_CLASS_IMPORTS.get(class_name) + if not import_info: + warnings.warn( + f"Embedding class '{class_name}' for provider '{provider_name}' is in " + "EMBEDDING_PROVIDER_CLASS_MAPPING but not in _EMBEDDING_CLASS_IMPORTS. " + "The import registry in unified_models.py may need updating.", + stacklevel=2, + ) + return set() + + module_path, _attr_name, install_hint = import_info + + # Use install_hint if provided (handles internal module paths like lfx.base.models.*) + if install_hint: + return {install_hint} + + top_level = module_path.split(".")[0] + if top_level in STDLIB_MODULES or top_level in _INTERNAL_IMPORT_NAMES: + return set() + + lfx_provided = _get_lfx_provided_imports() + if top_level in lfx_provided: + return set() + + return {_import_to_package(top_level)} + + +def _detect_providers_from_template(template: dict) -> set[str]: + """Detect model providers from a component's template field values. + + Looks at model-selection fields (e.g., ``model``, ``agent_llm``) and + extracts the ``provider`` string when the field is configured. + """ + providers: set[str] = set() + for field_name in _MODEL_FIELDS: + field = template.get(field_name) + if not isinstance(field, dict): + continue + value = field.get("value") + if isinstance(value, list): + for item in value: + if isinstance(item, dict) and "provider" in item: + providers.add(item["provider"]) + elif isinstance(value, dict) and "provider" in value: + providers.add(value["provider"]) + return providers + + +# =================================================================== +# Per-node analysis +# =================================================================== + + +def _extract_component_requirements(node: dict) -> tuple[set[str], set[str]]: + """Extract requirements from a single flow node. + + Returns: + A tuple of (package_names, provider_names) where package_names are + PyPI packages required by the component code and provider_names are + model provider strings detected from the template configuration. + """ + packages: set[str] = set() + + node_data = node.get("data", {}) + node_info = node_data.get("node", {}) + template = node_info.get("template", {}) + + lfx_provided = _get_lfx_provided_imports() + + # --- 1. Static analysis: parse the component code --- + code_field = template.get("code") + if isinstance(code_field, dict): + source = code_field.get("value") + if source and isinstance(source, str): + imports = _extract_imports(source) + for imp in imports: + # Skip stdlib + if imp in STDLIB_MODULES: + continue + # Skip lfx / langflow internal imports - lfx provides these + # interfaces at runtime so they should never be listed as + # separate requirements. + if imp in _INTERNAL_IMPORT_NAMES: + continue + + # Always check extra runtime deps (e.g. bs4 → lxml, tabulate) + # even if the import itself is provided by lfx, because the + # extras may not be. + if imp in MODULE_EXTRA_DEPS: + for extra in MODULE_EXTRA_DEPS[imp]: + packages.add(extra) + + # Skip imports already provided by lfx + if imp in lfx_provided: + continue + + pkg = _import_to_package(imp) + packages.add(pkg) + + # --- 2. Dynamic analysis: detect provider from template fields --- + providers = _detect_providers_from_template(template) + + return packages, providers + + +# =================================================================== +# Public API +# =================================================================== + + +def generate_requirements_from_flow( + flow: dict, + *, + lfx_package: str = "lfx", + include_lfx: bool = True, + pin_versions: bool = True, +) -> list[str]: + """Generate a requirements list from a Langflow flow JSON. + + Args: + flow: Parsed Langflow flow JSON (dict). + lfx_package: Name of the LFX package to include (e.g. ``"lfx"`` or + ``"lfx-nightly"``). + include_lfx: Whether to include the LFX package itself. + pin_versions: If True, pin each package to the version currently + installed in this environment (``pkg==X.Y.Z``). Falls back to + an unpinned name when the package is not installed. + + Returns: + Sorted list of PyPI package specifiers needed to run this flow. + """ + all_packages: set[str] = set() + all_providers: set[str] = set() + + data = flow.get("data", {}) + nodes = data.get("nodes", []) + + for node in nodes: + # Skip note nodes (annotations, not executable components) + if node.get("type") == "noteNode": + continue + + packages, providers = _extract_component_requirements(node) + all_packages.update(packages) + all_providers.update(providers) + + # Add provider-specific packages (resolved dynamically from component source) + for provider in all_providers: + all_packages.update(_resolve_provider_packages(provider)) + all_packages.update(_resolve_embedding_provider_packages(provider)) + + fmt = _pin_version if pin_versions else lambda p: p + + # Build final sorted list + result: list[str] = [] + if include_lfx: + result.append(fmt(lfx_package)) + result.extend(sorted(fmt(p) for p in all_packages)) + + return result + + +def generate_requirements_txt( + flow: dict, + *, + lfx_package: str = "lfx", + include_lfx: bool = True, + pin_versions: bool = True, +) -> str: + """Generate requirements.txt content from a Langflow flow JSON. + + Args: + flow: Parsed Langflow flow JSON (dict). + lfx_package: Name of the LFX package to include. + include_lfx: Whether to include the LFX package itself. + pin_versions: If True, pin each package to the currently installed + version. + + Returns: + String content suitable for writing to a requirements.txt file. + """ + reqs = generate_requirements_from_flow( + flow, + lfx_package=lfx_package, + include_lfx=include_lfx, + pin_versions=pin_versions, + ) + lines = [ + "# Auto-generated requirements for Langflow flow", + "# This file contains only the dependencies needed for this specific flow", + "", + ] + lines.extend(reqs) + lines.append("") # trailing newline + return "\n".join(lines) + + +def generate_requirements_from_file( + flow_path: str | Path, + *, + lfx_package: str = "lfx", + include_lfx: bool = True, + pin_versions: bool = True, +) -> list[str]: + """Generate requirements list from a flow JSON file path. + + Args: + flow_path: Path to a Langflow flow JSON file. + lfx_package: Name of the LFX package to include. + include_lfx: Whether to include the LFX package itself. + pin_versions: If True, pin each package to the currently installed + version. + + Returns: + Sorted list of PyPI package specifiers. + """ + path = Path(flow_path) + flow = json.loads(path.read_text(encoding="utf-8")) + return generate_requirements_from_flow( + flow, + lfx_package=lfx_package, + include_lfx=include_lfx, + pin_versions=pin_versions, + ) diff --git a/src/lfx/tests/unit/test_flow_requirements.py b/src/lfx/tests/unit/test_flow_requirements.py new file mode 100644 index 0000000000..73df568ee2 --- /dev/null +++ b/src/lfx/tests/unit/test_flow_requirements.py @@ -0,0 +1,998 @@ +"""Tests for flow_requirements module.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from lfx.utils.flow_requirements import ( + MODULE_EXTRA_DEPS, + _detect_providers_from_template, + _extract_component_requirements, + _extract_imports, + _get_import_to_dist_map, + _get_lfx_provided_imports, + _get_lfx_transitive_dists, + _import_to_package, + _pin_version, + _resolve_embedding_provider_packages, + _resolve_provider_packages, + generate_requirements_from_file, + generate_requirements_from_flow, + generate_requirements_txt, +) + + +def _find_starter_projects_dir() -> Path: + """Walk up from this test file to find the monorepo root and locate starter projects.""" + current = Path(__file__).resolve() + for parent in current.parents: + candidate = parent / "src" / "backend" / "base" / "langflow" / "initial_setup" / "starter_projects" + if candidate.is_dir(): + return candidate + return Path("STARTER_PROJECTS_NOT_FOUND") + + +STARTER_PROJECTS_DIR = _find_starter_projects_dir() + + +# --------------------------------------------------------------------------- +# Helpers to build minimal flow JSON structures +# --------------------------------------------------------------------------- + + +def _make_node( + component_type: str, + code: str = "", + template_extra: dict | None = None, + node_type: str = "genericNode", +) -> dict: + """Build a minimal flow node dict for testing.""" + template: dict = {"_type": "Component"} + if code: + template["code"] = { + "type": "code", + "value": code, + } + if template_extra: + template.update(template_extra) + return { + "id": f"{component_type}-test1", + "type": node_type, + "data": { + "display_name": component_type, + "id": f"{component_type}-test1", + "type": component_type, + "node": { + "display_name": component_type, + "template": template, + }, + }, + } + + +def _make_flow(*nodes: dict) -> dict: + """Build a minimal flow dict from nodes.""" + return { + "data": { + "nodes": list(nodes), + "edges": [], + }, + "name": "Test Flow", + } + + +# =================================================================== +# Unit tests: _extract_imports +# =================================================================== + + +class TestExtractImports: + def test_simple_import(self): + result = _extract_imports("import os") + assert "os" in result + + def test_from_import(self): + result = _extract_imports("from pathlib import Path") + assert "pathlib" in result + + def test_dotted_import(self): + result = _extract_imports("from langchain_openai.chat_models import ChatOpenAI") + assert "langchain_openai" in result + + def test_relative_import_skipped(self): + result = _extract_imports("from .utils import helper") + assert len(result) == 0 + + def test_multiple_imports(self): + code = """ +import os +import json +from typing import Any +from langchain_openai import ChatOpenAI +from bs4 import BeautifulSoup +""" + result = _extract_imports(code) + assert "os" in result + assert "json" in result + assert "typing" in result + assert "langchain_openai" in result + assert "bs4" in result + + def test_syntax_error_returns_empty(self): + result = _extract_imports("def broken(") + assert result == set() + + def test_empty_source(self): + result = _extract_imports("") + assert result == set() + + def test_lfx_imports(self): + code = "from lfx.schema.message import Message" + result = _extract_imports(code) + assert "lfx" in result + + def test_try_except_imports(self): + code = """ +try: + from openai import BadRequestError +except ImportError: + pass +""" + result = _extract_imports(code) + assert "openai" in result + + def test_conditional_import_in_function(self): + code = """ +def build_model(self): + from langchain_anthropic import ChatAnthropic + return ChatAnthropic() +""" + result = _extract_imports(code) + assert "langchain_anthropic" in result + + +# =================================================================== +# Unit tests: _import_to_package (now backed by importlib.metadata) +# =================================================================== + + +class TestImportToPackage: + def test_known_mapping_via_metadata(self): + """importlib.metadata.packages_distributions() resolves these.""" + assert _import_to_package("PIL") == "pillow" + assert _import_to_package("bs4") == "beautifulsoup4" + + def test_langchain_mapping_via_metadata(self): + assert _import_to_package("langchain_openai") == "langchain-openai" + assert _import_to_package("langchain_anthropic") == "langchain-anthropic" + + def test_fallback_underscore_to_hyphen(self): + """Unknown packages fall back to replacing _ with -.""" + assert _import_to_package("totally_unknown_pkg_xyz") == "totally-unknown-pkg-xyz" + + def test_simple_package_unchanged(self): + assert _import_to_package("requests") == "requests" + assert _import_to_package("numpy") == "numpy" + + def test_googleapiclient_mapping(self): + assert _import_to_package("googleapiclient") == "google-api-python-client" + + def test_mem0_mapping(self): + assert _import_to_package("mem0") == "mem0ai" + + +# =================================================================== +# Unit tests: dynamic resolution helpers +# =================================================================== + + +class TestDynamicResolution: + def test_import_to_dist_map_returns_mapping(self): + result = _get_import_to_dist_map() + # Returns a read-only MappingProxyType (not a plain dict) + from collections.abc import Mapping + + assert isinstance(result, Mapping) + assert len(result) > 0 + + def test_import_to_dist_map_has_known_entries(self): + result = _get_import_to_dist_map() + assert "PIL" in result + assert "pillow" in result["PIL"] + + def test_lfx_transitive_dists_includes_lfx(self): + dists = _get_lfx_transitive_dists() + assert "lfx" in dists + + def test_lfx_transitive_dists_includes_langchain(self): + dists = _get_lfx_transitive_dists() + assert "langchain" in dists + assert "langchain-core" in dists + + def test_lfx_transitive_dists_includes_pydantic(self): + dists = _get_lfx_transitive_dists() + assert "pydantic" in dists + + def test_lfx_provided_imports_includes_expected(self): + provided = _get_lfx_provided_imports() + import_map = _get_import_to_dist_map() + # Only assert for imports that are resolvable in this environment; + # packages_distributions() can only map installed packages. + expected = ["orjson", "fastapi", "pydantic", "langchain", "pandas", "PIL"] + resolvable = [imp for imp in expected if imp in import_map] + assert len(resolvable) > 0, "No expected imports are resolvable in this environment" + for imp in resolvable: + assert imp in provided, f"{imp} should be provided by lfx" + + def test_lfx_provided_imports_excludes_optional(self): + """Packages not in lfx's dep tree should NOT be in provided.""" + provided = _get_lfx_provided_imports() + # langchain-openai is an optional provider, not a core lfx dep + assert "langchain_openai" not in provided + + +# =================================================================== +# Unit tests: _detect_providers_from_template +# =================================================================== + + +class TestDetectProviders: + def test_no_model_field(self): + template = {"_type": "Component", "code": {"value": ""}} + assert _detect_providers_from_template(template) == set() + + def test_empty_model_field(self): + template = {"model": {"value": []}} + assert _detect_providers_from_template(template) == set() + + def test_openai_provider(self): + template = { + "model": { + "value": [{"provider": "OpenAI", "name": "gpt-4o"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"OpenAI"} + + def test_anthropic_provider(self): + template = { + "model": { + "value": [{"provider": "Anthropic", "name": "claude-3-opus"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"Anthropic"} + + def test_agent_llm_field(self): + template = { + "agent_llm": { + "value": [{"provider": "Google Generative AI", "name": "gemini-pro"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"Google Generative AI"} + + def test_azure_openai_provider(self): + template = { + "model": { + "value": [{"provider": "Azure OpenAI", "name": "gpt-4o"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"Azure OpenAI"} + + def test_amazon_bedrock_provider(self): + template = { + "model": { + "value": [{"provider": "Amazon Bedrock", "name": "anthropic.claude-3"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"Amazon Bedrock"} + + def test_ibm_watsonx_provider(self): + template = { + "model": { + "value": [{"provider": "IBM watsonx.ai", "name": "ibm/granite-13b"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"IBM watsonx.ai"} + + def test_multiple_providers(self): + template = { + "model": { + "value": [{"provider": "OpenAI", "name": "gpt-4o"}], + }, + "embeddings_model": { + "value": [{"provider": "Google Generative AI", "name": "embedding-001"}], + }, + } + result = _detect_providers_from_template(template) + assert result == {"OpenAI", "Google Generative AI"} + + def test_non_dict_value_skipped(self): + template = {"model": {"value": "not a list"}} + assert _detect_providers_from_template(template) == set() + + def test_model_field_not_dict(self): + template = {"model": "not a dict"} + assert _detect_providers_from_template(template) == set() + + +# =================================================================== +# Unit tests: _extract_component_requirements +# =================================================================== + + +class TestExtractComponentRequirements: + def test_lfx_only_component(self): + code = """ +from lfx.schema.message import Message +from lfx.io import Output +""" + node = _make_node("ChatInput", code) + packages, providers = _extract_component_requirements(node) + assert len(packages) == 0 + assert len(providers) == 0 + + def test_stdlib_filtered(self): + code = """ +import os +import json +import re +from typing import Any +from collections import OrderedDict +""" + node = _make_node("Custom", code) + packages, _ = _extract_component_requirements(node) + assert len(packages) == 0 + + def test_lfx_provided_filtered(self): + """Imports that are transitively provided by lfx should not appear as requirements. + + Only tests against imports that are resolvable in this environment, + since packages_distributions() can only map installed packages. + """ + provided = _get_lfx_provided_imports() + candidates = ["orjson", "fastapi", "pandas", "pydantic"] + resolvable = [imp for imp in candidates if imp in provided] + if not resolvable: + pytest.skip("None of the test imports are lfx-provided in this environment") + code = "\n".join(f"import {imp}" for imp in resolvable) + node = _make_node("ChatOutput", code) + packages, _ = _extract_component_requirements(node) + assert len(packages) == 0 + + def test_external_dep_detected(self): + code = """ +from langchain_openai import ChatOpenAI +""" + node = _make_node("OpenAIModel", code) + packages, _ = _extract_component_requirements(node) + assert "langchain-openai" in packages + + def test_provider_detected(self): + node = _make_node( + "LanguageModel", + "from lfx.base.models.model import LCModelComponent", + template_extra={ + "model": { + "value": [{"provider": "Anthropic", "name": "claude-3"}], + }, + }, + ) + _, providers = _extract_component_requirements(node) + assert "Anthropic" in providers + + def test_note_node_handled(self): + node = _make_node("ReadMe", node_type="noteNode") + packages, providers = _extract_component_requirements(node) + # Note nodes have no code, so empty results + assert len(packages) == 0 + assert len(providers) == 0 + + def test_no_code_field(self): + node = _make_node("Empty") + packages, providers = _extract_component_requirements(node) + assert len(packages) == 0 + assert len(providers) == 0 + + def test_module_extra_deps(self): + """Extra runtime deps (lxml, tabulate) must be included for bs4. + + Note: bs4 itself (beautifulsoup4) may or may not appear depending on + whether it's transitively provided by lfx, but the extra runtime deps + must always be included. + """ + code = """ +from bs4 import BeautifulSoup +""" + node = _make_node("URLTool", code) + packages, _ = _extract_component_requirements(node) + assert "lxml" in packages + assert "tabulate" in packages + + def test_langflow_imports_filtered(self): + """Components with langflow imports should NOT list langflow as a dep. + + lfx provides the langflow interfaces at runtime, so langflow/langflow_base + should be filtered out just like lfx itself. + """ + code = """ +from langflow.custom import Component +from langflow.io import MessageTextInput +""" + node = _make_node("LegacyComponent", code) + packages, _ = _extract_component_requirements(node) + assert "langflow" not in packages + assert "langflow-base" not in packages + + +# =================================================================== +# Unit tests: generate_requirements_from_flow +# =================================================================== + + +class TestGenerateRequirementsFromFlow: + def test_empty_flow(self): + flow = _make_flow() + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + def test_lfx_only_flow(self): + node = _make_node( + "ChatInput", + "from lfx.schema.message import Message", + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + def test_external_dep_flow(self): + node = _make_node( + "OpenAIModel", + "from langchain_openai import ChatOpenAI", + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "lfx" in result + assert "langchain-openai" in result + + def test_provider_adds_package(self): + node = _make_node( + "LLM", + "from lfx.base.models.model import LCModelComponent", + template_extra={ + "model": { + "value": [{"provider": "OpenAI", "name": "gpt-4o"}], + }, + }, + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-openai" in result + + def test_note_nodes_skipped(self): + note = _make_node("ReadMe", node_type="noteNode") + component = _make_node( + "ChatInput", + "from lfx.schema.message import Message", + ) + flow = _make_flow(note, component) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + def test_include_lfx_false(self): + flow = _make_flow() + result = generate_requirements_from_flow(flow, include_lfx=False, pin_versions=False) + assert "lfx" not in result + + def test_custom_lfx_package_name(self): + flow = _make_flow() + result = generate_requirements_from_flow(flow, lfx_package="lfx-nightly", pin_versions=False) + assert "lfx-nightly" in result + assert "lfx" not in result + + def test_results_sorted(self): + node1 = _make_node("A", "from langchain_openai import ChatOpenAI") + node2 = _make_node("B", "from bs4 import BeautifulSoup") + flow = _make_flow(node1, node2) + result = generate_requirements_from_flow(flow, pin_versions=False) + # lfx should be first, then sorted extras + assert result[0] == "lfx" + extras = result[1:] + assert extras == sorted(extras) + + def test_deduplication(self): + node1 = _make_node("A", "from langchain_openai import ChatOpenAI") + node2 = _make_node("B", "from langchain_openai import OpenAI") + flow = _make_flow(node1, node2) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result.count("langchain-openai") == 1 + + def test_azure_openai_provider_adds_package(self): + node = _make_node( + "LLM", + "", + template_extra={ + "model": {"value": [{"provider": "Azure OpenAI", "name": "gpt-4o"}]}, + }, + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-openai" in result + + def test_amazon_bedrock_provider_adds_package(self): + node = _make_node( + "LLM", + "", + template_extra={ + "model": {"value": [{"provider": "Amazon Bedrock", "name": "anthropic.claude-3"}]}, + }, + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-aws" in result + + def test_ibm_watsonx_provider_adds_package(self): + node = _make_node( + "LLM", + "", + template_extra={ + "model": {"value": [{"provider": "IBM watsonx.ai", "name": "ibm/granite-13b"}]}, + }, + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-ibm" in result + + def test_multiple_providers(self): + node1 = _make_node( + "LLM", + "", + template_extra={ + "model": {"value": [{"provider": "OpenAI", "name": "gpt-4o"}]}, + }, + ) + node2 = _make_node( + "Embeddings", + "", + template_extra={ + "embeddings_model": {"value": [{"provider": "Google Generative AI", "name": "embedding-001"}]}, + }, + ) + flow = _make_flow(node1, node2) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-openai" in result + assert "langchain-google-genai" in result + + +# =================================================================== +# Unit tests: version pinning +# =================================================================== + + +class TestVersionPinning: + def test_pin_version_installed_package(self): + """Installed packages should get ==X.Y.Z suffix.""" + result = _pin_version("lfx") + assert result.startswith("lfx==") + # Version should be a valid semver-ish string + version_part = result.split("==")[1] + assert len(version_part) > 0 + + def test_pin_version_uninstalled_package(self): + """Packages not installed should return bare name.""" + result = _pin_version("totally-nonexistent-package-xyz-999") + assert result == "totally-nonexistent-package-xyz-999" + + def test_pin_versions_true_by_default(self): + """Default behavior should pin versions.""" + flow = _make_flow() + result = generate_requirements_from_flow(flow) + assert result[0].startswith("lfx==") + + def test_pin_versions_false(self): + """pin_versions=False should return bare names.""" + flow = _make_flow() + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + def test_pinned_output_includes_versions_for_installed_deps(self): + """Installed deps should get pinned; uninstalled deps stay bare.""" + node = _make_node("A", "from langchain_openai import ChatOpenAI") + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=True) + # lfx is installed, so it should be pinned + assert result[0].startswith("lfx==") + # langchain-openai may or may not be installed depending on env; + # just verify it appears in the output + langchain_openai_entries = [r for r in result if r.startswith("langchain-openai")] + assert len(langchain_openai_entries) == 1 + + def test_pinned_txt_output(self): + """generate_requirements_txt should respect pin_versions.""" + flow = _make_flow() + txt_pinned = generate_requirements_txt(flow, pin_versions=True) + txt_unpinned = generate_requirements_txt(flow, pin_versions=False) + assert "==" in txt_pinned + assert "==" not in txt_unpinned + + +# =================================================================== +# Unit tests: generate_requirements_txt +# =================================================================== + + +class TestGenerateRequirementsTxt: + def test_has_header_comments(self): + flow = _make_flow() + txt = generate_requirements_txt(flow) + assert txt.startswith("# Auto-generated") + assert "# This file contains" in txt + + def test_has_trailing_newline(self): + flow = _make_flow() + txt = generate_requirements_txt(flow) + assert txt.endswith("\n") + + def test_packages_on_separate_lines(self): + node = _make_node("A", "from langchain_openai import ChatOpenAI") + flow = _make_flow(node) + txt = generate_requirements_txt(flow, pin_versions=False) + lines = [line for line in txt.strip().split("\n") if line and not line.startswith("#")] + assert "lfx" in lines + assert "langchain-openai" in lines + + +# =================================================================== +# Integration tests: real starter project templates +# =================================================================== + + +class TestStarterProjects: + """Integration tests using actual starter project JSON files.""" + + @pytest.fixture + def basic_prompting_flow(self) -> dict: + path = STARTER_PROJECTS_DIR / "Basic Prompting.json" + if not path.exists(): + pytest.skip("Basic Prompting.json not found") + return json.loads(path.read_text(encoding="utf-8")) + + @pytest.fixture + def simple_agent_flow(self) -> dict: + path = STARTER_PROJECTS_DIR / "Simple Agent.json" + if not path.exists(): + pytest.skip("Simple Agent.json not found") + return json.loads(path.read_text(encoding="utf-8")) + + def test_basic_prompting_includes_anthropic(self, basic_prompting_flow): + """Basic Prompting (Anthropic pre-selected) should need lfx + anthropic deps.""" + result = generate_requirements_from_flow(basic_prompting_flow, pin_versions=False) + assert "lfx" in result + assert "langchain-anthropic" in result + + def test_basic_prompting_with_openai_provider(self, basic_prompting_flow): + """When OpenAI is selected as provider, langchain-openai should be added.""" + for node in basic_prompting_flow["data"]["nodes"]: + node_data = node.get("data", {}) + if node_data.get("type") == "LanguageModelComponent": + template = node_data["node"]["template"] + template["model"] = { + "value": [{"provider": "OpenAI", "name": "gpt-4o-mini"}], + } + break + + result = generate_requirements_from_flow(basic_prompting_flow, pin_versions=False) + assert "lfx" in result + assert "langchain-openai" in result + + def test_basic_prompting_with_anthropic_provider(self, basic_prompting_flow): + """When Anthropic is selected, langchain-anthropic should be added.""" + for node in basic_prompting_flow["data"]["nodes"]: + node_data = node.get("data", {}) + if node_data.get("type") == "LanguageModelComponent": + template = node_data["node"]["template"] + template["model"] = { + "value": [{"provider": "Anthropic", "name": "claude-3-opus"}], + } + break + + result = generate_requirements_from_flow(basic_prompting_flow, pin_versions=False) + assert "lfx" in result + assert "langchain-anthropic" in result + + def test_simple_agent_has_community(self, simple_agent_flow): + """Simple Agent should require langchain-community for its tools.""" + result = generate_requirements_from_flow(simple_agent_flow, pin_versions=False) + assert "lfx" in result + assert "langchain-community" in result + + def test_basic_prompting_from_file(self): + """Test the file-based API.""" + path = STARTER_PROJECTS_DIR / "Basic Prompting.json" + if not path.exists(): + pytest.skip("Basic Prompting.json not found") + result = generate_requirements_from_file(path, pin_versions=False) + assert "lfx" in result + assert "langchain-anthropic" in result + + def test_lfx_nightly_package_name(self, basic_prompting_flow): + """Test specifying lfx-nightly as the package name.""" + result = generate_requirements_from_flow( + basic_prompting_flow, + lfx_package="lfx-nightly", + pin_versions=False, + ) + assert result[0] == "lfx-nightly" + assert "lfx" not in result + + def test_pinned_output_from_starter(self, basic_prompting_flow): + """Default (pinned) output should have version specifiers.""" + result = generate_requirements_from_flow(basic_prompting_flow) + assert result[0].startswith("lfx==") + + +# =================================================================== +# Data integrity tests +# =================================================================== + + +class TestDataIntegrity: + """Verify dynamic resolution and mapping tables are consistent.""" + + def test_known_langchain_packages_resolved_by_metadata(self): + """importlib.metadata should correctly resolve common langchain packages.""" + expected = { + "langchain_openai": "langchain-openai", + "langchain_anthropic": "langchain-anthropic", + "langchain_ollama": "langchain-ollama", + } + for import_name, pkg_name in expected.items(): + assert _import_to_package(import_name) == pkg_name + + def test_module_extra_deps_values_are_lists(self): + for mod, deps in MODULE_EXTRA_DEPS.items(): + assert isinstance(deps, list), f"Extra deps for {mod} should be a list" + + +class TestResolveProviderPackages: + """Verify dynamic provider resolution via inspect.""" + + @staticmethod + def _skip_if_provider_not_loaded(provider_name: str): + """Skip test if the provider's component class isn't loaded in MODEL_PROVIDERS_DICT. + + MODEL_PROVIDERS_DICT only contains providers whose packages are + installed in the current environment. + """ + try: + from lfx.base.models.model_input_constants import MODEL_PROVIDERS_DICT + except ImportError: + pytest.skip("MODEL_PROVIDERS_DICT not available") + if provider_name not in MODEL_PROVIDERS_DICT: + pytest.skip(f"{provider_name} component not loaded (package not installed)") + + def test_openai_provider_resolves(self): + self._skip_if_provider_not_loaded("OpenAI") + packages = _resolve_provider_packages("OpenAI") + assert "langchain-openai" in packages + + def test_anthropic_provider_resolves(self): + self._skip_if_provider_not_loaded("Anthropic") + packages = _resolve_provider_packages("Anthropic") + assert "langchain-anthropic" in packages + + def test_amazon_bedrock_provider_resolves(self): + self._skip_if_provider_not_loaded("Amazon Bedrock") + packages = _resolve_provider_packages("Amazon Bedrock") + assert "langchain-aws" in packages + + def test_google_provider_resolves(self): + self._skip_if_provider_not_loaded("Google Generative AI") + packages = _resolve_provider_packages("Google Generative AI") + assert "langchain-google-genai" in packages + + def test_ollama_provider_resolves(self): + self._skip_if_provider_not_loaded("Ollama") + packages = _resolve_provider_packages("Ollama") + assert "langchain-ollama" in packages + + def test_unknown_provider_returns_empty(self): + packages = _resolve_provider_packages("NonexistentProvider") + assert packages == set() + + def test_function_level_imports_captured(self): + """Verify imports inside function bodies (e.g. build_model) are captured. + + This is critical because many provider components use lazy imports + inside methods like ``build_model()`` rather than at module level. + """ + self._skip_if_provider_not_loaded("Amazon Bedrock") + packages = _resolve_provider_packages("Amazon Bedrock") + # boto3 and langchain_aws are imported inside build_model(), not at module level + assert "boto3" in packages + assert "langchain-aws" in packages + + def test_all_registered_providers_resolve_to_packages(self): + """Every provider in MODEL_PROVIDERS_DICT should resolve to at least one package.""" + try: + from lfx.base.models.model_input_constants import MODEL_PROVIDERS_DICT + except ImportError: + pytest.skip("MODEL_PROVIDERS_DICT not available") + for provider_name in MODEL_PROVIDERS_DICT: + packages = _resolve_provider_packages(provider_name) + assert len(packages) > 0, f"Provider {provider_name} resolved to no packages" + + +class TestResolveEmbeddingProviderPackages: + """Verify embedding provider resolution via unified models metadata.""" + + def test_openai_embedding_resolves(self): + packages = _resolve_embedding_provider_packages("OpenAI") + assert "langchain-openai" in packages + + def test_google_embedding_resolves(self): + packages = _resolve_embedding_provider_packages("Google Generative AI") + assert "langchain-google-genai" in packages + + def test_ollama_embedding_resolves(self): + packages = _resolve_embedding_provider_packages("Ollama") + assert "langchain-ollama" in packages + + def test_unknown_provider_returns_empty(self): + packages = _resolve_embedding_provider_packages("NonexistentProvider") + assert packages == set() + + def test_language_only_provider_returns_empty(self): + """Providers without embedding support should return empty (not warn).""" + packages = _resolve_embedding_provider_packages("Anthropic") + assert packages == set() + + def test_ibm_watsonx_embedding_resolves(self): + packages = _resolve_embedding_provider_packages("IBM WatsonX") + assert "langchain-ibm" in packages + + def test_all_embedding_providers_resolve(self): + """Every provider in EMBEDDING_PROVIDER_CLASS_MAPPING should resolve to a package.""" + from lfx.base.models.unified_models import EMBEDDING_PROVIDER_CLASS_MAPPING + + for provider in EMBEDDING_PROVIDER_CLASS_MAPPING: + packages = _resolve_embedding_provider_packages(provider) + assert len(packages) > 0, f"Embedding provider '{provider}' resolved to no packages" + + def test_embedding_only_flow(self): + """A flow with only an embedding model should still get provider packages.""" + node = _make_node( + "EmbeddingModel", + "from lfx.base.embeddings.model import LCEmbeddingsModel", + template_extra={ + "model": {"value": [{"provider": "OpenAI", "name": "text-embedding-3-small"}]}, + }, + ) + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert "langchain-openai" in result + + +# =================================================================== +# Error handling tests +# =================================================================== + + +class TestErrorHandling: + """Test error handling for edge cases.""" + + def test_generate_requirements_from_file_not_found(self, tmp_path): + """FileNotFoundError should propagate for missing files.""" + with pytest.raises(FileNotFoundError): + generate_requirements_from_file(tmp_path / "nonexistent.json") + + def test_generate_requirements_from_file_invalid_json(self, tmp_path): + """JSONDecodeError should propagate for invalid JSON.""" + bad_file = tmp_path / "bad.json" + bad_file.write_text("not json at all", encoding="utf-8") + with pytest.raises(json.JSONDecodeError): + generate_requirements_from_file(bad_file) + + def test_generate_requirements_from_file_wrong_structure(self, tmp_path): + """A valid JSON file that isn't a flow should still produce a result (just lfx).""" + wrong_file = tmp_path / "wrong.json" + wrong_file.write_text('{"not": "a flow"}', encoding="utf-8") + result = generate_requirements_from_file(wrong_file, pin_versions=False) + assert result == ["lfx"] + + def test_flow_with_empty_code_value(self): + """A node with an empty code string should not crash.""" + node = _make_node("Empty", "") + flow = _make_flow(node) + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + def test_flow_with_malformed_node(self): + """Nodes with missing expected fields should be handled gracefully.""" + flow = {"data": {"nodes": [{"type": "genericNode", "data": {}}]}} + result = generate_requirements_from_flow(flow, pin_versions=False) + assert result == ["lfx"] + + +# =================================================================== +# Typer CLI tests: lfx requirements +# =================================================================== + + +class TestTyperRequirementsCommand: + """Tests for the typer-based ``lfx requirements`` CLI command.""" + + @pytest.fixture + def runner(self): + from typer.testing import CliRunner + + return CliRunner() + + @pytest.fixture + def app(self): + from lfx.__main__ import app + + return app + + @pytest.fixture + def flow_file(self, tmp_path): + flow = _make_flow(_make_node("Simple", "import lfx")) + path = tmp_path / "flow.json" + path.write_text(json.dumps(flow), encoding="utf-8") + return path + + def test_happy_path_stdout(self, runner, app, flow_file): + result = runner.invoke(app, ["requirements", str(flow_file), "--no-pin"]) + assert result.exit_code == 0 + assert "lfx" in result.output + + def test_output_flag_writes_file(self, runner, app, flow_file, tmp_path): + out = tmp_path / "requirements.txt" + result = runner.invoke(app, ["requirements", str(flow_file), "-o", str(out), "--no-pin"]) + assert result.exit_code == 0 + assert out.exists() + assert "lfx" in out.read_text(encoding="utf-8") + assert "Requirements written to" in result.output + + def test_no_lfx_flag(self, runner, app, flow_file): + result = runner.invoke(app, ["requirements", str(flow_file), "--no-lfx", "--no-pin"]) + assert result.exit_code == 0 + # With --no-lfx and only lfx imports, no packages should appear after header + lines = [line for line in result.output.strip().split("\n") if line and not line.startswith("#")] + assert "lfx" not in lines + + def test_no_pin_flag(self, runner, app, flow_file): + result = runner.invoke(app, ["requirements", str(flow_file), "--no-pin"]) + assert result.exit_code == 0 + assert "==" not in result.output + + def test_default_pins_versions(self, runner, app, flow_file): + result = runner.invoke(app, ["requirements", str(flow_file)]) + assert result.exit_code == 0 + assert "lfx==" in result.output + + def test_lfx_package_flag(self, runner, app, flow_file): + result = runner.invoke(app, ["requirements", str(flow_file), "--lfx-package", "lfx-nightly", "--no-pin"]) + assert result.exit_code == 0 + assert "lfx-nightly" in result.output + # Should not contain bare "lfx" as a separate line + lines = [line for line in result.output.strip().split("\n") if line and not line.startswith("#")] + assert "lfx" not in lines + + def test_file_not_found(self, runner, app, tmp_path): + result = runner.invoke(app, ["requirements", str(tmp_path / "missing.json")]) + assert result.exit_code == 1 + assert "Error" in result.output + + def test_invalid_json(self, runner, app, tmp_path): + bad = tmp_path / "bad.json" + bad.write_text("not json", encoding="utf-8") + result = runner.invoke(app, ["requirements", str(bad)]) + assert result.exit_code == 1 + assert "Error" in result.output