Add functionality for retrieving callable definitions and converting them to callables (unitycatalog#912)

**PR Checklist**

- [X] A description of the changes is added to the description of this
PR.
- [ ] If there is a related issue, make sure it is linked to this PR.
- [ ] If you've fixed a bug or added code that should be tested, add
tests!
- [ ] If you've added or modified a feature, documentation in `docs` is
updated

**Description of changes**

Extension of unitycatalog#911 , bringing a unified functionality across both
clients. Adds a utility for converting the string representation to an
actual callable for local execution.

---------

Signed-off-by: Ben Wilson <[email protected]>

Author: BenWilson2

ai/core/README.md

@@ -298,6 +298,65 @@ full_func_name = f"{CATALOG}.{SCHEMA}.{func_name}"
 client.get_function(full_func_name)
 ```
 
+#### Retrieving a UC function callable
+
+The `get_function_source` API is used to retrieve a recreated python callable definition (as a string) from a registered Unity Catalog Python function.
+In order to use this API, the function that you are fetching **must be** an `EXTERNAL` (python function) type function. When called, the function's metadata will
+be retrieved and the structure of the original callable will be rebuilt and returned as a string.
+
+For example:
+
+```python
+# Define a python callable
+
+def sample_python_func(a: int, b: int) -> int:
+    """
+    Returns the sum of a and b.
+
+    Args:
+        a: an int
+        b: another int
+
+    Returns:
+        The sum of a and b
+    """
+    return a + b
+
+# Create the function within Unity Catalog
+client.create_python_function(catalog=CATALOG, schema=SCHEMA, func=sample_python_func, replace=True)
+
+# Fetch the callable definition
+my_func_def = client.get_function_source(function_name=f"{CATALOG}.{SCHEMA}.sample_python_function")
+```
+
+The returned value from the `get_function_source` API will be the same as the original input with a few caveats:
+
+- `tuple` types will be cast to `list` due to the inability to express a Python `tuple` within Unity Catalog
+- The docstring of the original function will be stripped out. Unity Catalog persists the docstring information in the logged function and it is available in the return of the `get_function` API call if needed.
+- Collection types for open source Unity Catalog will only capture the outer type (i.e., `list` or `dict`) as inner collection type metadata is not preserved
+within the `FunctionInfo` object. In Databricks, full typing is supported for collecitons.
+
+The result of calling the `get_function_source` API on the `sample_python_func` registered function will be (when printed):
+
+```text
+def sample_python_func(a: int, b: int) -> int:
+    """
+    Returns the sum of a and b.
+
+    Args:
+        a: an int
+        b: another int
+
+    Returns:
+        int
+    """
+    return a + b
+```
+
+Note: If you want to convert the extracted string back into an actual Python callable, you can use the utility `load_function_from_string` in the module `unitycatalog.ai.core.utils.execution_utils`. See below for further details on this API.
+
+This API is useful for extracting already-registered functions that will be used as additional in-line calls within another function through the use of the `create_wrapped_python_function` API, saving the effort required to either hand-craft a function definition or having to track down where the original implementation of a logged function was defined.
+
 #### List UC functions
 
 To get a list of functions stored in a catalog and schema, you can use list API with wildcards to do so.
@@ -315,6 +374,79 @@ result = client.execute_function(full_func_name, {"s": "some_string"})
 assert result.value == "some_string"
 ```
 
+#### Execute a UC Python function locally
+
+A utility `load_function_from_string` is available in `unitycatalog.ai.core.utils.execution_utils.py`. This utility allows you to couple the functionality
+in the `get_function_source` API to create a locally-available python callable that can be direclty accessed, precisely as if it were originally defined
+within your current REPL.
+
+```python
+from unitycatalog.ai.core.utils.execution_utils import load_function_from_string
+
+func_str = """
+def multiply_numbers(a: int, b: int) -> int:
+    \"\"\"
+    Multiplies two numbers.
+
+    Args:
+        a: first number.
+        b: second number.
+
+    Returns:
+        int
+    \"\"\"
+    return a * b
+"""
+
+# If specifying `register_global=False`, the original function name cannot be called and must be used
+# with the returned callable reference.
+my_new_multiplier = load_function_from_string(func_str, register_global=False)
+my_new_multiplier(a=1, b=2)  # returns `2`
+
+# Alternatively, if allowing for global reference `register_global=True` (default)
+# The original callable name can be used. This will not work in interactive environments like Jupyter.
+load_function_from_string(func_str)
+multiply_numbers(a=2, b=2)  # returns `4`
+
+# For interactive environments, setting the return object directly within globals() is required in order
+# to utilize the original function name
+alias = load_function_from_string(func_str)
+globals()["multiply_numbers"] = alias
+multiply_numbers(a=3, b=3)  # returns `9`
+
+# Additionally, a scoped namespace can be provided to restrict scope and access to scoped arguments
+from types import SimpleNamespace
+
+func_str2 = """
+def multiply_numbers_with_constant(a: int, b: int) -> int:
+    \"\"\"
+    Multiplies two numbers with a constant.
+
+    Args:
+        a: first number.
+        b: second number.
+    
+    Returns:
+        int
+    \"\"\"
+    return a * b * c
+"""
+
+c = 100  # Not part of the scoped namespace; local constant
+
+scoped_namespace = {
+    "__builtins__": __builtins__,
+    "c": 42,
+}
+    
+load_function_from_string(func_str, register_function=True, namespace=scoped_namespace)
+
+scoped_ns = SimpleNamespace(**scoped_namespace)
+
+scoped_ns.multiply_numbers_with_constant(a=2, b=3)  # returns 252, utilizing the `c` constant of the namespace
+
+```
+
 ##### Function execution arguments configuration
 
 To manage the function execution behavior using Databricks client under different configurations, we offer the following environment variables:

ai/core/src/unitycatalog/ai/core/base.py

@@ -222,6 +222,20 @@ def to_dict(self):
         Sensitive information should be excluded.
         """
 
+    @abstractmethod
+    def get_function_source(self, function_name: str) -> str:
+        """
+        Get the Python callable definition reconstructed from Unity Catalog
+          for a function by its name. The return of this method is a string
+          that contains the callable's definition.
+
+        Args:
+            function_name: The name of the function to retrieve from Unity Catalog.
+
+        Returns:
+            str: The Python callable definition as a string.
+        """
+
 
 # TODO: update BaseFunctionClient to Union[BaseFunctionClient, AsyncBaseFunctionClient] after async client is supported
 def get_uc_function_client() -> Optional[BaseFunctionClient]:

ai/core/src/unitycatalog/ai/core/client.py

@@ -12,6 +12,9 @@
 
 from unitycatalog.ai.core.base import BaseFunctionClient, FunctionExecutionResult
 from unitycatalog.ai.core.paged_list import PagedList
+from unitycatalog.ai.core.utils.callable_utils import (
+    dynamically_construct_python_function,
+)
 from unitycatalog.ai.core.utils.callable_utils_oss import (
     generate_function_info,
     generate_wrapped_function_info,
@@ -845,7 +848,7 @@ def _execute_uc_function(
             except Exception as e:
                 return FunctionExecutionResult(error=str(e))
         else:
-            python_function = dynamically_construct_python_function(function_info)
+            python_function = _get_callable_definition(function_info)
             exec(python_function, self.func_cache)
             try:
                 func = self.func_cache[function_info.name]
@@ -907,30 +910,28 @@ def to_dict(self) -> Dict[str, Any]:
         elements = ["uc"]
         return {k: getattr(self, k) for k in elements if getattr(self, k) is not None}
 
+    @override
+    def get_function_source(self, function_name: str) -> str:
+        """
+        Returns the Python callable definition as a string for an EXTERNAL Python function that
+        is stored within Unity Catalog. This function can only parse and extract the full callable
+        definition for Python functions and cannot be used on SQL or TABLE functions.
 
-def dynamically_construct_python_function(function_info: FunctionInfo) -> str:
-    """
-    Construct a Python function from the given FunctionInfo.
-
-    Args:
-        function_info: The FunctionInfo object containing the function metadata.
-
-    Returns:
-        The re-constructed function definition.
-    """
+        NOTE: To unify the behavior of creating a valid Python callable, existing indentation in the
+        stored function body will be unified to a consistent indentation level of `4` spaces.
 
-    param_names = []
-    if function_info.input_params and function_info.input_params.parameters:
-        param_names = [param.name for param in function_info.input_params.parameters]
-    function_head = f"{function_info.name}({', '.join(param_names)})"
-    func_def = f"def {function_head}:\n"
-    if function_info.routine_body == "EXTERNAL":
-        for line in function_info.routine_definition.split("\n"):
-            func_def += f"    {line}\n"
-    else:
-        raise NotImplementedError(f"routine_body {function_info.routine_body} not supported")
+        Args:
+            function_name: The name of the function to retrieve the Python callable definition for.
 
-    return func_def
+        Returns:
+            str: The Python callable definition as a string.
+        """
+        function_info = self.get_function(function_name)
+        if function_info.routine_body != "EXTERNAL":
+            raise ValueError(
+                f"Function {function_name} is not an EXTERNAL Python function and cannot be retrieved."
+            )
+        return dynamically_construct_python_function(function_info=function_info)
 
 
 def validate_input_parameter(
@@ -1002,3 +1003,31 @@ def validate_param(param: Any, column_type: str, param_type_text: str) -> None:
             f"Invalid interval type text: {param_type_text}, expecting 'interval day to second', "
             "python timedelta can only be used for day-time interval."
         )
+
+
+def _get_callable_definition(function_info: FunctionInfo) -> str:
+    """
+    Construct a Python function from the given FunctionInfo without docstring, comments, or types.
+    This funciton is purely used for local function execution encapsulated within a call to
+    `execute_function` and is not intended to be used for retrieving a callable definition.
+    Use `get_python_callable` instead.
+
+    Args:
+        function_info: The FunctionInfo object containing the function metadata.
+
+    Returns:
+        The minimal re-constructed function definition.
+    """
+
+    param_names = []
+    if function_info.input_params and function_info.input_params.parameters:
+        param_names = [param.name for param in function_info.input_params.parameters]
+    function_head = f"{function_info.name}({', '.join(param_names)})"
+    func_def = f"def {function_head}:\n"
+    if function_info.routine_body == "EXTERNAL":
+        for line in function_info.routine_definition.split("\n"):
+            func_def += f"    {line}\n"
+    else:
+        raise NotImplementedError(f"routine_body {function_info.routine_body} not supported")
+
+    return func_def

ai/core/src/unitycatalog/ai/core/databricks.py

@@ -20,6 +20,7 @@
 from unitycatalog.ai.core.paged_list import PagedList
 from unitycatalog.ai.core.types import Variant
 from unitycatalog.ai.core.utils.callable_utils import (
+    dynamically_construct_python_function,
     generate_sql_function_body,
     generate_wrapped_sql_function_body,
 )
@@ -742,6 +743,27 @@ def from_dict(cls, config: Dict[str, Any]):
         accept_keys = ["profile"]
         return cls(**{k: v for k, v in config.items() if k in accept_keys})
 
+    @override
+    def get_function_source(self, function_name: str) -> str:
+        """
+        Returns the Python callable definition as a string for an EXTERNAL Python function that
+        is stored within Unity Catalog. This function can only parse and extract the full callable
+        definition for Python functions and cannot be used on SQL or TABLE functions.
+
+        Args:
+            function_name: The name of the function to retrieve the Python callable definition for.
+
+        Returns:
+            str: The Python callable definition as a string.
+        """
+
+        function_info = self.get_function(function_name)
+        if function_info.routine_body.value != "EXTERNAL":
+            raise ValueError(
+                f"Function {function_name} is not an EXTERNAL Python function and cannot be retrieved."
+            )
+        return dynamically_construct_python_function(function_info)
+
 
 def is_scalar(function: "FunctionInfo") -> bool:
     from databricks.sdk.service.catalog import ColumnTypeName