Add shell command to Protean CLI (#402)

The `shell` command accepts a domain path as an argument.

A domain context is activated, and the domain instance is imported. The shell also preloads all domain elements to allow free-form domain exploration.

pyproject.toml

@@ -82,6 +82,7 @@ docutils = ">=0.20.1"
 pre-commit = ">=2.16.0"
 tox = ">=4.14.1"
 twine = ">=4.0.2"
+ipython = "^8.23.0"
 
 [tool.poetry.group.test]
 optional = true
@@ -101,7 +102,6 @@ pytest = ">=7.4.3"
 optional = true
 
 [tool.poetry.group.docs.dependencies]
-livereload = ">=2.6.3"
 sphinx = ">=7.2.6"
 sphinx-tabs = ">=3.4.4"
 mkdocs-material = "^9.5.15"

src/protean/cli/__init__.py

@@ -16,15 +16,19 @@
 """
 
 import subprocess
+import sys
+import typing
 
 from enum import Enum
 from typing import Optional
 
 import typer
 
+from IPython.terminal.embed import InteractiveShellEmbed
 from rich import print
 from typing_extensions import Annotated
 
+from protean.cli.docs import app as docs_app
 from protean.cli.generate import app as generate_app
 from protean.cli.new import new
 from protean.exceptions import NoDomainException
@@ -36,6 +40,7 @@
 
 app.command()(new)
 app.add_typer(generate_app, name="generate")
+app.add_typer(docs_app, name="docs")
 
 
 class Category(str, Enum):
@@ -119,17 +124,6 @@ def test(
             subprocess.call(commands)
 
 
-@app.command()
-def livereload_docs():
-    """Run in shell as `protean livereload-docs`"""
-    from livereload import Server, shell
-
-    server = Server()
-    server.watch("docs-sphinx/**/*.rst", shell("make html"))
-    server.watch("./*.rst", shell("make html"))
-    server.serve(root="build/html", debug=True)
-
-
 @app.command()
 def server(
     domain_path: Annotated[str, typer.Argument()] = "",
@@ -149,3 +143,40 @@ def server(
 
     engine = Engine(domain, test_mode=test_mode)
     engine.run()
+
+
+@app.command()
+def shell(domain_path: Annotated[str, typer.Argument()] = ""):
+    """Run an interactive Python shell in the context of a given
+    Protean domain.  The domain will populate the default
+    namespace of this shell according to its configuration.
+
+    This is useful for executing small snippets of code
+    without having to manually configure the application.
+
+    FIXME: Populate context in a decorator like Flask does:
+        https://github.com/pallets/flask/blob/b90a4f1f4a370e92054b9cc9db0efcb864f87ebe/src/flask/cli.py#L368
+        https://github.com/pallets/flask/blob/b90a4f1f4a370e92054b9cc9db0efcb864f87ebe/src/flask/cli.py#L984
+    """
+    domain = derive_domain(domain_path)
+    if not domain:
+        raise NoDomainException(
+            "Could not locate a Protean domain. You should provide a domain in"
+            '"PROTEAN_DOMAIN" environment variable or pass a domain file in options '
+            'and a "domain.py" module was not found in the current directory.'
+        )
+
+    with domain.domain_context():
+        domain.init()
+
+        ctx: dict[str, typing.Any] = {}
+        ctx.update(domain.make_shell_context())
+
+        banner = (
+            f"Python {sys.version} on {sys.platform}\n"
+            f"    location: {sys.executable}\n"
+            f"Domain: {domain.domain_name}\n"
+        )
+        ipshell = InteractiveShellEmbed(banner1=banner, user_ns=ctx)
+
+        ipshell()

src/protean/cli/docs.py

@@ -0,0 +1,26 @@
+import subprocess
+
+import typer
+
+app = typer.Typer(no_args_is_help=True)
+
+
+@app.callback()
+def callback():
+    """
+    If we want to create a CLI app with one single command but
+    still want it to be a command/subcommand, we need to add a callback.
+
+    This can be removed when we have more than one command/subcommand.
+
+    https://typer.tiangolo.com/tutorial/commands/one-or-multiple/#one-command-and-one-callback
+    """
+
+
+@app.command()
+def preview():
+    """Run a live preview server"""
+    try:
+        subprocess.call(["mkdocs", "serve"])
+    except KeyboardInterrupt:
+        pass

src/protean/domain/__init__.py

@@ -791,3 +791,22 @@ def get_email_provider(self, provider_name):
 
     def send_email(self, email):
         return self.email_providers.send_email(email)
+
+    def make_shell_context(self):
+        """Return a dictionary of context variables for a shell session."""
+        values = {"domain": self}
+
+        # For each domain element type in Domain Objects,
+        #   Cycle through all values in self.registry._elements[element_type]
+        #   and add each class to the shell context by the key
+        for element_type in DomainObjects:
+            values.update(
+                {
+                    v.name: v.cls
+                    for _, v in self._domain_registry._elements[
+                        element_type.value
+                    ].items()
+                }
+            )
+
+        return values

src/protean/utils/domain.py

@@ -1,4 +1,5 @@
 import ast
+import logging
 import os
 import re
 import sys
@@ -9,6 +10,8 @@
 from protean import Domain
 from protean.exceptions import NoDomainException
 
+logger = logging.getLogger(__name__)
+
 
 def find_domain_in_module(module: ModuleType) -> Domain:
     """Given a module instance, find an instance of Protean `Domain` class.
@@ -152,16 +155,16 @@ def locate_domain(module_name, domain_name, raise_if_not_found=True):
 
     try:
         __import__(module_name)
-    except ImportError:
+    except ImportError as exc:
         # Reraise the ImportError if it occurred within the imported module.
         # Determine this by checking whether the trace has a depth > 1.
         if sys.exc_info()[2].tb_next:
             raise NoDomainException(
                 f"While importing {module_name!r}, an ImportError was"
                 f" raised:\n\n{traceback.format_exc()}"
-            )
+            ) from exc
         elif raise_if_not_found:
-            raise NoDomainException(f"Could not import {module_name!r}.")
+            raise NoDomainException(f"Could not import {module_name!r}.") from exc
         else:
             return
 
@@ -189,7 +192,7 @@ def derive_domain(domain_path):
     domain_import_path = os.environ.get("PROTEAN_DOMAIN") or domain_path
 
     if domain_import_path:
-        print(f"Deriving domain from {domain_import_path}...")
+        logger.debug("Deriving domain from %s...", domain_import_path)
         path, name = (re.split(r":(?![\\/])", domain_import_path, 1) + [None])[:2]
         import_name = prepare_import(path)
         domain = locate_domain(import_name, name)

tests/cli/test_docs.py

@@ -0,0 +1,10 @@
+from typer.testing import CliRunner
+
+from protean.cli.docs import app
+
+
+def test_main():
+    runner = CliRunner()
+    result = runner.invoke(app)
+
+    assert result.exit_code == 0

tests/cli/test_domain_loading.py

@@ -1,20 +1,16 @@
+"""Test cases for domain loading from various sources"""
+
 import os
 import sys
 
 from pathlib import Path
 
 import pytest
 
-from typer.testing import CliRunner
-
 from protean import Domain
 from protean.cli import NoDomainException
 from protean.utils.domain import derive_domain, find_domain_in_module
-
-
-@pytest.fixture
-def runner():
-    return CliRunner()
+from tests.shared import change_working_directory_to
 
 
 def test_find_domain_in_module():
@@ -62,44 +58,36 @@ def reset_path(self, request):
         sys.path[:] = original_path
         os.chdir(cwd)
 
-    def change_working_directory_to(self, path):
-        test_path = (
-            Path(__file__) / ".." / ".." / "support" / "test_domains" / path
-        ).resolve()
-
-        os.chdir(test_path)
-        sys.path.insert(0, test_path)
-
     def test_loading_domain_named_as_domain(self):
-        self.change_working_directory_to("test1")
+        change_working_directory_to("test1")
 
         domain = derive_domain("basic")
         assert domain is not None
         assert domain.domain_name == "BASIC"
 
     def test_loading_domain_under_directory(self):
-        self.change_working_directory_to("test2")
+        change_working_directory_to("test2")
 
         domain = derive_domain("src/folder")
         assert domain is not None
         assert domain.domain_name == "FOLDER"
 
     def test_loading_domain_from_module(self):
-        self.change_working_directory_to("test3")
+        change_working_directory_to("test3")
 
         domain = derive_domain("nested.web")
         assert domain is not None
         assert domain.domain_name == "WEB"
 
     def test_loading_domain_from_instance(self):
-        self.change_working_directory_to("test4")
+        change_working_directory_to("test4")
 
         domain = derive_domain("instance:dom2")
         assert domain is not None
         assert domain.domain_name == "INSTANCE"
 
     def test_loading_domain_from_invalid_module(self):
-        self.change_working_directory_to("test5")
+        change_working_directory_to("test5")
 
         with pytest.raises(NoDomainException):
             derive_domain("dummy")

tests/cli/test_generate_docker_compose.py

@@ -9,27 +9,14 @@
 
 from protean.cli import derive_domain
 from protean.cli.generate import app, docker_compose
+from tests.shared import change_working_directory_to
 
 runner = CliRunner()
 
 
-def change_working_directory_to(path):
-    """Change working directory to a specific test directory
-    and add it to the Python path so that the test can import.
-
-    The test directory is expected to be in `support/test_domains`.
-    """
-    test_path = (
-        Path(__file__) / ".." / ".." / "support" / "test_domains" / path
-    ).resolve()
-
-    os.chdir(test_path)
-    sys.path.insert(0, test_path)
-
-
 class TestGenerateDockerCompose:
     @pytest.fixture(autouse=True)
-    def reset_path(self, request):
+    def reset_path(self):
         """Reset sys.path after every test run"""
         original_path = sys.path[:]
         cwd = Path.cwd()

tests/cli/test_shell.py

@@ -0,0 +1,48 @@
+import os
+import sys
+
+from pathlib import Path
+
+import pytest
+
+from typer.testing import CliRunner
+
+from protean.cli import app
+from protean.exceptions import NoDomainException
+from tests.shared import change_working_directory_to
+
+runner = CliRunner()
+
+
+class TestShellCommand:
+    @pytest.fixture(autouse=True)
+    def reset_path(self):
+        """Reset sys.path after every test run"""
+        original_path = sys.path[:]
+        cwd = Path.cwd()
+
+        yield
+
+        sys.path[:] = original_path
+        os.chdir(cwd)
+
+    def test_shell_command_success(self):
+        change_working_directory_to("test7")
+
+        args = ["shell", "publishing.py"]
+
+        # Run the shell command
+        result = runner.invoke(app, args)
+
+        # Assertions
+        print(result.output)
+        assert result.exit_code == 0
+
+    def test_shell_command_raises_no_domain_exception_when_no_domain_is_found(self):
+        change_working_directory_to("test7")
+
+        args = ["shell", "foobar"]
+
+        # Run the shell command and expect it to raise an exception
+        with pytest.raises(NoDomainException):
+            runner.invoke(app, args, catch_exceptions=False)