ISSUE-32: New command, scan directory

dosh/cli.py

@@ -4,7 +4,6 @@
 from typing import Final, List, Optional, Tuple
 
 from dosh import DoshInitializer
-from dosh.commands.base import CommandStatus
 from dosh.commands.internal import generate_help, init_config
 from dosh.config import ConfigParser
 from dosh.environments import ENVIRONMENTS
@@ -73,7 +72,7 @@ def get_task_param(self, tasks: List[str]) -> Tuple[str, List[str]]:
         return sys.argv[task_index], sys.argv[task_index + 1 :]
 
 
-class CLI:
+class CLI:  # pylint: disable=too-few-public-methods
     """DOSH command line interface."""
 
     def __init__(self) -> None:
@@ -105,9 +104,14 @@ def run(self) -> None:
         task_name, task_params = self.arg_parser.get_task_param(tasks)
 
         if task_name == "init":
-            self.run_init()
+            init_config(DoshInitializer.config_path)
         elif task_name == "help":
-            self.run_help()
+            output = generate_help(
+                tasks=self.conf_parser.tasks,
+                description=self.conf_parser.description,
+                epilog=self.conf_parser.epilog,
+            )
+            print(output)
         else:
             dosh_env = ENVIRONMENTS["DOSH_ENV"] or "not specified."
             working_directory = self.arg_parser.get_current_working_directory()
@@ -116,20 +120,3 @@ def run(self) -> None:
             logger.debug("WORKING DIRECTORY: %s", working_directory)
 
             self.conf_parser.run_task(task_name, task_params)
-
-    def run_init(self) -> None:
-        """Create new config."""
-        result = init_config(DoshInitializer.config_path)
-        if result.status == CommandStatus.OK:
-            print(result.message)
-        elif result.status == CommandStatus.ERROR:
-            print(result.message, file=sys.stderr)
-
-    def run_help(self) -> None:
-        """Print help output."""
-        output = generate_help(
-            tasks=self.conf_parser.tasks,
-            description=self.conf_parser.description,
-            epilog=self.conf_parser.epilog,
-        )
-        print(output)

dosh/commands/__init__.py

@@ -12,6 +12,7 @@
 COMMANDS: Final = {
     # general purpose
     "clone": cmd.clone,
+    "ls": cmd.scan_directory,
     "run": cmd.run,
     "run_url": cmd.run_url,
     # package managers

dosh/commands/base.py

@@ -7,7 +7,7 @@
 from dataclasses import dataclass, field
 from enum import Enum
 from pathlib import Path
-from typing import Any, Callable, Dict, Generic, List, Optional, TypeVar
+from typing import Any, Callable, Dict, List, TypeVar
 from urllib.parse import urlparse
 
 from dosh import DoshInitializer
@@ -19,11 +19,8 @@
 logger = get_logger()
 
 
-class CommandStatus(Enum):
-    """Command status for handling the results."""
-
-    OK = "ok"
-    ERROR = "error"
+class CommandException(Exception):
+    """Dosh command exception."""
 
 
 class OperatingSystem(str, Enum):
@@ -51,24 +48,7 @@ def get_current(cls) -> OperatingSystem:
         return cls.UNSUPPORTED
 
 
-@dataclass
-class CommandResult(Generic[T]):
-    """Return type of command functions."""
-
-    status: CommandStatus
-    message: Optional[str] = None
-    response: Optional[T] = None
-
-    def __repr__(self) -> str:
-        """Return result as repr."""
-        return str(self.response)
-
-    def __bool__(self) -> bool:
-        """Return false if command status is not ok."""
-        return self.status == CommandStatus.OK
-
-
-CommandCallable = Callable[..., CommandResult[Any]]
+CommandCallable = Callable[..., Any]
 
 
 @dataclass
@@ -95,10 +75,11 @@ def check_command(
 
     def decorator(func: CommandCallable) -> CommandCallable:
         @functools.wraps(func)
-        def wrapper(*args: Any, **kwargs: Any) -> CommandResult[Any]:
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
             if shutil.which(command_name) is None:
-                message = f"The command `{command_name}` doesn't exist in this system."
-                return CommandResult(CommandStatus.ERROR, message=message)
+                raise CommandException(
+                    f"The command `{command_name}` doesn't exist in this system."
+                )
             return func(*args, **kwargs)
 
         return wrapper

dosh/commands/external.py

@@ -1,16 +1,16 @@
 """Available commands for `dosh.star`."""
 from __future__ import annotations
 
+import logging
+import os
 import shutil
 import subprocess
 import urllib.request
 from pathlib import Path
-from subprocess import CompletedProcess
 from typing import List, Optional
 
 from dosh.commands.base import (
-    CommandResult,
-    CommandStatus,
+    CommandException,
     check_command,
     copy_tree,
     is_url_valid,
@@ -23,18 +23,15 @@
 
 
 @check_command("apt")
-def apt_install(packages: List[str]) -> CommandResult[None]:
+def apt_install(packages: List[str]) -> None:
     """Install packages with apt."""
     command = "apt install"
 
     run(f"{command} {' '.join(packages)}")
-    return CommandResult(CommandStatus.OK)
 
 
 @check_command("brew")
-def brew_install(
-    packages: LuaTable, options: Optional[LuaTable] = None
-) -> CommandResult[None]:
+def brew_install(packages: LuaTable, options: Optional[LuaTable] = None) -> None:
     """Install packages with brew."""
     if options is None:
         options = lua_runtime.table()
@@ -48,23 +45,21 @@ def brew_install(
         command = f"{command} --cask"
 
     run(f"{command} {' '.join(list(packages.values()))}")
-    return CommandResult(CommandStatus.OK)
 
 
 @check_command("winget")
-def winget_install(packages: List[str]) -> CommandResult[None]:
+def winget_install(packages: List[str]) -> None:
     """Install packages with winget."""
     command = "winget install -e --id"
-
     run(f"{command} {' '.join(packages)}")
-    return CommandResult(CommandStatus.OK)
 
 
-def copy(src: str, dst: str) -> CommandResult[None]:
+def copy(src: str, dst: str) -> None:
     """Copy files from source to destination. It works like `cp` command."""
     dst_path = normalize_path(dst)
     glob_index = -1
     src_splitted = src.split("/")
+
     for index, value in enumerate(src_splitted):
         if "*" in value:
             glob_index = index
@@ -78,11 +73,9 @@ def copy(src: str, dst: str) -> CommandResult[None]:
         path = normalize_path(src)
         copy_tree(path, dst_path / path.name if dst_path.exists() else dst_path)
 
-    return CommandResult(CommandStatus.OK)
-
 
 @check_command("git")
-def clone(url: str, options: Optional[LuaTable] = None) -> CommandResult[None]:
+def clone(url: str, options: Optional[LuaTable] = None) -> None:
     """Clone repository from VCS."""
     if options is None:
         options = lua_runtime.table()
@@ -97,18 +90,54 @@ def clone(url: str, options: Optional[LuaTable] = None) -> CommandResult[None]:
         command = f"git clone {url} {destination}"
         run(command)
 
-    return CommandResult(CommandStatus.OK)
 
+def scan_directory(parent_dir: str = ".", opts: Optional[LuaTable] = None) -> LuaTable:
+    """
+    List files and directories.
 
-def run(command: str) -> CommandResult[str]:
-    """Run a shell command using subprocess."""
-    logger.info("[RUN] %s", command)
+    Optional parameters:
+        include_files: boolean (default: true)
+        include_dirs: boolean (default: true)
+        excludes: list[str] (default: [])
+    """
+    parent = normalize_path(parent_dir)
+    options = opts or lua_runtime.table()
+
+    if not parent.is_dir():
+        raise CommandException(f"Not a folder: {parent_dir}")
+
+    files, directories = [], []
+
+    for item in parent.iterdir():
+        if item.is_dir():
+            directories.append(str(item))
+        elif item.is_file():
+            files.append(str(item))
+
+    items = []
 
-    response_lines = []
+    if options["include_files"] is not False:
+        items.extend(files)
+
+    if options["include_dirs"] is not False:
+        items.extend(directories)
+
+    excludes = list((options["excludes"] or lua_runtime.table()).values())
+    if excludes:
+        items = list(filter(lambda i: i.rsplit(os.sep, 1)[-1] not in excludes, items))
+
+    response: LuaTable = lua_runtime.table_from(sorted(items))
+    return response
+
+
+def __run_command_and_log_output(content: str) -> int:
     with subprocess.Popen(
-        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
+        content,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        shell=True,
     ) as proc:
-        for out, log in [(proc.stdout, logger.debug), (proc.stderr, logger.error)]:
+        for out, level in [(proc.stdout, logging.DEBUG), (proc.stderr, logging.ERROR)]:
             if out is None:
                 continue
 
@@ -117,43 +146,34 @@ def run(command: str) -> CommandResult[str]:
                 if not line:
                     break
 
-                line_str = line.decode().rstrip()
-                response_lines.append(line_str)
-                log(line_str)
+                logger.log(level, line.decode().rstrip())
 
-        returncode = proc.wait()
-        if returncode == 0:
-            status = CommandStatus.OK
-        else:
-            status = CommandStatus.ERROR
+    return proc.wait()
 
-    return CommandResult(status, response="\n".join(response_lines))
 
+def run(command: str) -> int:
+    """Run a shell command using subprocess."""
+    logger.info("[RUN] %s", command)
+    return __run_command_and_log_output(command)
 
-def run_url(url: str) -> CommandResult[CompletedProcess[bytes]]:
+
+def run_url(url: str) -> int:
     """Run a remote shell script directly."""
     if not is_url_valid(url):
-        message = f"URL is not valid: {url}"
-        return CommandResult(CommandStatus.ERROR, message=message)
+        raise CommandException(f"URL is not valid: {url}")
+
     with urllib.request.urlopen(url) as response:
         content = response.read()
 
     logger.info("[RUN_URL] %s", url)
-    result = subprocess.run(content, shell=True, capture_output=True, check=True)
-    return CommandResult(CommandStatus.OK, response=result)
+    return __run_command_and_log_output(content)
 
 
-def exists(path: str) -> CommandResult[bool]:
+def exists(path: str) -> bool:
     """Check if the path exists in the file system."""
-    if not Path(path).exists():
-        message = f"The path `{path}` doesn't exist in this system."
-        return CommandResult(CommandStatus.ERROR, message=message, response=False)
-    return CommandResult(CommandStatus.OK, response=True)
+    return Path(path).exists()
 
 
-def exists_command(command: str) -> CommandResult[bool]:
+def exists_command(command: str) -> bool:
     """Check if the command exists."""
-    if shutil.which(command) is None:
-        message = f"The command `{command}` doesn't exist in this system."
-        return CommandResult(CommandStatus.ERROR, message=message, response=False)
-    return CommandResult(CommandStatus.OK, response=True)
+    return shutil.which(command) is not None

dosh/commands/internal.py

@@ -3,15 +3,18 @@
 from pathlib import Path
 from typing import List, Optional
 
-from dosh.commands.base import CommandResult, CommandStatus, Task
+from dosh.commands.base import Task
+from dosh.logger import get_logger
 
+logger = get_logger()
 
-def init_config(config_path: Path) -> CommandResult[None]:
+
+def init_config(config_path: Path) -> None:
     """Initialize dosh config in the current working directory."""
     if config_path.exists():
-        return CommandResult(
-            CommandStatus.ERROR,
-            message=f"The file `{config_path.name}` already exists in current working directory.",
+        logger.error(
+            "The file `%s` already exists in current working directory.",
+            config_path.name,
         )
 
     content = textwrap.dedent(
@@ -33,11 +36,8 @@ def init_config(config_path: Path) -> CommandResult[None]:
             }
         """
     )
-
     config_path.write_text(content)
-    return CommandResult(
-        CommandStatus.OK, message=f"The file `{config_path.name}` is created."
-    )
+    logger.info("The file `%s` is created.", config_path.name)
 
 
 def generate_help(

dosh/environments.py

@@ -5,7 +5,7 @@
 import os
 from typing import Final
 
-from dosh.commands.base import CommandStatus, OperatingSystem
+from dosh.commands.base import OperatingSystem
 
 __all__ = ["ENVIRONMENTS"]
 
@@ -25,7 +25,4 @@
     "IS_MACOS": CURRENT_OS == OperatingSystem.MACOS,
     "IS_LINUX": CURRENT_OS == OperatingSystem.LINUX,
     "IS_WINDOWS": CURRENT_OS == OperatingSystem.WINDOWS,
-    # command statuses
-    "STATUS_OK": CommandStatus.OK,
-    "STATUS_ERROR": CommandStatus.ERROR,
 }

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dosh"
-version = "0.0.0"  # dynamic versioning enabled
+version = "0.0.0-post.102+e331da0"  # dynamic versioning enabled
 description = "Shell-independent task manager."
 authors = ["Gökmen Görgen <gkmngrgn@gmail.com>"]
 license = "MIT"
@@ -43,7 +43,7 @@ help = "Filter tests by $name"
 addopts = "--cov=dosh --cov-report term-missing --cov-report html:./tests/cov_html"
 
 [tool.poetry-dynamic-versioning]
-enable = true
+enable = false
 vcs = "git"
 style = "semver"