feat: refactor code outside of endpoints for reusability

Dockerfile

@@ -11,4 +11,6 @@ RUN pip install "."
 
 ENTRYPOINT [ "python", "-m", "eidos" ]
 
+EXPOSE 6004
+
 CMD [ "server" ]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "eidos"
-version = "0.99.0"
+version = "0.99.1"
 authors = [
   { name="José F. Aldana Martín", email="jfaldanam@uma.es" },
 ]

src/eidos/api/routes/execution.py

@@ -4,9 +4,8 @@
 from fastapi.responses import JSONResponse
 
 from eidos.api.secure import get_api_key
-from eidos.execution import get_eidos_function_definition, import_function
+from eidos.execution import execute
 from eidos.logs import get_logger
-from eidos.validation.schema import validate_input_schema, validate_output_schema
 
 logger = get_logger("eidos.api.execute")
 
@@ -19,7 +18,7 @@
     tags=["execution"],
     response_model=dict[str, Any],
 )
-async def execute(
+async def execute_endpoint(
     function_name: str, arguments: dict, api_key: str = Security(get_api_key)
 ) -> dict[str, Any]:
     """
@@ -33,65 +32,5 @@ async def execute(
     Returns:
         Result of the function execution.
     """
-    function_definition = get_eidos_function_definition(function_name)
-
-    # Validate inputs
-    try:
-        validate_input_schema(arguments, schema=function_definition["parameters"])
-    except (ValueError, TypeError) as e:
-        logger.error(f"Invalid input: {e}")
-        status = 400
-        response = {
-            "status": {
-                "code": status,
-                "message": f"Error: malformed function call.\n{str(e)}",
-            },
-            "data": None,
-        }
-
-        return JSONResponse(response, status_code=status)
-
-    # Execute function
-    try:
-        result = import_function(function_definition["module"])(**arguments)
-    except Exception as e:
-        logger.error(f"Error executing function {function_name}: {e}")
-        status = 500
-        response = {
-            "status": {
-                "code": status,
-                "message": f"Error: function execution failed.\n{str(e)}",
-            },
-            "data": None,
-        }
-
-        return JSONResponse(response, status_code=status)
-
-    # Validate and transform result
-    try:
-        validated_result = validate_output_schema(
-            result, schema=function_definition["response"].copy()
-        )
-    except (ValueError, TypeError) as e:
-        status = 500
-        response = {
-            "status": {
-                "code": status,
-                "message": f"Error: function return malformed results.\n{str(e)}",
-            },
-            "data": None,
-        }
-
-        return JSONResponse(response, status_code=status)
-
-    # Return validated results
-    status = 200
-    response = {
-        "status": {
-            "code": status,
-            "message": "Success",
-        },
-        "data": validated_result,
-    }
-
+    response, status = execute(function_name, arguments)
     return JSONResponse(response, status_code=status)

src/eidos/api/routes/functions.py

@@ -2,9 +2,10 @@
 
 from eidos.api.secure import get_api_key
 from eidos.execution import (
-    available_functions,
-    get_eidos_function_definition,
+    get_function_schema,
     get_openai_function_definition,
+    list_functions_names,
+    list_functions_openai,
 )
 from eidos.logs import get_logger
 
@@ -19,7 +20,7 @@
     tags=["functions"],
     response_model=list[dict],
 )
-async def list_functions(api_key: str = Security(get_api_key)) -> list[dict]:
+async def list_functions_endpoint(api_key: str = Security(get_api_key)) -> list[dict]:
     """
     List all available AI functions.
     \f
@@ -29,10 +30,7 @@ async def list_functions(api_key: str = Security(get_api_key)) -> list[dict]:
     Returns:
         List of available AI functions.
     """
-    return [
-        get_openai_function_definition(function_["name"])
-        for function_ in available_functions()
-    ]
+    return list_functions_openai()
 
 
 @router.get(
@@ -41,7 +39,9 @@ async def list_functions(api_key: str = Security(get_api_key)) -> list[dict]:
     tags=["functions"],
     response_model=list[str],
 )
-async def list_functions_names(api_key: str = Security(get_api_key)) -> list[str]:
+async def list_functions_names_endpoint(
+    api_key: str = Security(get_api_key),
+) -> list[str]:
     """
     List the names of all available AI functions.
     \f
@@ -51,7 +51,7 @@ async def list_functions_names(api_key: str = Security(get_api_key)) -> list[str
     Returns:
         List of names of available AI functions.
     """
-    return [function_["name"] for function_ in available_functions()]
+    return list_functions_names()
 
 
 @router.get(
@@ -72,9 +72,8 @@ async def function_definition(
     Returns:
         dict: Definition of the function.
     """
-    function_json = get_openai_function_definition(function)
 
-    return function_json
+    return get_openai_function_definition(function)
 
 
 @router.get(
@@ -93,6 +92,4 @@ async def function_schema(function: str, api_key: str = Security(get_api_key)) -
     Returns:
         dict: Response schema of the function.
     """
-    function_json = get_eidos_function_definition(function)
-
-    return function_json["response"]
+    return get_function_schema(function)

src/eidos/execution.py

@@ -1,4 +1,3 @@
-import importlib
 import json
 from functools import lru_cache
 from pathlib import Path
@@ -7,6 +6,8 @@
 from eidos.logs import get_logger
 from eidos.models.function import load_model
 from eidos.settings import config
+from eidos.utils import import_function
+from eidos.validation.schema import validate_input_schema, validate_output_schema
 
 logger = get_logger()
 
@@ -69,34 +70,111 @@ def available_functions() -> list[dict[str, Any]]:
     ]
 
 
-def import_function(module: str) -> callable:
-    """Import a function from a module.
+def get_function_schema(function: str) -> dict[str, Any]:
+    """Get the response schema of a function.
 
     Args:
-        module (str): The module name, e.g., "pprint.pprint"
+        function (str): Name of the function.
 
     Returns:
-        callable: The function object
+        dict: Response schema of the function.
     """
+    function_json = get_eidos_function_definition(function)
 
-    if "." not in module:
-        raise ValueError("You can't import built-in modules")
+    return function_json["response"]
 
+
+def list_functions_openai() -> list[dict[str, Any]]:
+    """List all available AI functions.
+
+    Returns:
+        List of available AI functions.
+    """
+    return [
+        get_openai_function_definition(function_["name"])
+        for function_ in available_functions()
+    ]
+
+
+def list_functions_names() -> list[str]:
+    """List the names of all available AI functions.
+
+    Returns:
+        List of names of available AI functions.
+    """
+    return [function_["name"] for function_ in available_functions()]
+
+
+def execute(function_name: str, arguments: dict) -> tuple[dict[str, Any], int]:
+    """
+    Executes an AI function.
+
+    Args:
+        function_name: Name of the function to execute.
+        arguments: Arguments to pass to the function.
+
+    Returns:
+        tuple[dict[str, Any], int]: Result of the function execution and a status code.
+    """
+    function_definition = get_eidos_function_definition(function_name)
+
+    # Validate inputs
+    try:
+        validate_input_schema(arguments, schema=function_definition["parameters"])
+    except (ValueError, TypeError) as e:
+        logger.error(f"Invalid input: {e}")
+        status = 400
+        response = {
+            "status": {
+                "code": status,
+                "message": f"Error: malformed function call.\n{str(e)}",
+            },
+            "data": None,
+        }
+
+        return response, status
+
+    # Execute function
     try:
-        module_name, function_name = module.rsplit(".", 1)
-        module = importlib.import_module(module_name)
-        try:
-            function_ = getattr(module, function_name)
-            return function_
-        except AttributeError as err:
-            logger.error(
-                f"Error: Function '{function_name}' "
-                f"not found in module '{module_name}'."
-            )
-            raise err
-    except (ValueError, ImportError) as err:
-        logger.error(
-            "Error: Unable to import module or "
-            f"function from the provided name: '{module}'"
+        result = import_function(function_definition["module"])(**arguments)
+    except Exception as e:
+        logger.error(f"Error executing function {function_name}: {e}")
+        status = 500
+        response = {
+            "status": {
+                "code": status,
+                "message": f"Error: function execution failed.\n{str(e)}",
+            },
+            "data": None,
+        }
+
+        return response, status
+
+    # Validate and transform result
+    try:
+        validated_result = validate_output_schema(
+            result, schema=function_definition["response"].copy()
         )
-        raise err
+    except (ValueError, TypeError) as e:
+        status = 500
+        response = {
+            "status": {
+                "code": status,
+                "message": f"Error: function return malformed results.\n{str(e)}",
+            },
+            "data": None,
+        }
+
+        return response, status
+
+    # Return validated results
+    status = 200
+    response = {
+        "status": {
+            "code": status,
+            "message": "Success",
+        },
+        "data": validated_result,
+    }
+
+    return response, status

src/eidos/utils.py

@@ -0,0 +1,38 @@
+import importlib
+
+from eidos.logs import get_logger
+
+logger = get_logger()
+
+
+def import_function(module: str) -> callable:
+    """Import a function from a module.
+
+    Args:
+        module (str): The module name, e.g., "pprint.pprint"
+
+    Returns:
+        callable: The function object
+    """
+
+    if "." not in module:
+        raise ValueError("You can't import built-in modules")
+
+    try:
+        module_name, function_name = module.rsplit(".", 1)
+        module = importlib.import_module(module_name)
+        try:
+            function_ = getattr(module, function_name)
+            return function_
+        except AttributeError as err:
+            logger.error(
+                f"Error: Function '{function_name}' "
+                f"not found in module '{module_name}'."
+            )
+            raise err
+    except (ValueError, ImportError) as err:
+        logger.error(
+            "Error: Unable to import module or "
+            f"function from the provided name: '{module}'"
+        )
+        raise err