Send a request to the daemon. Return the JSON dict with the response. Raise BadStatus if there is something wrong with the status file or if the process whose pid is in the status file has died. Return {'error': } if an IPC operation or receive() raised OSError. This
(
status_file: str, command: str, *, timeout: int | None = None, **kwds: object
)
| 640 | |
| 641 | |
| 642 | def request( |
| 643 | status_file: str, command: str, *, timeout: int | None = None, **kwds: object |
| 644 | ) -> dict[str, Any]: |
| 645 | """Send a request to the daemon. |
| 646 | |
| 647 | Return the JSON dict with the response. |
| 648 | |
| 649 | Raise BadStatus if there is something wrong with the status file |
| 650 | or if the process whose pid is in the status file has died. |
| 651 | |
| 652 | Return {'error': <message>} if an IPC operation or receive() |
| 653 | raised OSError. This covers cases such as connection refused or |
| 654 | closed prematurely as well as invalid JSON received. |
| 655 | """ |
| 656 | response: dict[str, str] = {} |
| 657 | args = dict(kwds) |
| 658 | args["command"] = command |
| 659 | # Tell the server whether this request was initiated from a human-facing terminal, |
| 660 | # so that it can format the type checking output accordingly. |
| 661 | args["is_tty"] = sys.stdout.isatty() or should_force_color() |
| 662 | args["terminal_width"] = get_terminal_width() |
| 663 | _, name = get_status(status_file) |
| 664 | try: |
| 665 | with IPCClient(name, timeout) as client: |
| 666 | send(client, args) |
| 667 | |
| 668 | final = False |
| 669 | while not final: |
| 670 | response = receive(client) |
| 671 | final = bool(response.pop("final", False)) |
| 672 | # Display debugging output written to stdout/stderr in the server process for convenience. |
| 673 | # This should not be confused with "out" and "err" fields in the response. |
| 674 | # Those fields hold the output of the "check" command, and are handled in check_output(). |
| 675 | stdout = response.pop("stdout", None) |
| 676 | if stdout: |
| 677 | sys.stdout.write(stdout) |
| 678 | stderr = response.pop("stderr", None) |
| 679 | if stderr: |
| 680 | sys.stderr.write(stderr) |
| 681 | except (OSError, IPCException) as err: |
| 682 | return {"error": str(err)} |
| 683 | # TODO: Other errors, e.g. ValueError, UnicodeError |
| 684 | |
| 685 | return response |
| 686 | |
| 687 | |
| 688 | def get_status(status_file: str) -> tuple[int, str]: |
no test coverage detected
searching dependent graphs…