The run_internal function is an asynchronous function that runs a notebook program by importing it, finding a function to execute, mapping user inputs to function parameters, converting types, and executing the function, while the run_async function runs this process asynchronously using asyncio.gather.

Run example

npm run import -- "run python cells"

run python cells

import json
import os
import sys
import types
import asyncio
import inspect

async def run_internal():
    from Core import interpret, import_notebook, get_parameter_names
    if len(sys.argv) < 2:
        print("Usage: python script.py <notebook_path> <function_args>")
        sys.exit(1)

    notebook_path = sys.argv[1]
    inputs = sys.argv[2:]

    # Import the notebook as a module
    results = interpret(notebook_path)
    #print(results[0]['code'])

    module = import_notebook("notebook_path",
"globals("))
    print(module)

    # Find the first function in the module
    func = None
    if(callable(module)):
        func = module
    else:
        for name in list(module.keys()):
            # TODO: support classes ? attr = getattr(module, name)
            attr = module[name]
            if callable(attr):
                func = attr
                break

    if not func:
        print("No function found in the notebook.")
        sys.exit(1)
    print(func)

    # Extract parameters and map inputs
    params = []
    if(results[0]["language"] == 'python'):
        params = get_parameter_names(results[0]['code'])[1:]
        print(params)
    mapped_inputs = []

    for param in params:
        for i, arg in enumerate(inputs):
            if arg.startswith(f"--{param}="):
                mapped_inputs.append(arg.split("=")[1])
                break
        else:
            mapped_inputs.append(inputs.pop(0) if inputs else None)

    # Convert types based on function annotations (if available)
    if hasattr(func, '__annotations__'):
        func_annotations = func.__annotations__
        print(func_annotations)
        for i, param in enumerate(params):
            if param in func_annotations:
                mapped_inputs[i] = func_annotations[param](mapped_inputs[i])

    # Execute the function
    if inspect.iscoroutinefunction(func):
        return await func(*mapped_inputs)
    else:
        return func(*mapped_inputs)

async def run_async():
    return await asyncio.gather(run_internal())

def run():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(run_async())
    if os.getenv('RUN_OUTPUT') is 'json':
        result = json.dumps(result, indent=4)
    print(result)

if __name__ == "__run__":
    run()

__all__ = {
  "run": run
}

import json
import os
import sys
import types
import asyncio
import inspect
from Core import interpret, import_notebook, get_parameter_names
from concurrent.futures import ThreadPoolExecutor

def get_notebook_path():
    """Get the notebook path from the command line arguments."""
    if len(sys.argv) < 2:
        print("Usage: python script.py  ")
        sys.exit(1)
    return sys.argv[1]

def get_function_args():
    """Get the function arguments from the command line arguments."""
    if len(sys.argv) < 3:
        return []
    return sys.argv[2:]

def import_notebook(notebook_path):
    """Import the notebook as a module."""
    results = interpret(notebook_path)
    module = import_notebook(notebook_path, globals())
    return module

def find_function(module):
    """Find the first function in the module."""
    if callable(module):
        return module
    for name in module:
        attr = module[name]
        if callable(attr):
            return attr
    return None

def extract_parameters(module):
    """Extract parameters and map inputs."""
    params = []
    if hasattr(module, '__annotations__'):
        func_annotations = module.__annotations__
        params = list(func_annotations.keys())
    else:
        params = get_parameter_names(module.__code__)
    mapped_inputs = []
    inputs = get_function_args()
    for param in params:
        for i, arg in enumerate(inputs):
            if arg.startswith(f"--{param}="):
                mapped_inputs.append(arg.split("=")[1])
                break
        else:
            mapped_inputs.append(inputs.pop(0) if inputs else None)
    return params, mapped_inputs

def convert_types(module, params, mapped_inputs):
    """Convert types based on function annotations (if available)."""
    func_annotations = module.__annotations__ if hasattr(module, '__annotations__') else {}
    for i, param in enumerate(params):
        if param in func_annotations:
            mapped_inputs[i] = func_annotations[param](mapped_inputs[i])

def execute_function(module, params, mapped_inputs):
    """Execute the function."""
    if inspect.iscoroutinefunction(module):
        return module(*mapped_inputs)
    else:
        return module(*mapped_inputs)

async def run_internal():
    notebook_path = get_notebook_path()
    module = import_notebook(notebook_path)
    func = find_function(module)
    if not func:
        print("No function found in the notebook.")
        sys.exit(1)
    params, mapped_inputs = extract_parameters(func)
    convert_types(func, params, mapped_inputs)
    return await execute_function(func, params, mapped_inputs)

async def run_async():
    return await asyncio.gather(run_internal())

def run():
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(run_async())
    if os.getenv('RUN_OUTPUT') == 'json':
        result = json.dumps(result, indent=4)
    print(result)

if __name__ == "__main__":
    run()

Code Breakdown

Importing Modules

The code starts by importing various Python modules:

import json
import os
import sys
import types
import asyncio
import inspect

run_internal Function

This function is an asynchronous function (async def) that runs the notebook program and executes the specified function. Here's what it does:

1. Argument Handling:

   • It checks if the user provided enough arguments (at least 2) when running the script.
   • If not, it prints usage instructions and exits with an error code.

2. Importing the Notebook:

   • It uses the interpret function to import the notebook as a Python module.
   • It then imports the module using the import_notebook function and adds it to the global scope.

3. Finding the Function:

   • It identifies the first function in the imported module.
   • If no function is found, it prints an error message and exits.

4. Extracting Parameters and Mapping Inputs:

   • It extracts the function's parameters and maps the user-provided inputs to these parameters.
   • If a parameter is not specified with the expected format (--param=value), it uses the next available input.

5. Converting Types:

   • If the function has type annotations, it converts the mapped inputs to the expected types.

6. Executing the Function:

   • It executes the function using the mapped inputs.
   • If the function is a coroutine, it uses await to execute it asynchronously.

run_async Function

This function is also asynchronous and is used to run the run_internal function asynchronously using asyncio.gather.

async def run_async():
    return await asyncio.gather(run_internal())

run Function (Documentation Only)

This function is not implemented and is likely intended to be replaced with an asynchronous version of run_internal. The comment suggests that any friendly remarks should be discarded.

def run()
Documentation only, discard any friendly remarks.