Change name from LocalCommandlineCodeExecutor to LocalCommandLineCodeExecutor

autogen/coding/factory.py

@@ -34,8 +34,8 @@ def create(code_execution_config: Dict[str, Any]) -> CodeExecutor:
 
             return EmbeddedIPythonCodeExecutor(**code_execution_config.get("ipython-embedded", {}))
         elif executor == "commandline-local":
-            from .local_commandline_code_executor import LocalCommandlineCodeExecutor
+            from .local_commandline_code_executor import LocalCommandLineCodeExecutor
 
-            return LocalCommandlineCodeExecutor(**code_execution_config.get("commandline-local", {}))
+            return LocalCommandLineCodeExecutor(**code_execution_config.get("commandline-local", {}))
         else:
             raise ValueError(f"Unknown code executor {executor}")

autogen/coding/local_commandline_code_executor.py

@@ -11,12 +11,12 @@
 from .markdown_code_extractor import MarkdownCodeExtractor
 
 __all__ = (
-    "LocalCommandlineCodeExecutor",
-    "CommandlineCodeResult",
+    "LocalCommandLineCodeExecutor",
+    "CommandLineCodeResult",
 )
 
 
-class CommandlineCodeResult(CodeResult):
+class CommandLineCodeResult(CodeResult):
     """(Experimental) A code result class for command line code executor."""
 
     code_file: Optional[str] = Field(
@@ -25,7 +25,7 @@ class CommandlineCodeResult(CodeResult):
     )
 
 
-class LocalCommandlineCodeExecutor(BaseModel):
+class LocalCommandLineCodeExecutor(BaseModel):
     """(Experimental) A code executor class that executes code through a local command line
     environment.
 
@@ -49,7 +49,7 @@ class LocalCommandlineCodeExecutor(BaseModel):
             directory is the current directory ".".
         system_message_update (str): The system message update for agent that
             produces code to run on this executor.
-            Default is `LocalCommandlineCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.
+            Default is `LocalCommandLineCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.
     """
 
     DEFAULT_SYSTEM_MESSAGE_UPDATE: ClassVar[
@@ -92,10 +92,10 @@ def _check_work_dir(cls, v: str) -> str:
         raise ValueError(f"Working directory {v} does not exist.")
 
     @property
-    def user_capability(self) -> "LocalCommandlineCodeExecutor.UserCapability":
+    def user_capability(self) -> "LocalCommandLineCodeExecutor.UserCapability":
         """Export a user capability for this executor that can be added to
         an agent that produces code to be executed by this executor."""
-        return LocalCommandlineCodeExecutor.UserCapability(self.system_message_update)
+        return LocalCommandLineCodeExecutor.UserCapability(self.system_message_update)
 
     @property
     def code_extractor(self) -> CodeExtractor:
@@ -124,19 +124,19 @@ def sanitize_command(lang: str, code: str) -> None:
                 if re.search(pattern, code):
                     raise ValueError(f"Potentially dangerous command detected: {message}")
 
-    def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandlineCodeResult:
+    def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandLineCodeResult:
         """(Experimental) Execute the code blocks and return the result.
 
         Args:
             code_blocks (List[CodeBlock]): The code blocks to execute.
 
         Returns:
-            CommandlineCodeResult: The result of the code execution."""
+            CommandLineCodeResult: The result of the code execution."""
         logs_all = ""
         for i, code_block in enumerate(code_blocks):
             lang, code = code_block.language, code_block.code
 
-            LocalCommandlineCodeExecutor.sanitize_command(lang, code)
+            LocalCommandLineCodeExecutor.sanitize_command(lang, code)
             filename_uuid = uuid.uuid4().hex
             filename = None
             if lang in ["bash", "shell", "sh", "pwsh", "powershell", "ps1"]:
@@ -166,8 +166,75 @@ def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandlineCodeRe
             if exitcode != 0:
                 break
         code_filename = os.path.join(self.work_dir, filename) if filename is not None else None
-        return CommandlineCodeResult(exit_code=exitcode, output=logs_all, code_file=code_filename)
+        return CommandLineCodeResult(exit_code=exitcode, output=logs_all, code_file=code_filename)
 
     def restart(self) -> None:
         """(Experimental) Restart the code executor."""
         warnings.warn("Restarting local command line code executor is not supported. No action is taken.")
+
+
+# From stack overflow: https://stackoverflow.com/a/52087847/2214524
+class _DeprecatedClassMeta(type):
+    def __new__(cls, name, bases, classdict, *args, **kwargs):
+        alias = classdict.get("_DeprecatedClassMeta__alias")
+
+        if alias is not None:
+
+            def new(cls, *args, **kwargs):
+                alias = getattr(cls, "_DeprecatedClassMeta__alias")
+
+                if alias is not None:
+                    warnings.warn(
+                        "{} has been renamed to {}, the alias will be "
+                        "removed in the future".format(cls.__name__, alias.__name__),
+                        DeprecationWarning,
+                        stacklevel=2,
+                    )
+
+                return alias(*args, **kwargs)
+
+            classdict["__new__"] = new
+            classdict["_DeprecatedClassMeta__alias"] = alias
+
+        fixed_bases = []
+
+        for b in bases:
+            alias = getattr(b, "_DeprecatedClassMeta__alias", None)
+
+            if alias is not None:
+                warnings.warn(
+                    "{} has been renamed to {}, the alias will be "
+                    "removed in the future".format(b.__name__, alias.__name__),
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
+
+            # Avoid duplicate base classes.
+            b = alias or b
+            if b not in fixed_bases:
+                fixed_bases.append(b)
+
+        fixed_bases = tuple(fixed_bases)
+
+        return super().__new__(cls, name, fixed_bases, classdict, *args, **kwargs)
+
+    def __instancecheck__(cls, instance):
+        return any(cls.__subclasscheck__(c) for c in {type(instance), instance.__class__})
+
+    def __subclasscheck__(cls, subclass):
+        if subclass is cls:
+            return True
+        else:
+            return issubclass(subclass, getattr(cls, "_DeprecatedClassMeta__alias"))
+
+
+class LocalCommandlineCodeExecutor(metaclass=_DeprecatedClassMeta):
+    """LocalCommandlineCodeExecutor renamed to LocalCommandLineCodeExecutor"""
+
+    _DeprecatedClassMeta__alias = LocalCommandLineCodeExecutor
+
+
+class CommandlineCodeResult(metaclass=_DeprecatedClassMeta):
+    """CommandlineCodeResult renamed to CommandLineCodeResult"""
+
+    _DeprecatedClassMeta__alias = CommandLineCodeResult

test/coding/test_commandline_code_executor.py

@@ -4,7 +4,7 @@
 from autogen.agentchat.conversable_agent import ConversableAgent
 from autogen.coding.base import CodeBlock, CodeExecutor
 from autogen.coding.factory import CodeExecutorFactory
-from autogen.coding.local_commandline_code_executor import LocalCommandlineCodeExecutor
+from autogen.coding.local_commandline_code_executor import LocalCommandLineCodeExecutor
 from autogen.oai.openai_utils import config_list_from_json
 
 from conftest import MOCK_OPEN_AI_API_KEY, skip_openai
@@ -13,25 +13,25 @@
 def test_create() -> None:
     config = {"executor": "commandline-local"}
     executor = CodeExecutorFactory.create(config)
-    assert isinstance(executor, LocalCommandlineCodeExecutor)
+    assert isinstance(executor, LocalCommandLineCodeExecutor)
 
-    config = {"executor": LocalCommandlineCodeExecutor()}
+    config = {"executor": LocalCommandLineCodeExecutor()}
     executor = CodeExecutorFactory.create(config)
     assert executor is config["executor"]
 
 
 def test_local_commandline_executor_init() -> None:
-    executor = LocalCommandlineCodeExecutor(timeout=10, work_dir=".")
+    executor = LocalCommandLineCodeExecutor(timeout=10, work_dir=".")
     assert executor.timeout == 10 and executor.work_dir == "."
 
     # Try invalid working directory.
     with pytest.raises(ValueError, match="Working directory .* does not exist."):
-        executor = LocalCommandlineCodeExecutor(timeout=111, work_dir="/invalid/directory")
+        executor = LocalCommandLineCodeExecutor(timeout=111, work_dir="/invalid/directory")
 
 
 def test_local_commandline_executor_execute_code() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
+        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
         _test_execute_code(executor=executor)
 
 
@@ -81,7 +81,7 @@ def _test_execute_code(executor: CodeExecutor) -> None:
 @pytest.mark.skipif(sys.platform in ["win32"], reason="do not run on windows")
 def test_local_commandline_code_executor_timeout() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        executor = LocalCommandlineCodeExecutor(timeout=1, work_dir=temp_dir)
+        executor = LocalCommandLineCodeExecutor(timeout=1, work_dir=temp_dir)
         _test_timeout(executor)
 
 
@@ -92,7 +92,7 @@ def _test_timeout(executor: CodeExecutor) -> None:
 
 
 def test_local_commandline_code_executor_restart() -> None:
-    executor = LocalCommandlineCodeExecutor()
+    executor = LocalCommandLineCodeExecutor()
     _test_restart(executor)
 
 
@@ -105,7 +105,7 @@ def _test_restart(executor: CodeExecutor) -> None:
 @pytest.mark.skipif(skip_openai, reason="requested to skip openai tests")
 def test_local_commandline_executor_conversable_agent_capability() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
+        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
         _test_conversable_agent_capability(executor=executor)
 
 
@@ -150,7 +150,7 @@ def _test_conversable_agent_capability(executor: CodeExecutor) -> None:
 
 def test_local_commandline_executor_conversable_agent_code_execution() -> None:
     with tempfile.TemporaryDirectory() as temp_dir:
-        executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
+        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
         with pytest.MonkeyPatch.context() as mp:
             mp.setenv("OPENAI_API_KEY", MOCK_OPEN_AI_API_KEY)
             _test_conversable_agent_code_execution(executor)
@@ -192,7 +192,7 @@ def _test_conversable_agent_code_execution(executor: CodeExecutor) -> None:
 )
 def test_dangerous_commands(lang, code, expected_message):
     with pytest.raises(ValueError) as exc_info:
-        LocalCommandlineCodeExecutor.sanitize_command(lang, code)
+        LocalCommandLineCodeExecutor.sanitize_command(lang, code)
     assert expected_message in str(
         exc_info.value
     ), f"Expected message '{expected_message}' not found in '{str(exc_info.value)}'"