Перейти к содержанию

flayer.main

Module source

flayer.__main__

Explicit diagnostics and opt-in owned lifecycle CLI without implicit cloud discovery.

DiagnosticArgumentParser

Bases: ArgumentParser

Suppress raw invalid argument values that may contain private endpoint material.

error 

error(message: str) -> NoReturn

Preserve argparse's exit contract while keeping untrusted argument text private.

Source code in installed/flayer/__main__.py 
28
29
30
31
def error(self, message: str) -> NoReturn:
    """Preserve argparse's exit contract while keeping untrusted argument text private."""

    self.exit(2, f"{self.prog}: error: invalid command arguments\n")

Parser 

Parser() -> ArgumentParser

Build an offline help surface and bounded command options.

Source code in installed/flayer/__main__.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def Parser() -> argparse.ArgumentParser:
    """Build an offline help surface and bounded command options."""

    parser = DiagnosticArgumentParser(prog="python -m flayer", description="F-Layer diagnostics and owned lifecycle")
    commands = parser.add_subparsers(dest="command", required=True)

    for command in ("status", "check", "health", "benchmark"):
        child = commands.add_parser(command)
        child.add_argument("--format", choices=("text", "json"), default="text")

        if command in ("health", "benchmark"):
            child.add_argument("--endpoint", help="Explicit HTTP(S) endpoint without credentials")
            child.add_argument("--timeout", type=float, default=3.0, metavar="SECONDS")

        if command == "benchmark":
            child.add_argument("--count", type=int, default=3)
            child.add_argument("--bytes", type=int, default=64 * 1024, dest="byte_limit")

    lifecycle = commands.add_parser("lifecycle", help="Explicit owned stack lifecycle")
    actions = lifecycle.add_subparsers(dest="action", required=True)

    for action in ("create", "status", "destroy", "recover"):
        child = actions.add_parser(action)
        child.add_argument("--config", required=True, help="Explicit lifecycle plan TOML path")
        child.add_argument("--state", required=True, help="Explicit owned snapshot path")
        child.add_argument("--yc-profile", required=True, help="Explicit preauthenticated yc profile")
        child.add_argument("--format", choices=("text", "json"), default="text")

        if action != "status":
            child.add_argument("--allow-mutation", action="store_true")
            child.add_argument("--scope-confirm", required=True, help="Repeat the exact folder ID")

        if action == "recover":
            child.add_argument("--rollback", action="store_true")

    return parser

Main 

Main(arguments: Sequence[str] | None = None) -> int

Run explicit diagnostics or authorized lifecycle actions and return their outcome.

Source code in installed/flayer/__main__.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def Main(arguments: Sequence[str] | None = None) -> int:
    """Run explicit diagnostics or authorized lifecycle actions and return their outcome."""

    args = Parser().parse_args(arguments)

    if args.command == "lifecycle":
        return _Lifecycle(args)

    try:
        if args.command == "status":
            checks = (*RuntimeChecks(), ProviderStatus())

        elif args.command == "check":
            checks = RuntimeChecks()

        elif not args.endpoint:
            checks = (CheckResult(
                "endpoint", DiagnosticStatus.UNSUPPORTED, "An explicit endpoint is required"
            ),)

        elif args.command == "health":
            checks = (ProbeEndpoint(args.endpoint, args.timeout),)

        else:
            limits = BenchmarkLimits(args.timeout, args.count, args.byte_limit)
            checks = RunBenchmark(args.endpoint, limits)

    except ValueError:
        report = DiagnosticReport(args.command, (CheckResult(
            "input", DiagnosticStatus.FAILED, "Invalid endpoint or diagnostic limits"
        ),))
        print(RenderJson(report) if args.format == "json" else RenderText(report))

        return 2

    report = DiagnosticReport(args.command, checks)
    print(RenderJson(report) if args.format == "json" else RenderText(report))

    return report.ExitCode()