Add parser stages for the planned new command-line argument parser (9a259fe) - derivepassphrase.git

src/derivepassphrase/_internals/cli_machinery.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 import abc
 import collections
 import dataclasses
+import enum
 import importlib.metadata
 import inspect
 import logging
@@ -52,6 +53,7 @@ VERSION_OUTPUT_WRAPPING_WIDTH = 72
 NOT_AN_INTEGER = "not an integer"
 NOT_A_NONNEGATIVE_INTEGER = "not a non-negative integer"
 NOT_A_POSITIVE_INTEGER = "not a positive integer"
+COMMAND_LINE_NOT_NORMALIZED = "command-line not normalized"
 
 
 # CLI parsing machinery
@@ -312,6 +314,50 @@ class CLICompositeSubcommand:
         return _canon_map_cache[self]
 
 
+ParsedCommandLineToken: TypeAlias = tuple[str, ...]
+ParsedCommandLineSection: TypeAlias = tuple[ParsedCommandLineToken, ...]
+ParsedCommandLine: TypeAlias = tuple[
+    tuple[CLISubcommand, ParsedCommandLineSection], ...
+]
+
+
+class CommandLineTokenType(str, enum.Enum):
+    """The type of a command-line token.
+
+    Attributes:
+        LONG_OPTION:
+            A long option, perhaps with an attached option value.
+        SHORT_OPTION:
+            A (perhaps trivial) cluster of short options, perhaps
+            including an option value.
+        POSITIONAL:
+            A positional argument.
+        END_OF_OPTIONS:
+            The `--` pseudo-argument.
+
+    """
+
+    LONG_OPTION = "LONG_OPTION"
+    """"""
+    SHORT_OPTION = "SHORT_OPTION"
+    """"""
+    POSITIONAL = "POSITIONAL"
+    """"""
+    END_OF_OPTIONS = "END_OF_OPTIONS"
+    """"""
+
+    @classmethod
+    def classify(cls, token: str, /) -> CommandLineTokenType:
+        """Return the type of the given token."""
+        if token == "--":  # noqa: S105
+            return cls.END_OF_OPTIONS
+        if token.startswith("--"):
+            return cls.LONG_OPTION
+        if token.startswith("-"):
+            return cls.SHORT_OPTION
+        return cls.POSITIONAL
+
+
 class ParseState(NamedTuple):
     """The internal state of a command-line parser.
 
@@ -472,6 +518,212 @@ class ParseFailure(ParseResult):
 # --------------
 
 
+def _is_valid_option_argument(
+    option: CLIOption,
+    argument: str,
+    /,
+) -> bool:
+    if isinstance(option.has_argument, bool):
+        # If any argument value doesn't matter, then the argument is
+        # valid if and only if the option takes an argument.
+        return option.has_argument
+    return argument in option.has_argument
+
+
+def _handle_option(
+    args: list[str],
+    /,
+    *,
+    state: ParseState,
+    old_command_line: collections.deque[str],
+) -> ParseFailure | str:
+    canon_map = state.subcommand.canon_map
+    arg_queue = list(args)
+    opt_ = arg_queue.pop(0)
+    if opt_ not in canon_map:
+        error = f"Unknown option {opt_!r}."
+        return ParseFailure(state, error)
+    option = cast("CLIOption", canon_map[opt_])
+    if option.has_argument:
+        if not arg_queue:
+            if not old_command_line:
+                error = f"Missing argument for option {opt_!r}."
+                return ParseFailure(state, error)
+            arg_queue.append(old_command_line.popleft())
+        value = arg_queue.pop(0)
+        if not _is_valid_option_argument(option, value):
+            error = f"Invalid argument {value!r} for option {opt_!r}."
+            return ParseFailure(state, error)
+        return "=".join([option.names[0], value])
+    return option.names[0]
+
+
+# TODO(the-13th-letter): Find a way to include debug logging in this
+# parser step.
+def normalize_options(
+    state: ParseState,
+    /,
+) -> ParseFailure | ParseSuccess[None]:
+    old_command_line = collections.deque(state.command_line)
+    new_command_line: collections.deque[str] = collections.deque()
+    while old_command_line:
+        arg_ = old_command_line.popleft()
+        token_type = CommandLineTokenType.classify(arg_)
+        # TODO(the-13th-letter): Rewrite using structural pattern matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if token_type == CommandLineTokenType.LONG_OPTION:
+            arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
+            result = _handle_option(  # may consume from old_command_line
+                arg_queue,
+                state=state,
+                old_command_line=old_command_line,
+            )
+            if isinstance(result, ParseResult):
+                return result
+            new_command_line.append(result)
+        elif token_type == CommandLineTokenType.SHORT_OPTION:
+            replacement_args: collections.deque[str] = collections.deque()
+            letters = collections.deque(arg_[1:])
+            while letters:
+                letter = letters.popleft()
+                arg_queue = (
+                    [f"-{letter}", "".join(letters)]
+                    if letters
+                    else [f"-{letter}"]
+                )
+                result = _handle_option(
+                    arg_queue,
+                    state=state,
+                    old_command_line=old_command_line,
+                )
+                if isinstance(result, ParseResult):
+                    return result
+                replacement_args.append(result)
+                if "=" in result:  # pragma: no cover [external]
+                    letters.clear()
+            old_command_line.extendleft(reversed(replacement_args))
+        else:
+            new_command_line.extend([arg_] if arg_ == "--" else ["--", arg_])
+            new_command_line.extend(old_command_line)
+            old_command_line.clear()
+            break
+    else:
+        new_command_line.append("--")
+    return ParseSuccess(
+        state._replace(command_line=tuple(new_command_line)), None
+    )
+
+
+def scan_for_eager_options(
+    state: ParseState,
+    /,
+) -> ParseEarlyExit | ParseFailure | ParseSuccess[None]:
+    option_groups = state.subcommand.contents[0]
+    eager_option_objects = {
+        option
+        for group in option_groups
+        for option in group.options
+        if option.eager
+    }
+    eager_option_names = {
+        name for option in eager_option_objects for name in option.names
+    }
+    command_line = collections.deque(state.command_line)
+    canon_map = state.subcommand.canon_map
+    while command_line:
+        arg_ = command_line.popleft()
+        token_type = CommandLineTokenType.classify(arg_)
+        if token_type == CommandLineTokenType.LONG_OPTION:
+            arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
+            long_opt_name = arg_queue.pop(0)
+            if long_opt_name not in canon_map:
+                # Unknown option.  Either the command-line is faulty,
+                # or we default to the default subcommand and let *it*
+                # make sense of this option.  Regardless, no early exit
+                # has been detected so far, so we are done.
+                return ParseSuccess(state, None)
+            option = cast("CLIOption", canon_map[long_opt_name])
+            if long_opt_name in eager_option_names:
+                canon_opt = "=".join([option.names[0], *arg_queue])
+                new_state = state._replace(command_line=(canon_opt,))
+                return ParseEarlyExit(new_state, canon_opt)
+        elif token_type == CommandLineTokenType.END_OF_OPTIONS:
+            return ParseSuccess(state, None)
+        else:  # pragma: no cover [failsafe]
+            return ParseFailure(state, COMMAND_LINE_NOT_NORMALIZED)
+    return ParseSuccess(state, None)
+
+
+def parse_until_subcommand_boundary(
+    state: ParseState,
+    /,
+) -> (
+    ParseFailure
+    | ParseSuccess[tuple[ParsedCommandLineSection, ParseState | None]]
+):
+    canon_map = state.subcommand.canon_map
+    command_line = collections.deque(state.command_line)
+    result: collections.deque[ParsedCommandLineToken] = collections.deque()
+    while command_line:
+        arg_ = command_line.popleft()
+        token_type = CommandLineTokenType.classify(arg_)
+        if token_type == CommandLineTokenType.LONG_OPTION:
+            arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
+            result.append(tuple(arg_queue))
+        elif token_type == CommandLineTokenType.END_OF_OPTIONS:
+            break
+        else:  # pragma: no cover [failsafe]
+            return ParseFailure(state, COMMAND_LINE_NOT_NORMALIZED)
+    if isinstance(state.subcommand, CLITerminalSubcommand):
+        return ParseSuccess(state, (tuple(result), None))
+    next_command_line = tuple(command_line)
+    next_subcommand_name = command_line.popleft() if command_line else None
+    if (
+        isinstance(next_subcommand_name, str)
+        and next_subcommand_name not in canon_map
+        and isinstance(state.subcommand, CLICompositeSubcommand)
+        and state.subcommand.contents[2] is not None
+    ):
+        # Unknown subcommand name; fallback exists. Run the
+        # default subcommand and reinterpret the argument
+        # (positional/subcommand) within the context of the default
+        # subcommand.
+        command_line.appendleft(next_subcommand_name)
+        next_subcommand_name = None
+    if next_subcommand_name not in canon_map:
+        # Unknown subcommand name; no fallback exists.
+        error = f"Unknown subcommand {next_subcommand_name!r}."
+        return ParseFailure(state, error)
+    next_subcommand = cast(
+        "CLISubcommand", canon_map.get(next_subcommand_name)
+    )
+    new_state = ParseState(next_command_line, next_subcommand)
+    return ParseSuccess(state, (tuple(result), new_state))
+
+
+def complete(
+    state: ParseState,
+    /,
+) -> ParseSuccess[ParsedCommandLine] | ParseFailure | ParseEarlyExit:
+    result: collections.deque[
+        tuple[CLISubcommand, ParsedCommandLineSection]
+    ] = collections.deque()
+    step_state = state
+    while True:
+        subcommand = step_state.subcommand
+        tmp3_result = ParseResult.unit(step_state)
+        tmp2_result = tmp3_result.bind(normalize_options)
+        tmp1_result = tmp2_result.bind(scan_for_eager_options)
+        step_result = tmp1_result.bind(parse_until_subcommand_boundary)
+        if isinstance(step_result, (ParseEarlyExit, ParseFailure)):
+            return step_result
+        section_data, next_state = step_result.result
+        result.append((subcommand, section_data))
+        if next_state is None:
+            return ParseSuccess(step_state, tuple(result))
+        step_state = next_state
+
+
 # Logging
 # =======
 


...	...	@@ -17,6 +17,7 @@ from __future__ import annotations
17	17	import abc
18	18	import collections
19	19	import dataclasses
	20	+import enum
20	21	import importlib.metadata
21	22	import inspect
22	23	import logging
...	...	@@ -52,6 +53,7 @@ VERSION_OUTPUT_WRAPPING_WIDTH = 72
52	53	NOT_AN_INTEGER = "not an integer"
53	54	NOT_A_NONNEGATIVE_INTEGER = "not a non-negative integer"
54	55	NOT_A_POSITIVE_INTEGER = "not a positive integer"
	56	+COMMAND_LINE_NOT_NORMALIZED = "command-line not normalized"
55	57
56	58
57	59	# CLI parsing machinery
...	...	@@ -312,6 +314,50 @@ class CLICompositeSubcommand:
312	314	return _canon_map_cache[self]
313	315
314	316
	317	+ParsedCommandLineToken: TypeAlias = tuple[str, ...]
	318	+ParsedCommandLineSection: TypeAlias = tuple[ParsedCommandLineToken, ...]
	319	+ParsedCommandLine: TypeAlias = tuple[
	320	+ tuple[CLISubcommand, ParsedCommandLineSection], ...
	321	+]
	322	+
	323	+
	324	+class CommandLineTokenType(str, enum.Enum):
	325	+ """The type of a command-line token.
	326	+
	327	+ Attributes:
	328	+ LONG_OPTION:
	329	+ A long option, perhaps with an attached option value.
	330	+ SHORT_OPTION:
	331	+ A (perhaps trivial) cluster of short options, perhaps
	332	+ including an option value.
	333	+ POSITIONAL:
	334	+ A positional argument.
	335	+ END_OF_OPTIONS:
	336	+ The `--` pseudo-argument.
	337	+
	338	+ """
	339	+
	340	+ LONG_OPTION = "LONG_OPTION"
	341	+ """"""
	342	+ SHORT_OPTION = "SHORT_OPTION"
	343	+ """"""
	344	+ POSITIONAL = "POSITIONAL"
	345	+ """"""
	346	+ END_OF_OPTIONS = "END_OF_OPTIONS"
	347	+ """"""
	348	+
	349	+ @classmethod
	350	+ def classify(cls, token: str, /) -> CommandLineTokenType:
	351	+ """Return the type of the given token."""
	352	+ if token == "--": # noqa: S105
	353	+ return cls.END_OF_OPTIONS
	354	+ if token.startswith("--"):
	355	+ return cls.LONG_OPTION
	356	+ if token.startswith("-"):
	357	+ return cls.SHORT_OPTION
	358	+ return cls.POSITIONAL
	359	+
	360	+
315	361	class ParseState(NamedTuple):
316	362	"""The internal state of a command-line parser.
317	363
...	...	@@ -472,6 +518,212 @@ class ParseFailure(ParseResult):
472	518	# --------------
473	519
474	520
	521	+def _is_valid_option_argument(
	522	+ option: CLIOption,
	523	+ argument: str,
	524	+ /,
	525	+) -> bool:
	526	+ if isinstance(option.has_argument, bool):
	527	+ # If any argument value doesn't matter, then the argument is
	528	+ # valid if and only if the option takes an argument.
	529	+ return option.has_argument
	530	+ return argument in option.has_argument
	531	+
	532	+
	533	+def _handle_option(
	534	+ args: list[str],
	535	+ /,
	536	+ *,
	537	+ state: ParseState,
	538	+ old_command_line: collections.deque[str],
	539	+) -> ParseFailure \| str:
	540	+ canon_map = state.subcommand.canon_map
	541	+ arg_queue = list(args)
	542	+ opt_ = arg_queue.pop(0)
	543	+ if opt_ not in canon_map:
	544	+ error = f"Unknown option {opt_!r}."
	545	+ return ParseFailure(state, error)
	546	+ option = cast("CLIOption", canon_map[opt_])
	547	+ if option.has_argument:
	548	+ if not arg_queue:
	549	+ if not old_command_line:
	550	+ error = f"Missing argument for option {opt_!r}."
	551	+ return ParseFailure(state, error)
	552	+ arg_queue.append(old_command_line.popleft())
	553	+ value = arg_queue.pop(0)
	554	+ if not _is_valid_option_argument(option, value):
	555	+ error = f"Invalid argument {value!r} for option {opt_!r}."
	556	+ return ParseFailure(state, error)
	557	+ return "=".join([option.names[0], value])
	558	+ return option.names[0]
	559	+
	560	+
	561	+# TODO(the-13th-letter): Find a way to include debug logging in this
	562	+# parser step.
	563	+def normalize_options(
	564	+ state: ParseState,
	565	+ /,
	566	+) -> ParseFailure \| ParseSuccess[None]:
	567	+ old_command_line = collections.deque(state.command_line)
	568	+ new_command_line: collections.deque[str] = collections.deque()
	569	+ while old_command_line:
	570	+ arg_ = old_command_line.popleft()
	571	+ token_type = CommandLineTokenType.classify(arg_)
	572	+ # TODO(the-13th-letter): Rewrite using structural pattern matching.
	573	+ # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
	574	+ if token_type == CommandLineTokenType.LONG_OPTION:
	575	+ arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
	576	+ result = _handle_option( # may consume from old_command_line
	577	+ arg_queue,
	578	+ state=state,
	579	+ old_command_line=old_command_line,
	580	+ )
	581	+ if isinstance(result, ParseResult):
	582	+ return result
	583	+ new_command_line.append(result)
	584	+ elif token_type == CommandLineTokenType.SHORT_OPTION:
	585	+ replacement_args: collections.deque[str] = collections.deque()
	586	+ letters = collections.deque(arg_[1:])
	587	+ while letters:
	588	+ letter = letters.popleft()
	589	+ arg_queue = (
	590	+ [f"-{letter}", "".join(letters)]
	591	+ if letters
	592	+ else [f"-{letter}"]
	593	+ )
	594	+ result = _handle_option(
	595	+ arg_queue,
	596	+ state=state,
	597	+ old_command_line=old_command_line,
	598	+ )
	599	+ if isinstance(result, ParseResult):
	600	+ return result
	601	+ replacement_args.append(result)
	602	+ if "=" in result: # pragma: no cover [external]
	603	+ letters.clear()
	604	+ old_command_line.extendleft(reversed(replacement_args))
	605	+ else:
	606	+ new_command_line.extend([arg_] if arg_ == "--" else ["--", arg_])
	607	+ new_command_line.extend(old_command_line)
	608	+ old_command_line.clear()
	609	+ break
	610	+ else:
	611	+ new_command_line.append("--")
	612	+ return ParseSuccess(
	613	+ state._replace(command_line=tuple(new_command_line)), None
	614	+ )
	615	+
	616	+
	617	+def scan_for_eager_options(
	618	+ state: ParseState,
	619	+ /,
	620	+) -> ParseEarlyExit \| ParseFailure \| ParseSuccess[None]:
	621	+ option_groups = state.subcommand.contents[0]
	622	+ eager_option_objects = {
	623	+ option
	624	+ for group in option_groups
	625	+ for option in group.options
	626	+ if option.eager
	627	+ }
	628	+ eager_option_names = {
	629	+ name for option in eager_option_objects for name in option.names
	630	+ }
	631	+ command_line = collections.deque(state.command_line)
	632	+ canon_map = state.subcommand.canon_map
	633	+ while command_line:
	634	+ arg_ = command_line.popleft()
	635	+ token_type = CommandLineTokenType.classify(arg_)
	636	+ if token_type == CommandLineTokenType.LONG_OPTION:
	637	+ arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
	638	+ long_opt_name = arg_queue.pop(0)
	639	+ if long_opt_name not in canon_map:
	640	+ # Unknown option. Either the command-line is faulty,
	641	+ # or we default to the default subcommand and let it
	642	+ # make sense of this option. Regardless, no early exit
	643	+ # has been detected so far, so we are done.
	644	+ return ParseSuccess(state, None)
	645	+ option = cast("CLIOption", canon_map[long_opt_name])
	646	+ if long_opt_name in eager_option_names:
	647	+ canon_opt = "=".join([option.names[0], *arg_queue])
	648	+ new_state = state._replace(command_line=(canon_opt,))
	649	+ return ParseEarlyExit(new_state, canon_opt)
	650	+ elif token_type == CommandLineTokenType.END_OF_OPTIONS:
	651	+ return ParseSuccess(state, None)
	652	+ else: # pragma: no cover [failsafe]
	653	+ return ParseFailure(state, COMMAND_LINE_NOT_NORMALIZED)
	654	+ return ParseSuccess(state, None)
	655	+
	656	+
	657	+def parse_until_subcommand_boundary(
	658	+ state: ParseState,
	659	+ /,
	660	+) -> (
	661	+ ParseFailure
	662	+ \| ParseSuccess[tuple[ParsedCommandLineSection, ParseState \| None]]
	663	+):
	664	+ canon_map = state.subcommand.canon_map
	665	+ command_line = collections.deque(state.command_line)
	666	+ result: collections.deque[ParsedCommandLineToken] = collections.deque()
	667	+ while command_line:
	668	+ arg_ = command_line.popleft()
	669	+ token_type = CommandLineTokenType.classify(arg_)
	670	+ if token_type == CommandLineTokenType.LONG_OPTION:
	671	+ arg_queue = arg_.split("=", 1) if "=" in arg_ else [arg_]
	672	+ result.append(tuple(arg_queue))
	673	+ elif token_type == CommandLineTokenType.END_OF_OPTIONS:
	674	+ break
	675	+ else: # pragma: no cover [failsafe]
	676	+ return ParseFailure(state, COMMAND_LINE_NOT_NORMALIZED)
	677	+ if isinstance(state.subcommand, CLITerminalSubcommand):
	678	+ return ParseSuccess(state, (tuple(result), None))
	679	+ next_command_line = tuple(command_line)
	680	+ next_subcommand_name = command_line.popleft() if command_line else None
	681	+ if (
	682	+ isinstance(next_subcommand_name, str)
	683	+ and next_subcommand_name not in canon_map
	684	+ and isinstance(state.subcommand, CLICompositeSubcommand)
	685	+ and state.subcommand.contents[2] is not None
	686	+ ):
	687	+ # Unknown subcommand name; fallback exists. Run the
	688	+ # default subcommand and reinterpret the argument
	689	+ # (positional/subcommand) within the context of the default
	690	+ # subcommand.
	691	+ command_line.appendleft(next_subcommand_name)
	692	+ next_subcommand_name = None
	693	+ if next_subcommand_name not in canon_map:
	694	+ # Unknown subcommand name; no fallback exists.
	695	+ error = f"Unknown subcommand {next_subcommand_name!r}."
	696	+ return ParseFailure(state, error)
	697	+ next_subcommand = cast(
	698	+ "CLISubcommand", canon_map.get(next_subcommand_name)
	699	+ )
	700	+ new_state = ParseState(next_command_line, next_subcommand)
	701	+ return ParseSuccess(state, (tuple(result), new_state))
	702	+
	703	+
	704	+def complete(
	705	+ state: ParseState,
	706	+ /,
	707	+) -> ParseSuccess[ParsedCommandLine] \| ParseFailure \| ParseEarlyExit:
	708	+ result: collections.deque[
	709	+ tuple[CLISubcommand, ParsedCommandLineSection]
	710	+ ] = collections.deque()
	711	+ step_state = state
	712	+ while True:
	713	+ subcommand = step_state.subcommand
	714	+ tmp3_result = ParseResult.unit(step_state)
	715	+ tmp2_result = tmp3_result.bind(normalize_options)
	716	+ tmp1_result = tmp2_result.bind(scan_for_eager_options)
	717	+ step_result = tmp1_result.bind(parse_until_subcommand_boundary)
	718	+ if isinstance(step_result, (ParseEarlyExit, ParseFailure)):
	719	+ return step_result
	720	+ section_data, next_state = step_result.result
	721	+ result.append((subcommand, section_data))
	722	+ if next_state is None:
	723	+ return ParseSuccess(step_state, tuple(result))
	724	+ step_state = next_state
	725	+
	726	+
475	727	# Logging
476	728	# =======
477	729
478	730