ls cli (#204)

* Working on ls

* ls working with tests

* Fix test

Author: Kimahriman

python/hdfs_native/cli.py

@@ -6,19 +6,35 @@
 import stat
 import sys
 from argparse import ArgumentParser, Namespace
+from collections import defaultdict
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import List, Optional, Sequence, Tuple
+from datetime import datetime
+from typing import Dict, List, Optional, Sequence, Tuple, Union
 from urllib.parse import urlparse
 
 from hdfs_native import Client
-from hdfs_native._internal import WriteOptions
+from hdfs_native._internal import FileStatus, WriteOptions
+
+__all__ = ["main"]
 
 
 @functools.cache
 def _get_client(connection_url: Optional[str] = None):
     return Client(connection_url)
 
 
+def _prefix_for_url(url: str) -> str:
+    parsed = urlparse(url)
+
+    if parsed.scheme:
+        prefix = f"{parsed.scheme}://{parsed.hostname}"
+        if parsed.port:
+            prefix += f":{parsed.port}"
+        return prefix
+
+    return ""
+
+
 def _client_for_url(url: str) -> Client:
     parsed = urlparse(url)
 
@@ -217,6 +233,125 @@ def get(args: Namespace):
                 f.result()
 
 
+def ls(args: Namespace):
+    def human_size(num: int):
+        if num < 1024:
+            return str(num)
+
+        adjusted = num / 1024.0
+        for unit in ("K", "M", "G", "T", "P", "E", "Z"):
+            if abs(adjusted) < 1024.0:
+                return f"{adjusted:.1f} {unit}"
+            adjusted /= 1024.0
+        return f"{adjusted:.1f} Y"
+
+    def parse_status(status: FileStatus, prefix: str) -> Dict[str, Union[int, str]]:
+        file_time = status.modification_time
+        if args.access_time:
+            file_time = status.access_time
+
+        file_time_string = datetime.fromtimestamp(file_time / 1000).strftime(
+            r"%Y-%m-%d %H:%M"
+        )
+
+        permission = status.permission
+        if status.isdir:
+            permission |= stat.S_IFDIR
+        else:
+            permission |= stat.S_IFREG
+
+        mode = stat.filemode(permission)
+
+        if args.human_readable:
+            length_string = human_size(status.length)
+        else:
+            length_string = str(status.length)
+
+        path = prefix + status.path
+
+        return {
+            "mode": mode,
+            "replication": str(status.replication) if status.replication else "-",
+            "owner": status.owner,
+            "group": status.group,
+            "length": status.length,
+            "length_formatted": length_string,
+            "time": file_time,
+            "time_formatted": file_time_string,
+            "path": path,
+        }
+
+    def get_widths(parsed: list[dict]) -> dict[str, int]:
+        widths: dict[str, int] = defaultdict(lambda: 0)
+
+        for file in parsed:
+            for key, value in file.items():
+                if isinstance(value, str):
+                    widths[key] = max(widths[key], len(value))
+
+        return widths
+
+    def print_files(
+        parsed: List[Dict[str, Union[int, str]]],
+        widths: Optional[Dict[str, int]] = None,
+    ):
+        if args.sort_time:
+            parsed = sorted(parsed, key=lambda x: x["time"], reverse=not args.reverse)
+        elif args.sort_size:
+            parsed = sorted(parsed, key=lambda x: x["length"], reverse=not args.reverse)
+
+        def format(
+            file: Dict[str, Union[int, str]],
+            field: str,
+            right_align: bool = False,
+        ):
+            value = str(file[field])
+
+            width = len(value)
+            if widths and field in widths:
+                width = widths[field]
+
+            if right_align:
+                return f"{value:>{width}}"
+            return f"{value:{width}}"
+
+        for file in parsed:
+            if args.path_only:
+                print(file["path"])
+            else:
+                formatted_fields = [
+                    format(file, "mode"),
+                    format(file, "replication"),
+                    format(file, "owner"),
+                    format(file, "group"),
+                    format(file, "length_formatted", True),
+                    format(file, "time_formatted"),
+                    format(file, "path"),
+                ]
+                print(" ".join(formatted_fields))
+
+    for url in args.path:
+        client = _client_for_url(url)
+        for path in _glob_path(client, _path_for_url(url)):
+            status = client.get_file_info(path)
+
+            prefix = _prefix_for_url(url)
+
+            if status.isdir:
+                parsed = [
+                    parse_status(status, prefix)
+                    for status in client.list_status(path, args.recursive)
+                ]
+
+                if not args.path_only:
+                    print(f"Found {len(parsed)} items")
+
+                widths = get_widths(parsed)
+                print_files(parsed, widths)
+            else:
+                print_files([parse_status(status, prefix)])
+
+
 def mkdir(args: Namespace):
     create_parent = args.parent
 
@@ -427,6 +562,64 @@ def main(in_args: Optional[Sequence[str]] = None):
     )
     get_parser.set_defaults(func=get)
 
+    ls_parser = subparsers.add_parser(
+        "ls",
+        help="List contents that match the specified patterns",
+        description="""List contents that match the specified patterns. For a directory, list its
+        direct children.""",
+    )
+    ls_parser.add_argument(
+        "-C",
+        "--path-only",
+        action="store_true",
+        default=False,
+        help="Display the path of files and directories only.",
+    )
+    ls_parser.add_argument(
+        "-H",
+        "--human-readable",
+        action="store_true",
+        default=False,
+        help="Formats the sizes of files in a human-readable fashion rather than a number of bytes",
+    )
+    ls_parser.add_argument(
+        "-R",
+        "--recursive",
+        action="store_true",
+        default=False,
+        help="Recursively list the contents of directories",
+    )
+    ls_parser.add_argument(
+        "-t",
+        "--sort-time",
+        action="store_true",
+        default=False,
+        help="Sort files by modification time (most recent first)",
+    )
+    ls_parser.add_argument(
+        "-S",
+        "--sort-size",
+        action="store_true",
+        default=False,
+        help="Sort files by size (largest first)",
+    )
+    ls_parser.add_argument(
+        "-r",
+        "--reverse",
+        action="store_true",
+        default=False,
+        help="Reverse the order of the sort",
+    )
+    ls_parser.add_argument(
+        "-u",
+        "--access-time",
+        action="store_true",
+        default=False,
+        help="Use the last access time instead of modification time for display and sorting",
+    )
+    ls_parser.add_argument("path", nargs="+", help="Path to display contents of")
+    ls_parser.set_defaults(func=ls)
+
     mkdir_parser = subparsers.add_parser(
         "mkdir",
         help="Create a directory",

python/tests/test_cli.py

@@ -1,8 +1,11 @@
 import contextlib
+import dataclasses
 import io
 import os
+import re
 import stat
 from tempfile import TemporaryDirectory
+from typing import Callable, Iterator, List, Literal, Optional, Tuple, overload
 
 import pytest
 
@@ -18,22 +21,37 @@ def assert_not_exists(client: Client, path: str):
         pass
 
 
+@overload
+def capture_stdout(func: Callable[[], None], text: Literal[False]) -> bytes: ...
+
+
+@overload
+def capture_stdout(func: Callable[[], None], text: Literal[True] = True) -> str: ...
+
+
+def capture_stdout(func: Callable[[], None], text: bool = True):
+    buf = io.BytesIO()
+    with contextlib.redirect_stdout(io.TextIOWrapper(buf)) as wrapper:
+        func()
+        if text:
+            wrapper.seek(0)
+            return wrapper.read()
+        else:
+            return buf.getvalue()
+
+
 def test_cat(client: Client):
     with client.create("/testfile") as file:
         file.write(b"1234")
 
-    buf = io.BytesIO()
-    with contextlib.redirect_stdout(io.TextIOWrapper(buf)):
-        cli_main(["cat", "/testfile"])
-        assert buf.getvalue() == b"1234"
+    output = capture_stdout(lambda: cli_main(["cat", "/testfile"]), False)
+    assert output == b"1234"
 
     with client.create("/testfile2") as file:
         file.write(b"5678")
 
-    buf = io.BytesIO()
-    with contextlib.redirect_stdout(io.TextIOWrapper(buf)):
-        cli_main(["cat", "/testfile", "/testfile2"])
-        assert buf.getvalue() == b"12345678"
+    output = capture_stdout(lambda: cli_main(["cat", "/testfile", "/testfile2"]), False)
+    assert output == b"12345678"
 
     with pytest.raises(FileNotFoundError):
         cli_main(["cat", "/nonexistent"])
@@ -159,6 +177,81 @@ def test_get(client: Client):
             assert file.read() == data
 
 
+def test_ls(client: Client):
+    @dataclasses.dataclass
+    class FileOutput:
+        permission: str
+        replication: str
+        size: str
+        path: str
+
+    def parse_output(output: str) -> Iterator[Tuple[int, List[FileOutput]]]:
+        current_items: Optional[int] = None
+        current_batch: List[FileOutput] = []
+
+        for line in output.split("\n"):
+            if match := re.match(r"Found (\d)+ items", line):
+                if current_items is not None:
+                    yield (current_items, current_batch)
+
+                current_items = int(match.group(1))
+                current_batch = []
+
+            elif line.strip():
+                match = re.match(
+                    r"(\S+)\s+(\S+)\s+\S+\s+\S+\s+([0-9.]+(?: \w)?)\s+\S+\s+\S+\s+(\S+)",
+                    line,
+                )
+                assert match is not None
+                current_batch.append(
+                    FileOutput(
+                        permission=match.group(1),
+                        replication=match.group(2),
+                        size=match.group(3),
+                        path=match.group(4),
+                    )
+                )
+
+        if current_items is not None and len(current_batch) > 0:
+            yield (current_items, current_batch)
+
+    with pytest.raises(FileNotFoundError):
+        cli_main(["ls", "/fake"])
+
+    with client.create("/testfile1") as f:
+        f.write(bytes(range(10)))
+
+    with client.create("/testfile2") as f:
+        for i in range(1024):
+            f.write(i.to_bytes(4, "big"))
+
+    client.mkdirs("/testdir")
+
+    directory = FileOutput("drwxr-xr-x", "-", "0", "/testdir")
+    file1 = FileOutput("-rw-r--r--", "3", "10", "/testfile1")
+    file2 = FileOutput("-rw-r--r--", "3", "4096", "/testfile2")
+
+    def check_output(command: List[str], expected: List[FileOutput]):
+        groups = list(parse_output(capture_stdout(lambda: cli_main(command))))
+        assert len(groups) == 1
+        assert groups[0][0] == 3
+        assert len(groups[0][1]) == 3
+        assert groups[0][1] == expected
+
+    check_output(["ls", "/"], [directory, file1, file2])
+    check_output(["ls", "-t", "/"], [directory, file2, file1])
+    check_output(["ls", "-r", "-t", "/"], [file1, file2, directory])
+    check_output(["ls", "-S", "/"], [file2, file1, directory])
+    check_output(["ls", "-r", "-S", "/"], [directory, file1, file2])
+
+    check_output(
+        ["ls", "-H", "/"], [directory, file1, dataclasses.replace(file2, size="4.0 K")]
+    )
+
+    output = capture_stdout(lambda: cli_main(["ls", "-C", "/"])).strip().split("\n")
+    assert output == [directory.path, file1.path, file2.path]
+
+
 def test_mkdir(client: Client):
     cli_main(["mkdir", "/testdir"])
     assert client.get_file_info("/testdir").isdir