orchestr8.execution_runtime

create_project function

Create a uv project from a python script.

create_project(
    "/path/to/script.py",
    id="project_id", # Not required, defaults to script name.
    requirements=["package=0.0.0"], # Set `True` to auto generate
)

Refer orchestr8.execution_runtime.package_utils.generate_requirements to know more about auto-generation of requirements from python source file and dependency_overrides.

Parameters:

Name	Type	Description	Default
__script	str | Path	The python script to create project from	required
id	str | None	The id to use for the project. Defaults to the script name	None
requirements	List[str] | Literal[True] | None	List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate	None
dependency_overrides	Dict[str, Dependency] | None	Depedency overrides. Applicable only if requirements=True	None
force	bool	Whether to force create the project even if it's already available	False

Raises:

Type	Description
FileNotFoundError	If the script file is not found

Source code in orchestr8/execution_runtime/__init__.py

def create_project(
    __script: str | Path,
    *,
    id: str | None = None,  # noqa: A002
    requirements: List[str] | Literal[True] | None = None,
    dependency_overrides: Dict[str, Dependency] | None = None,
    force: bool = False,
) -> None:
    """
    Create a uv project from a python script.

    ```python
    create_project(
        "/path/to/script.py",
        id="project_id", # Not required, defaults to script name.
        requirements=["package=0.0.0"], # Set `True` to auto generate
    )
    ```

    Refer `orchestr8.execution_runtime.package_utils.generate_requirements` to know more
    about auto-generation of requirements from python source file and `dependency_overrides`.

    Args:
        __script: The python script to create project from
        id: The id to use for the project. Defaults to the script name
        requirements: List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate
        dependency_overrides: Depedency overrides. Applicable only if requirements=True
        force: Whether to force create the project even if it's already available

    Raises:
        FileNotFoundError: If the script file is not found

    """
    project_id = id or Path(__script).stem
    project_dir = RUNTIME_PROJECTS_DIR / project_id
    if project_exists(project_id):
        if not force:
            raise FileExistsError(f"Project with id {project_id!r} already exists." " Use `force=True` to overwrite.")
        shutil.rmtree(project_dir, ignore_errors=True)

    script, requirements = _validate_script_and_requirements(
        __script, requirements=requirements, dependency_overrides=dependency_overrides
    )

    shell = Shell(workdir=RUNTIME_PROJECTS_DIR)

    # Creating a new project
    shell.run("uv", "init", "--vcs", "none", "--no-readme", "--no-workspace", "--no-pin-python", project_id)

    # Deleting default hello.py created by uv
    (project_dir / "hello.py").unlink(missing_ok=True)
    # Copy the script to `main.py` in the project directory
    (project_dir / "main.py").write_bytes(script.read_bytes())

    if requirements:
        # Add dependencies to the project
        shell.run("uv", "add", "--directory", project_id, *requirements)

        # Create a requirements.txt file for isolated runtime virtual environments
        shell.run(
            "uv",
            "export",
            "--directory",
            project_id,
            "--no-editable",
            "--no-emit-project",
            "--no-header",
            "--locked",
            "--format",
            "requirements-txt",
            "-o",
            "requirements.txt",
        )

---

create_execution_runtime function

Create execution runtime instance for running Python scripts and created projects.

runtime = create_execution_runtime(isolate=True, python_tag="3.10-alpine3.20")

# Run a script with auto-generated requirements
output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

# Run a script with pre-defined requirements
output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

# Run a created project
output = runtime.run_project('project_id', '--param1', 'value1')

Parameters:

Name	Type	Description	Default
isolate	bool	Whether to isolate the runtime. Defaults to False	False
python_tag	str | None	Python image tag. Defaults to 3.10-alpine3.20. Applicable, if isolate is True	None
docker_config	Any	Docker client configuration. Applicable, if isolate is True	{}

Returns:

Type	Description
ExecutionRuntime | IsolatedExecutionRuntime	Runtime instance

Source code in orchestr8/execution_runtime/__init__.py

def create_execution_runtime(
    isolate: bool = False, python_tag: str | None = None, **docker_config: Any
) -> ExecutionRuntime | IsolatedExecutionRuntime:
    """
    Create execution runtime instance for running Python scripts and created projects.

    ```python
    runtime = create_execution_runtime(isolate=True, python_tag="3.10-alpine3.20")

    # Run a script with auto-generated requirements
    output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

    # Run a script with pre-defined requirements
    output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

    # Run a created project
    output = runtime.run_project('project_id', '--param1', 'value1')
    ```

    Args:
        isolate: Whether to isolate the runtime. Defaults to False
        python_tag: Python image tag. Defaults to 3.10-alpine3.20. Applicable, if `isolate is True`
        docker_config: Docker client configuration. Applicable, if `isolate is True`

    Returns:
        Runtime instance
    """
    if isolate:
        if (proc := Path("/proc/1/cgroup")).exists() and "docker" in proc.read_text():
            ExecutionRuntime.logger.info("Already inside a container. Switched to host.")
            return ExecutionRuntime()

        IsolatedExecutionRuntime.logger.info("Setting up the instance")
        return IsolatedExecutionRuntime(python_tag=python_tag, **docker_config)

    ExecutionRuntime.logger.info("Setting up the instance")
    return ExecutionRuntime()

---

ExecutionRuntime (host-based)

Interface for running Python scripts and created projects in the host machine.

# Create a host execution runtime instance
runtime = ExecutionRuntime()

# Run a script with auto-generated requirements
output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

# Run a script with pre-defined requirements
output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

# Run a created project
output = runtime.run_project('project_id', '--param1', 'value1')

Source code in orchestr8/execution_runtime/__init__.py

class ExecutionRuntime(Logger):
    """
    Interface for running Python scripts and created projects in the host machine.

    ```python
    # Create a host execution runtime instance
    runtime = ExecutionRuntime()

    # Run a script with auto-generated requirements
    output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

    # Run a script with pre-defined requirements
    output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

    # Run a created project
    output = runtime.run_project('project_id', '--param1', 'value1')
    ```
    """

    def __init__(self) -> None:
        """Initialize the runtime."""
        self.__shell = Shell(workdir=RUNTIME_PROJECTS_DIR)
        if os.name == "nt":
            self._py_exe_loc = "Scripts"  # windows
        else:
            self._py_exe_loc = "bin"  # unix

    @property
    def shell(self) -> Shell:
        """The shell instance."""
        return self.__shell

    def run_project(self, __id: str, *args: SupportsStr, env: Dict[str, str] | None = None) -> str | None:
        """
        Run the project with the given id.

        Args:
            __id: The id of the project to run
            args: The arguments to pass to the script associated with the project
            env: Environment variables required by the project

        Returns:
            The output of the script

        Raises:
            FileNotFoundError: If the project with the given id does not exist
        """
        if not project_exists(__id):
            raise FileNotFoundError(f"Project with id {__id!r} not found.")

        project_dir = RUNTIME_PROJECTS_DIR / __id
        python_exe = str(project_dir / ".venv" / self._py_exe_loc / "python")
        main_py = str(project_dir / "main.py")
        return self.__shell.run(python_exe, main_py, *list(map(str, args)), env=env)

    def run_script(
        self,
        __script: str | Path | BytesIO,
        *args: SupportsStr,
        env: Dict[str, str] | None = None,
        requirements: List[str] | Literal[True] | None = None,
        dependency_overrides: Dict[str, Dependency] | None = None,
    ) -> str | None:
        """
        Run a python script inside on-demand environment.

        Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

        To know more about auto-generation of requirements from python source
        file, refer `orchestr8.execution_runtime.package_utils.generate_requirements`.

        Args:
            __script: The python script to run
            args: The arguments to pass to the script
            env: Environment variables required by the script
            requirements: List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate
            dependency_overrides: Depedency overrides. Applicable only if requirements=True

        Returns:
            The output of the script
        """
        script, requirements = _validate_script_and_requirements(
            script=__script, requirements=requirements, dependency_overrides=dependency_overrides
        )

        self.logger.info(f"Running script {script.name!r}")
        cmd = ["uv", "run", "--no-project", "--quiet"]
        if requirements:
            cmd.extend(["--with", *requirements])
        cmd.append(str(script.absolute()))
        if args:
            cmd.extend(list(map(str, args)))
        return self.__shell.run(*cmd, env=env)

shell: Shell property

The shell instance.

run_project

Run the project with the given id.

Parameters:

Name	Type	Description	Default
__id	str	The id of the project to run	required
args	SupportsStr	The arguments to pass to the script associated with the project	()
env	Dict[str, str] | None	Environment variables required by the project	None

Returns:

Type	Description
str | None	The output of the script

Raises:

Type	Description
FileNotFoundError	If the project with the given id does not exist

Source code in orchestr8/execution_runtime/__init__.py

def run_project(self, __id: str, *args: SupportsStr, env: Dict[str, str] | None = None) -> str | None:
    """
    Run the project with the given id.

    Args:
        __id: The id of the project to run
        args: The arguments to pass to the script associated with the project
        env: Environment variables required by the project

    Returns:
        The output of the script

    Raises:
        FileNotFoundError: If the project with the given id does not exist
    """
    if not project_exists(__id):
        raise FileNotFoundError(f"Project with id {__id!r} not found.")

    project_dir = RUNTIME_PROJECTS_DIR / __id
    python_exe = str(project_dir / ".venv" / self._py_exe_loc / "python")
    main_py = str(project_dir / "main.py")
    return self.__shell.run(python_exe, main_py, *list(map(str, args)), env=env)

run_script

Run a python script inside on-demand environment.

Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

To know more about auto-generation of requirements from python source file, refer orchestr8.execution_runtime.package_utils.generate_requirements.

Parameters:

Name	Type	Description	Default
__script	str | Path | BytesIO	The python script to run	required
args	SupportsStr	The arguments to pass to the script	()
env	Dict[str, str] | None	Environment variables required by the script	None
requirements	List[str] | Literal[True] | None	List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate	None
dependency_overrides	Dict[str, Dependency] | None	Depedency overrides. Applicable only if requirements=True	None

Returns:

Type	Description
str | None	The output of the script

Source code in orchestr8/execution_runtime/__init__.py

def run_script(
    self,
    __script: str | Path | BytesIO,
    *args: SupportsStr,
    env: Dict[str, str] | None = None,
    requirements: List[str] | Literal[True] | None = None,
    dependency_overrides: Dict[str, Dependency] | None = None,
) -> str | None:
    """
    Run a python script inside on-demand environment.

    Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

    To know more about auto-generation of requirements from python source
    file, refer `orchestr8.execution_runtime.package_utils.generate_requirements`.

    Args:
        __script: The python script to run
        args: The arguments to pass to the script
        env: Environment variables required by the script
        requirements: List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate
        dependency_overrides: Depedency overrides. Applicable only if requirements=True

    Returns:
        The output of the script
    """
    script, requirements = _validate_script_and_requirements(
        script=__script, requirements=requirements, dependency_overrides=dependency_overrides
    )

    self.logger.info(f"Running script {script.name!r}")
    cmd = ["uv", "run", "--no-project", "--quiet"]
    if requirements:
        cmd.extend(["--with", *requirements])
    cmd.append(str(script.absolute()))
    if args:
        cmd.extend(list(map(str, args)))
    return self.__shell.run(*cmd, env=env)

---

IsolatedExecutionRuntime

Interface for running Python scripts and created projects in isolated environments.

# Create an isolated execution runtime instance
runtime = IsolatedExecutionRuntime(isolate=True, python_tag="3.10-alpine3.20")

# Run a script with auto-generated requirements
output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

# Run a script with pre-defined requirements
output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

# Run a created project
output = runtime.run_project('project_id', '--param1', 'value1')

Source code in orchestr8/execution_runtime/__init__.py

class IsolatedExecutionRuntime(Logger):
    """
    Interface for running Python scripts and created projects in isolated environments.

    ```python
    # Create an isolated execution runtime instance
    runtime = IsolatedExecutionRuntime(isolate=True, python_tag="3.10-alpine3.20")

    # Run a script with auto-generated requirements
    output = runtime.run_script('script1.py', '-p1', 'value1', requirements=True)

    # Run a script with pre-defined requirements
    output = runtime.run_script('script2.py', requirements=["requests==2.25.1"])

    # Run a created project
    output = runtime.run_project('project_id', '--param1', 'value1')
    ```
    """

    def __init__(self, *, python_tag: str | None = None, **docker_config: Any) -> None:
        """
        Initialize the runtime.

        :param python_tag: Python image tag. Defaults to 3.10-alpine3.20.
        :type python_tag: str | None
        :param docker_config: Docker client configuration.
        :type docker_config: dict[str, Any]
        """
        minor, major = sys.version_info[:2]
        python_tag = python_tag or f"{minor}.{major}-alpine3.20"

        if not (m := re_match(r"^(\d+\.\d+)", python_tag)):
            raise ValueError(f"Invalid python tag: {python_tag!r}")

        python_version: str = m.group(1)
        self.__venvs_dir = ISOLATED_RUNTIME_VENVS_VOLUME_DIR / python_version
        if not self.__venvs_dir.is_dir():
            self.__venvs_dir.mkdir(exist_ok=True)

        py_image = f"python:{python_tag}"
        self.__sb_client = SandboxClient(**docker_config)
        if not self.__sb_client.image_exists(py_image) and not (
            self.__sb_client.image_exists(py_image, where="registry")
        ):
            raise Exception(f"Python image {py_image!r} not found on docker hub.")

        executor_image = f"orchestr8-runtime:py-{python_tag}"
        if not self.__sb_client.image_exists(executor_image):
            self.__sb_client.build_image(
                image=executor_image, dockerfile=BytesIO(EXECUTOR_IMAGE_TEMPLATE.format(python_tag=python_tag).encode())
            )

        # Project directory inside the container, host project directory is mounted
        projs_dir_mount = "/var/o8-projects"
        container = self.__sb_client.run_container(
            executor_image,
            volumes={
                RUNTIME_PROJECTS_DIR: {"bind": projs_dir_mount, "mode": "ro"},
                self.__venvs_dir: {"bind": "/var/o8-venvs", "mode": "rw"},
            },
            detach=True,
            stderr=True,
            remove=True,
        )
        self.__shell = IsolatedShell(container=container, workdir=projs_dir_mount)

    @property
    def shell(self) -> IsolatedShell:
        """The shell instance."""
        return self.__shell

    def run_project(self, __id: str, *args: SupportsStr, env: Dict[str, str] | None = None) -> str | None:
        """
        Run the project with the given id.

        Args:
            __id: The id of the project to run
            args: The arguments to pass to the script associated with the project
            env: Environment variables required by the project

        Returns:
            The output of the script

        Raises:
            FileNotFoundError: If the project with the given id does not exist
        """
        if not project_exists(__id):
            raise FileNotFoundError(f"Project with id {__id!r} not found.")

        venv_dir_mount = Path("/var") / "o8-venvs" / __id
        proj_dir_mount = Path("/var") / "o8-projects" / __id
        if not (self.__venvs_dir / __id).exists():
            self.logger.info(f"Creating virtual environment for project {__id!r}")
            self.__shell.run("uv", "venv", "--no-project", venv_dir_mount.as_posix())

            if (RUNTIME_PROJECTS_DIR / __id / "requirements.txt").exists():
                self.logger.info(f"Installing requirements for project {__id!r}")
                self.__shell.run(
                    "uv",
                    "pip",
                    "install",
                    "-r",
                    (proj_dir_mount / "requirements.txt").as_posix(),
                    env={"VIRTUAL_ENV": venv_dir_mount.as_posix()},
                )

        python_exe = (venv_dir_mount / "bin" / "python").as_posix()
        main_py = (proj_dir_mount / "main.py").as_posix()
        return self.__shell.run(python_exe, main_py, *list(map(str, args)), env=env)

    def run_script(
        self,
        __script: str | Path | BytesIO,
        *args: SupportsStr,
        env: Dict[str, str] | None = None,
        requirements: List[str] | Literal[True] | None = None,
        dependency_overrides: Dict[str, Dependency] | None = None,
    ) -> str | None:
        """
        Run a python script inside on-demand environment.

        Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

        To know more about auto-generation of requirements from python source
        file, refer `orchestr8.execution_runtime.package_utils.generate_requirements`.

        Args:
            __script: The python script to run.
            args: The arguments to pass to the script
            env: Environment variables required by the script
            requirements: List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate
            dependency_overrides: Depedency overrides. Applicable only if requirements=True

        Returns:
            The output of the script
        """
        script, requirements = _validate_script_and_requirements(
            script=__script, requirements=requirements, dependency_overrides=dependency_overrides
        )
        self.__sb_client.copy_path_to_container(self.__shell.container.id, src=script, target="/tmp")  # noqa: S108

        self.logger.info(f"Running script {script.name!r}")
        cmd = ["uv", "run", "--no-project", "--quiet"]
        if requirements:
            cmd.extend(["--with", *requirements])
        cmd.append(f"/tmp/{script.name}")  # noqa: S108
        if args:
            cmd.extend(list(map(str, args)))
        return self.__shell.run(*cmd, env=env)

shell: IsolatedShell property

The shell instance.

run_project

Run the project with the given id.

Parameters:

Name	Type	Description	Default
__id	str	The id of the project to run	required
args	SupportsStr	The arguments to pass to the script associated with the project	()
env	Dict[str, str] | None	Environment variables required by the project	None

Returns:

Type	Description
str | None	The output of the script

Raises:

Type	Description
FileNotFoundError	If the project with the given id does not exist

Source code in orchestr8/execution_runtime/__init__.py

def run_project(self, __id: str, *args: SupportsStr, env: Dict[str, str] | None = None) -> str | None:
    """
    Run the project with the given id.

    Args:
        __id: The id of the project to run
        args: The arguments to pass to the script associated with the project
        env: Environment variables required by the project

    Returns:
        The output of the script

    Raises:
        FileNotFoundError: If the project with the given id does not exist
    """
    if not project_exists(__id):
        raise FileNotFoundError(f"Project with id {__id!r} not found.")

    venv_dir_mount = Path("/var") / "o8-venvs" / __id
    proj_dir_mount = Path("/var") / "o8-projects" / __id
    if not (self.__venvs_dir / __id).exists():
        self.logger.info(f"Creating virtual environment for project {__id!r}")
        self.__shell.run("uv", "venv", "--no-project", venv_dir_mount.as_posix())

        if (RUNTIME_PROJECTS_DIR / __id / "requirements.txt").exists():
            self.logger.info(f"Installing requirements for project {__id!r}")
            self.__shell.run(
                "uv",
                "pip",
                "install",
                "-r",
                (proj_dir_mount / "requirements.txt").as_posix(),
                env={"VIRTUAL_ENV": venv_dir_mount.as_posix()},
            )

    python_exe = (venv_dir_mount / "bin" / "python").as_posix()
    main_py = (proj_dir_mount / "main.py").as_posix()
    return self.__shell.run(python_exe, main_py, *list(map(str, args)), env=env)

run_script

Run a python script inside on-demand environment.

Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

To know more about auto-generation of requirements from python source file, refer orchestr8.execution_runtime.package_utils.generate_requirements.

Parameters:

Name	Type	Description	Default
__script	str | Path | BytesIO	The python script to run.	required
args	SupportsStr	The arguments to pass to the script	()
env	Dict[str, str] | None	Environment variables required by the script	None
requirements	List[str] | Literal[True] | None	List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate	None
dependency_overrides	Dict[str, Dependency] | None	Depedency overrides. Applicable only if requirements=True	None

Returns:

Type	Description
str | None	The output of the script

Source code in orchestr8/execution_runtime/__init__.py

def run_script(
    self,
    __script: str | Path | BytesIO,
    *args: SupportsStr,
    env: Dict[str, str] | None = None,
    requirements: List[str] | Literal[True] | None = None,
    dependency_overrides: Dict[str, Dependency] | None = None,
) -> str | None:
    """
    Run a python script inside on-demand environment.

    Reference: https://docs.astral.sh/uv/guides/scripts/#running-a-script-with-dependencies

    To know more about auto-generation of requirements from python source
    file, refer `orchestr8.execution_runtime.package_utils.generate_requirements`.

    Args:
        __script: The python script to run.
        args: The arguments to pass to the script
        env: Environment variables required by the script
        requirements: List of requirement strings (e.g., ["package==1.0.0"]) or True to auto-generate
        dependency_overrides: Depedency overrides. Applicable only if requirements=True

    Returns:
        The output of the script
    """
    script, requirements = _validate_script_and_requirements(
        script=__script, requirements=requirements, dependency_overrides=dependency_overrides
    )
    self.__sb_client.copy_path_to_container(self.__shell.container.id, src=script, target="/tmp")  # noqa: S108

    self.logger.info(f"Running script {script.name!r}")
    cmd = ["uv", "run", "--no-project", "--quiet"]
    if requirements:
        cmd.extend(["--with", *requirements])
    cmd.append(f"/tmp/{script.name}")  # noqa: S108
    if args:
        cmd.extend(list(map(str, args)))
    return self.__shell.run(*cmd, env=env)

---

package_utils

Dependency

Dependency specification with package name and version requirements.

Source code in orchestr8/execution_runtime/package_utils.py

class Dependency(TypedDict):
    """Dependency specification with package name and version requirements."""

    package_name: NotRequired[str]
    specifiers: NotRequired[List[Specifier]]

ReleaseInfo

Release information

Source code in orchestr8/execution_runtime/package_utils.py

class ReleaseInfo(TypedDict):
    """Release information"""

    ids: List[str]
    latest: str

Specifier

Version specifier.

Represents a package version specifier with an operator and version string.

Source code in orchestr8/execution_runtime/package_utils.py

class Specifier(TypedDict):
    """
    Version specifier.

    Represents a package version specifier with an operator and version string.
    """

    version: str
    op: Literal["==", ">=", "<=", ">", "<", "~=", "===", "!="]

extract_module_names

Extract imported module names from a Python script or AST Module.

Analyzes Python source code to find all imported module names, including those within conditional statements (if/else) and try/except blocks. For 'import' statements, only the root module name is extracted (e.g., 'pandas' from 'pandas.DataFrame'). For 'from' imports, the module being imported from is extracted.

```python script = Path('example.py') # Contains: import pandas as pd; from numpy import array extract_module_names(script)

output: ['pandas', 'numpy']

Args: script: Either a Path to a Python file or an ast.Module object

Returns: List of unique module names found in the script

Raises: FileNotFoundError: If the script path does not exist

Source code in orchestr8/execution_runtime/package_utils.py

def extract_module_names(script: Path | ast.Module) -> List[str]:  # noqa: C901
    """
    Extract imported module names from a Python script or AST Module.

    Analyzes Python source code to find all imported module names, including those within
    conditional statements (if/else) and try/except blocks. For 'import' statements,
    only the root module name is extracted (e.g., 'pandas' from 'pandas.DataFrame').
    For 'from' imports, the module being imported from is extracted.


    ```python
    script = Path('example.py')  # Contains: import pandas as pd; from numpy import array
    extract_module_names(script)
    # output: ['pandas', 'numpy']

    Args:
        script: Either a Path to a Python file or an ast.Module object

    Returns:
        List of unique module names found in the script

    Raises:
        FileNotFoundError: If the script path does not exist
    """

    class ModuleExtractor(ast.NodeVisitor):
        def __init__(self) -> None:
            self.module_names: Set[str] = set()

        def visit_Import(self, node: ast.Import) -> None:
            for alias in node.names:
                self.module_names.add(alias.name.split(".")[0])

        def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
            if node.level == 0 and node.module:
                self.module_names.add(node.module)

        def visit_If(self, node: ast.If) -> None:
            self.visit(node.test)

            for stmt in node.body:
                self.visit(stmt)

            if node.orelse:
                for stmt in node.orelse:
                    self.visit(stmt)

        def visit_Try(self, node: ast.Try) -> None:
            for stmt in node.body:
                self.visit(stmt)

            for handler in node.handlers:
                for stmt in handler.body:
                    self.visit(stmt)

            if node.orelse:
                for stmt in node.orelse:
                    self.visit(stmt)

            if node.finalbody:
                for stmt in node.finalbody:
                    self.visit(stmt)

    if isinstance(script, Path):
        if not script.is_file():
            raise FileNotFoundError(f"Script {script!r} not found.")
        script = ast.parse(script.read_bytes())

    extractor = ModuleExtractor()
    extractor.visit(script)
    return list(extractor.module_names)

generate_requirements

Generate a list of Python package requirements from module names.

For extracting module names from a Python script, use:

from orchestr8.runtime.utils import extract_module_names
module_names = extract_module_names("/path/to/script.py")

The function follows a systematic approach to resolving package dependencies:

First, standard library modules are automatically excluded from the processing. The remaining modules are then searched among globally installed packages. When a module is found in installed packages, its package name and version are used.

If a module is not installed locally, the function searches the PyPI package index. When a package is found on PyPI, its latest version is utilized. If no package can be found, the function raises an error.

The function provides flexibility through the dependency_overrides parameter. This is particularly useful in scenarios where: - The package name differs from the module name - Custom version specifications are required

A prime example is the yaml module, which corresponds to the PyYAML package. Through dependency overrides, you can specify custom package names or version constraints.

The function performs comprehensive validation: - Checks if specified versions exist in package releases - Validates version specifiers against available package versions - Ensures precise and correct dependency specification

generate_requirements(['pandas', 'numpy'])
# output: ['pandas==2.0.0', 'numpy==1.24.0']

overrides = {
    'yaml': {
        'package_name': 'PyYAML',
        'specifiers': [{'version': '6.0', 'op': '=='}]
    }
}
generate_requirements(['yaml'], overrides)
# output: ['PyYAML==6.0']

Parameters:

Name	Type	Description	Default
module_names	List[str]	List of Python module names to process	required
dependency_overrides	Dict[str, Dependency] | None	Dependency overrides for specific modules	None

Returns:

Type	Description
List[str]	List of requirement strings (e.g., ["package==1.0.0"])

Raises:

Type	Description
ValueError	If specified package name or versions in overrides are invalid or not found
LookupError	If a package cannot be located on PyPI
ConnectionError	If network-related issues occur when fetching package information

Source code in orchestr8/execution_runtime/package_utils.py

def generate_requirements(
    module_names: List[str], dependency_overrides: Dict[str, Dependency] | None = None
) -> List[str]:
    """
    Generate a list of Python package requirements from module names.

    For extracting module names from a Python script, use:

    ```python
    from orchestr8.runtime.utils import extract_module_names
    module_names = extract_module_names("/path/to/script.py")
    ```

    The function follows a systematic approach to resolving package dependencies:

    First, standard library modules are automatically excluded from the processing.
    The remaining modules are then searched among globally installed packages.
    When a module is found in installed packages, its package name and version are used.

    If a module is not installed locally, the function searches the PyPI package index.
    When a package is found on PyPI, its latest version is utilized.
    If no package can be found, the function raises an error.

    The function provides flexibility through the `dependency_overrides` parameter.
    This is particularly useful in scenarios where:
    - The package name differs from the module name
    - Custom version specifications are required

    A prime example is the `yaml` module, which corresponds to the `PyYAML` package.
    Through dependency overrides, you can specify custom package names or version constraints.

    The function performs comprehensive validation:
    - Checks if specified versions exist in package releases
    - Validates version specifiers against available package versions
    - Ensures precise and correct dependency specification

    ```python
    generate_requirements(['pandas', 'numpy'])
    # output: ['pandas==2.0.0', 'numpy==1.24.0']

    overrides = {
        'yaml': {
            'package_name': 'PyYAML',
            'specifiers': [{'version': '6.0', 'op': '=='}]
        }
    }
    generate_requirements(['yaml'], overrides)
    # output: ['PyYAML==6.0']
    ```

    Args:
        module_names: List of Python module names to process
        dependency_overrides: Dependency overrides for specific modules

    Returns:
        List of requirement strings (e.g., ["package==1.0.0"])

    Raises:
        ValueError: If specified package name or versions in overrides are invalid or not found
        LookupError: If a package cannot be located on PyPI
        ConnectionError: If network-related issues occur when fetching package information
    """
    requirements = []
    installables = set(module_names) - get_stdlib_modules()
    for mod_name, dep in (dependency_overrides or {}).items():
        if mod_name in installables:
            installables.remove(mod_name)
        pkg_name = dep.get("package_name") or mod_name
        release_info = get_package_release_info(pkg_name)
        if specifiers := dep.get("specifiers"):
            spec_versions = {spec["version"] for spec in specifiers}
            specifier_set = SpecifierSet(",".join(f"{spec['op']}{spec['version']}" for spec in specifiers))
            available_versions = release_info["ids"]
            not_found = list(spec_versions.difference(available_versions))
            if not_found:
                raise ValueError(f"Versions {not_found!r} mentioned in specifiers not found in {pkg_name!r} releases.")

            if not any(Version(v) in specifier_set for v in available_versions):
                raise ValueError(
                    f"Specifiers specified for package {pkg_name!r} are invalid. "
                    f"Avaiable versions: {available_versions!r}"
                )
            requirements.append(f"{pkg_name}{specifier_set!s}")
        else:
            requirements.append(f"{pkg_name}=={release_info['latest']}")

    mod_mapped_pkgs = get_module_mapped_packages()
    for mod_name in installables:
        if local_pkgs := mod_mapped_pkgs.get(mod_name):
            # Some modules have more than one package, so we'll include all available ones.
            for pkg_name, version in local_pkgs.items():
                release_info = get_package_release_info(pkg_name)
                if version in release_info["ids"]:
                    requirements.append(f"{pkg_name}=={version}")
        else:
            # We use the module name as the package name
            release_info = get_package_release_info(mod_name)
            requirements.append(f"{mod_name}=={release_info['latest']}")

    return requirements

get_module_mapped_packages

Map Python module names to their corresponding package information.

Creates a mapping of module names to dictionaries containing package names and their versions based on installed distributions.

Returns:

Type	Description
Dict[str, Dict[str, str]]	Dictionary mapping module names to {package_name: version} dictionaries

Source code in orchestr8/execution_runtime/package_utils.py

def get_module_mapped_packages() -> Dict[str, Dict[str, str]]:
    """
    Map Python module names to their corresponding package information.

    Creates a mapping of module names to dictionaries containing package names
    and their versions based on installed distributions.

    Returns:
        Dictionary mapping module names to {package_name: version} dictionaries
    """
    module_to_packages = defaultdict(dict[str, str])  # type: ignore[var-annotated]
    for dist in distributions():
        try:
            if not (
                pkg_name := getattr(
                    dist,
                    "name",
                    dist.metadata.get("Name", ""),  # type: ignore[attr-defined]
                )
            ):
                continue

            top_level_text = dist.read_text("top_level.txt")
            if top_level_text:
                for mod_name in top_level_text.split():
                    module_to_packages[mod_name][pkg_name] = (
                        getattr(dist, "version", None) or dist.metadata.get("Version", "") or ""  # type: ignore[attr-defined]
                    )
        except Exception:  # noqa: S112
            # Silently skip distributions that cause issues
            continue

    return dict(module_to_packages)

get_package_release_info

Fetch package release information from PyPI or custom package index.

Parameters:

Name	Type	Description	Default
__name	str	Name of the package to look up	required
resolver_url	str | None	Custom package index URL. Defaults to PyPI	None
urllib_request_kwargs	Any	Additional keyword arguments for urllib.request.Request	{}

Returns:

Type	Description
ReleaseInfo	Dictionary containing 'ids' (list of versions) and 'latest' (latest version)

Raises:

Type	Description
LookupError	If the package is not found
ConnectionError	If there are network issues or invalid responses

Source code in orchestr8/execution_runtime/package_utils.py

def get_package_release_info(__name: str, resolver_url: str | None = None, **urllib_request_kwargs: Any) -> ReleaseInfo:
    """
    Fetch package release information from PyPI or custom package index.

    Args:
        __name: Name of the package to look up
        resolver_url: Custom package index URL. Defaults to PyPI
        urllib_request_kwargs: Additional keyword arguments for urllib.request.Request

    Returns:
        Dictionary containing 'ids' (list of versions) and 'latest' (latest version)

    Raises:
        LookupError: If the package is not found
        ConnectionError: If there are network issues or invalid responses
    """
    from urllib.error import HTTPError, URLError
    from urllib.request import Request, urlopen

    base_url = (resolver_url or "https://pypi.python.org/pypi/").rstrip("/")
    url = f"{base_url}/{__name}/json"
    try:
        with urlopen(Request(url, **urllib_request_kwargs)) as response:  # noqa: S310
            if response.status != 200:
                raise ConnectionError(
                    f"An error occurred requesting {base_url!r} for package {__name!r}. "
                    f"Status code: {response.status}"
                )

            response_data = response.read().decode("utf-8")
            json_data: Dict[str, dict] = json.loads(response_data)

            return {
                "ids": list(json_data.get("releases", {}).keys()),
                "latest": json_data.get("info", {}).get("version"),  # type: ignore[typeddict-item]
            }

    except HTTPError as e:
        if e.code == 404:
            raise LookupError(f"Package {__name!r} not found.") from e
        raise ConnectionError(
            f"An error occurred requesting {base_url!r} for package {__name!r}. "
            f"Status code: {e.code}, Response: {e.read().decode('utf-8')}"
        ) from e

    except URLError as e:
        raise ConnectionError(f"Failed to connect to {base_url!r} for package {__name!r}. " f"Error: {e!s}") from e

get_stdlib_modules

Get a set of all Python standard library module names.

Returns:

Type	Description
FrozenSet[str]	Set of standard library module names read from stdlib.txt

Source code in orchestr8/execution_runtime/package_utils.py

def get_stdlib_modules() -> FrozenSet[str]:
    """
    Get a set of all Python standard library module names.

    Returns:
        Set of standard library module names read from stdlib.txt
    """
    return frozenset(line.strip() for line in (Path(__file__).parent / "stdlib.txt").open().readlines())