test_importable_all_via_subprocess() — langchain Function Reference

Architecture documentation for the test_importable_all_via_subprocess() function in test_imports.py from the langchain codebase.

Function python LangChainCore MessageInterface

Entity Profile

LangChainCore→ MessageInterface→ test_importable_all_via_subprocess() — langchain Function Reference

Dependency Diagram

graph TD
  f9c5a24b_d4d5_8e0b_43f9_ef0bc6b9eb1b["test_importable_all_via_subprocess()"]
  9facfe9c_3d96_0e99_01dc_62f4bbfd8a09["test_imports.py"]
  f9c5a24b_d4d5_8e0b_43f9_ef0bc6b9eb1b -->|defined in| 9facfe9c_3d96_0e99_01dc_62f4bbfd8a09
  style f9c5a24b_d4d5_8e0b_43f9_ef0bc6b9eb1b fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/test_imports.py lines 34–60

def test_importable_all_via_subprocess() -> None:
    """Test import in isolation.

    !!! note
        ImportErrors due to circular imports can be raised for one sequence of imports
        but not another.
    """
    module_names = []
    for path in Path("../core/langchain_core/").glob("*"):
        module_name = path.stem
        if (
            not module_name.startswith(".")
            and path.suffix != ".typed"
            and module_name != "pydantic_v1"
        ):
            module_names.append(module_name)

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        futures = [
            executor.submit(try_to_import, module_name) for module_name in module_names
        ]
        for future in concurrent.futures.as_completed(futures):
            result = future.result()  # Will raise an exception if the callable raised
            code, module_name = result
            if code != 0:
                msg = f"Failed to import {module_name}."
                raise ValueError(msg)