Custom commands

sanic/app.py

@@ -74,6 +74,7 @@
 from sanic.log import LOGGING_CONFIG_DEFAULTS, error_logger, logger
 from sanic.logging.setup import setup_logging
 from sanic.middleware import Middleware, MiddlewareLocation
+from sanic.mixins.commands import CommandMixin
 from sanic.mixins.listeners import ListenerEvent
 from sanic.mixins.startup import StartupMixin
 from sanic.mixins.static import StaticHandleMixin
@@ -119,6 +120,7 @@ class Sanic(
     StaticHandleMixin,
     BaseSanic,
     StartupMixin,
+    CommandMixin,
     metaclass=TouchUpMeta,
 ):
     """The main application instance
@@ -189,6 +191,7 @@ class to use for the application. Defaults to `None`.
         "_blueprint_order",
         "_delayed_tasks",
         "_ext",
+        "_future_commands",
         "_future_exceptions",
         "_future_listeners",
         "_future_middleware",

sanic/base/root.py

@@ -4,6 +4,7 @@
 
 from sanic.base.meta import SanicMeta
 from sanic.exceptions import SanicException
+from sanic.mixins.commands import CommandMixin
 from sanic.mixins.exceptions import ExceptionMixin
 from sanic.mixins.listeners import ListenerMixin
 from sanic.mixins.middleware import MiddlewareMixin
@@ -22,6 +23,7 @@ class BaseSanic(
     ListenerMixin,
     ExceptionMixin,
     SignalMixin,
+    CommandMixin,
     metaclass=SanicMeta,
 ):
     __slots__ = ("name",)

sanic/blueprints.py

@@ -97,6 +97,7 @@ class Blueprint(BaseSanic):
 
     __slots__ = (
         "_apps",
+        "_future_commands",
         "_future_routes",
         "_future_statics",
         "_future_middleware",
@@ -510,6 +511,11 @@ def register(self, app, options):
                 ),
             )
 
+        if self._future_commands:
+            raise SanicException(
+                "Registering commands with blueprints is not supported."
+            )
+
     async def dispatch(self, *args, **kwargs):
         """Dispatch a signal event
 

sanic/cli/app.py

@@ -12,6 +12,7 @@
 from sanic.cli.arguments import Group
 from sanic.cli.base import SanicArgumentParser, SanicHelpFormatter
 from sanic.cli.console import SanicREPL
+from sanic.cli.executor import Executor, make_executor_parser
 from sanic.cli.inspector import make_inspector_parser
 from sanic.cli.inspector_client import InspectorClient
 from sanic.helpers import _default, is_atty
@@ -64,11 +65,11 @@ def __init__(self) -> None:
         )
         self.args: Namespace = Namespace()
         self.groups: List[Group] = []
-        self.inspecting = False
+        self.run_mode = "serve"
 
     def attach(self):
         if len(sys.argv) > 1 and sys.argv[1] == "inspect":
-            self.inspecting = True
+            self.run_mode = "inspect"
             self.parser.description = get_logo(True)
             make_inspector_parser(self.parser)
             return
@@ -78,8 +79,13 @@ def attach(self):
             instance.attach()
             self.groups.append(instance)
 
+        if len(sys.argv) > 2 and sys.argv[2] == "exec":
+            self.run_mode = "exec"
+            self.parser.description = get_logo(True)
+            make_executor_parser(self.parser)
+
     def run(self, parse_args=None):
-        if self.inspecting:
+        if self.run_mode == "inspect":
             self._inspector()
             return
 
@@ -92,13 +98,22 @@ def run(self, parse_args=None):
             parse_args = ["--version"]
 
         if not legacy_version:
+            if self.run_mode == "exec":
+                parse_args = [
+                    a
+                    for a in (parse_args or sys.argv[1:])
+                    if a not in "-h --help".split()
+                ]
             parsed, unknown = self.parser.parse_known_args(args=parse_args)
             if unknown and parsed.factory:
                 for arg in unknown:
                     if arg.startswith("--"):
                         self.parser.add_argument(arg.split("=")[0])
 
-        self.args = self.parser.parse_args(args=parse_args)
+        if self.run_mode == "exec":
+            self.args, _ = self.parser.parse_known_args(args=parse_args)
+        else:
+            self.args = self.parser.parse_args(args=parse_args)
         self._precheck()
         app_loader = AppLoader(
             self.args.target, self.args.factory, self.args.simple, self.args
@@ -110,6 +125,12 @@ def run(self, parse_args=None):
         except ValueError as e:
             error_logger.exception(f"Failed to run app: {e}")
         else:
+            if self.run_mode == "exec":
+                self._executor(app, kwargs)
+                return
+            elif self.run_mode != "serve":
+                raise ValueError(f"Unknown run mode: {self.run_mode}")
+
             if self.args.repl:
                 self._repl(app)
             for http_version in self.args.http:
@@ -152,6 +173,10 @@ def _inspector(self):
                 kwargs["args"] = positional[1:]
         InspectorClient(host, port, secure, raw, api_key).do(action, **kwargs)
 
+    def _executor(self, app: Sanic, kwargs: dict):
+        args = sys.argv[3:]
+        Executor(app, kwargs).run(self.args.command, args)
+
     def _repl(self, app: Sanic):
         if is_atty():
 

sanic/cli/arguments.py

@@ -68,6 +68,21 @@ def attach(self):
             ),
         )
 
+        self.container.add_argument(
+            "action",
+            nargs="?",
+            default="serve",
+            choices=[
+                "serve",
+                "exec",
+            ],
+            help=(
+                "Action to perform.\n"
+                "\tserve: Run the Sanic app\n"
+                "\texec: Execute a command in the Sanic app context\n"
+            ),
+        )
+
 
 class ApplicationGroup(Group):
     name = "Application"

sanic/cli/executor.py

@@ -0,0 +1,90 @@
+import shutil
+
+from argparse import ArgumentParser
+from asyncio import run
+from inspect import signature
+from typing import Callable, Dict, List
+
+from sanic import Sanic
+from sanic.application.logo import get_logo
+from sanic.cli.base import (
+    SanicArgumentParser,
+    SanicHelpFormatter,
+)
+
+
+def make_executor_parser(parser: ArgumentParser) -> None:
+    parser.add_argument(
+        "command",
+        help="Command to execute",
+    )
+
+
+class ExecutorSubParser(ArgumentParser):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if not self.description:
+            self.description = ""
+        self.description = get_logo(True) + self.description
+
+
+class Executor:
+    def __init__(self, app: Sanic, kwargs: dict) -> None:
+        self.app = app
+        self.kwargs = kwargs
+        self.commands = self._make_commands()
+        self.parser = self._make_parser()
+
+    def run(self, command: str, args: List[str]) -> None:
+        if command == "exec":
+            args = ["--help"]
+        parsed_args = self.parser.parse_args(args)
+        if command not in self.commands:
+            raise ValueError(f"Unknown command: {command}")
+        parsed_kwargs = vars(parsed_args)
+        parsed_kwargs.pop("command")
+        run(self.commands[command](**parsed_kwargs))
+
+    def _make_commands(self) -> Dict[str, Callable]:
+        commands = {c.name: c.func for c in self.app._future_commands}
+        return commands
+
+    def _make_parser(self) -> SanicArgumentParser:
+        width = shutil.get_terminal_size().columns
+        parser = SanicArgumentParser(
+            prog="sanic",
+            description=get_logo(True),
+            formatter_class=lambda prog: SanicHelpFormatter(
+                prog,
+                max_help_position=36 if width > 96 else 24,
+                indent_increment=4,
+                width=None,
+            ),
+        )
+
+        subparsers = parser.add_subparsers(
+            dest="command",
+            title="  Commands",
+            parser_class=ExecutorSubParser,
+        )
+        for command in self.app._future_commands:
+            sub = subparsers.add_parser(
+                command.name,
+                help=command.func.__doc__ or f"Execute {command.name}",
+                formatter_class=SanicHelpFormatter,
+            )
+            self._add_arguments(sub, command.func)
+
+        return parser
+
+    def _add_arguments(self, parser: ArgumentParser, func: Callable) -> None:
+        sig = signature(func)
+        for param in sig.parameters.values():
+            kwargs = {}
+            if param.default is not param.empty:
+                kwargs["default"] = param.default
+            parser.add_argument(
+                f"--{param.name}",
+                help=param.annotation,
+                **kwargs,
+            )

sanic/mixins/commands.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from functools import wraps
+from inspect import isawaitable
+from typing import Callable, Optional, Set, Union
+
+from sanic.base.meta import SanicMeta
+from sanic.models.futures import FutureCommand
+
+
+class CommandMixin(metaclass=SanicMeta):
+    def __init__(self, *args, **kwargs) -> None:
+        self._future_commands: Set[FutureCommand] = set()
+
+    def command(
+        self, maybe_func: Optional[Callable] = None, *, name: str = ""
+    ) -> Union[Callable, Callable[[Callable], Callable]]:
+        def decorator(f):
+            @wraps(f)
+            async def decorated_function(*args, **kwargs):
+                response = f(*args, **kwargs)
+                if isawaitable(response):
+                    response = await response
+                return response
+
+            self._future_commands.add(
+                FutureCommand(name or f.__name__, decorated_function)
+            )
+            return decorated_function
+
+        return decorator(maybe_func) if maybe_func else decorator

sanic/mixins/static.py

@@ -179,15 +179,16 @@ def _register_static(
         Register a static directory handler with Sanic by adding a route to the
         router and registering a handler.
         """
+        file_or_directory: PathLike
 
         if isinstance(static.file_or_directory, bytes):
-            file_or_directory = static.file_or_directory.decode("utf-8")
+            file_or_directory = Path(static.file_or_directory.decode("utf-8"))
         elif isinstance(static.file_or_directory, PurePath):
-            file_or_directory = str(static.file_or_directory)
-        elif not isinstance(static.file_or_directory, str):
-            raise ValueError("Invalid file path string.")
-        else:
             file_or_directory = static.file_or_directory
+        elif isinstance(static.file_or_directory, str):
+            file_or_directory = Path(static.file_or_directory)
+        else:
+            raise ValueError("Invalid file path string.")
 
         uri = static.uri
         name = static.name
@@ -224,7 +225,7 @@ def _register_static(
         _handler = wraps(self._static_request_handler)(
             partial(
                 self._static_request_handler,
-                file_or_directory=file_or_directory,
+                file_or_directory=str(file_or_directory),
                 use_modified_since=static.use_modified_since,
                 use_content_range=static.use_content_range,
                 stream_large_files=static.stream_large_files,

sanic/models/futures.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Dict, Iterable, List, NamedTuple, Optional, Union
+from typing import Callable, Dict, Iterable, List, NamedTuple, Optional, Union
 
 from sanic.handlers.directory import DirectoryHandler
 from sanic.models.handler_types import (
@@ -70,3 +70,8 @@ class FutureSignal(NamedTuple):
 
 
 class FutureRegistry(set): ...
+
+
+class FutureCommand(NamedTuple):
+    name: str
+    func: Callable

tests/fake/server.py

@@ -52,3 +52,18 @@ def create_app_with_args(args):
         logger.info(f"target={args.target}")
 
     return app
+
+
+@app.command
+async def foo(one, two: str, three: str = "..."):
+    logger.info(f"FOO {one=} {two=} {three=}")
+
+
+@app.command
+def bar():
+    logger.info("BAR")
+
+
+@app.command(name="qqq")
+async def baz():
+    logger.info("BAZ")

tests/test_cli.py

@@ -369,3 +369,38 @@ def run():
 
     run()
     assert record in caplog.record_tuples
+
+
+def test_command_no_args(caplog):
+    args = ["fake.server.app", "exec", "foo"]
+    with patch("sys.argv", ["sanic", *args]):
+        lines = capture(args, caplog)
+    assert "FOO one=None two=None three='...'" in lines
+
+
+def test_command_with_args(caplog):
+    args = [
+        "fake.server.app",
+        "exec",
+        "foo",
+        "--one=1",
+        "--two=2",
+        "--three=3",
+    ]
+    with patch("sys.argv", ["sanic", *args]):
+        lines = capture(args, caplog)
+    assert "FOO one='1' two='2' three='3'" in lines
+
+
+def test_command_with_sync_handler(caplog):
+    args = ["fake.server.app", "exec", "bar"]
+    with patch("sys.argv", ["sanic", *args]):
+        lines = capture(args, caplog)
+    assert "BAR" in lines
+
+
+def test_command_with_renamed_command(caplog):
+    args = ["fake.server.app", "exec", "qqq"]
+    with patch("sys.argv", ["sanic", *args]):
+        lines = capture(args, caplog)
+    assert "BAZ" in lines