mypy.dmypy.client

module documentation

(source)

Client for mypy daemon mode.

This manages a daemon process which keeps useful state in memory rather than having to read it back from disk on each run.

Class	`AugmentedHelpFormatter`	Undocumented
Class	`BadStatus`	Exception raised when there is something wrong with the status file.
Function	`action`	Decorator to tie an action function to a subparser.
Function	`check_output`	Print the output from a check or recheck command.
Function	`check_status`	Check if the process is alive.
Function	`console_entry`	Undocumented
Function	`do_check`	Ask the daemon to check a list of files.
Function	`do_daemon`	Serve requests in the foreground.
Function	`do_hang`	Hang for 100 seconds, as a debug hack.
Function	`do_help`	Print full help (same as dmypy --help).
Function	`do_kill`	Kill daemon process with SIGKILL.
Function	`do_recheck`	Ask the daemon to recheck the previous list of files, with optional modifications.
Function	`do_restart`	Restart daemon (it may or may not be running; but not hanging).
Function	`do_run`	Do a check, starting (or restarting) the daemon as necessary
Function	`do_start`	Start daemon (it must not already be running).
Function	`do_status`	Print daemon status.
Function	`do_stop`	Stop daemon via a 'stop' request.
Function	`do_suggest`	Ask the daemon for a suggested signature.
Function	`fail`	Undocumented
Function	`get_status`	Read status file and check if the process is alive.
Function	`is_running`	Check if the server is running cleanly
Function	`main`	The code is top-down.
Function	`read_status`	Read status file.
Function	`request`	Send a request to the daemon.
Function	`restart_server`	Restart daemon (it may or may not be running; but not hanging).
Function	`show_stats`	Undocumented
Function	`start_server`	Start the server from command arguments and wait for it.
Function	`wait_for_server`	Wait until the server is up.
Variable	`ActionFunction`	Undocumented
Variable	`check_parser`	Undocumented
Variable	`daemon_parser`	Undocumented
Variable	`hang_parser`	Undocumented
Variable	`help_parser`	Undocumented
Variable	`kill_parser`	Undocumented
Variable	`p`	Undocumented
Variable	`parser`	Undocumented
Variable	`recheck_parser`	Undocumented
Variable	`restart_parser`	Undocumented
Variable	`run_parser`	Undocumented
Variable	`start_parser`	Undocumented
Variable	`status_parser`	Undocumented
Variable	`stop_parser`	Undocumented
Variable	`subparsers`	Undocumented
Variable	`suggest_parser`	Undocumented

def action(subparser): (source)

Decorator to tie an action function to a subparser.

Parameters
subparser:`argparse.ArgumentParser`	Undocumented
Returns
`Callable[[ActionFunction], ActionFunction]`	Undocumented

def check_output(response, verbose, junit_xml, perf_stats_file): (source)

Print the output from a check or recheck command.

Call sys.exit() unless the status code is zero.

Parameters
response:`Dict[str, Any]`	Undocumented
verbose:`bool`	Undocumented
junit_xml:`Optional[str]`	Undocumented
perf_stats_file:`Optional[str]`	Undocumented

def check_status(data): (source)

Check if the process is alive.

Return (pid, connection_name) on success.

Raise BadStatus if something's wrong.

Parameters
data:`Dict[str, Any]`	Undocumented
Returns
`Tuple[int, str]`	Undocumented

def console_entry(): (source)

Undocumented

@action(check_parser)
def do_check(args): (source)

Ask the daemon to check a list of files.

Parameters
args:`argparse.Namespace`	Undocumented

@action(daemon_parser)
def do_daemon(args): (source)

Serve requests in the foreground.

Parameters
args:`argparse.Namespace`	Undocumented

@action(hang_parser)
def do_hang(args): (source)

Hang for 100 seconds, as a debug hack.

Parameters
args:`argparse.Namespace`	Undocumented

@action(help_parser)
def do_help(args): (source)

Print full help (same as dmypy --help).

Parameters
args:`argparse.Namespace`	Undocumented

@action(kill_parser)
def do_kill(args): (source)

Kill daemon process with SIGKILL.

Parameters
args:`argparse.Namespace`	Undocumented

@action(recheck_parser)
def do_recheck(args): (source)

Ask the daemon to recheck the previous list of files, with optional modifications.

If at least one of --remove or --update is given, the server will update the list of files to check accordingly and assume that any other files are unchanged. If none of these flags are given, the server will call stat() on each file last checked to determine its status.

Files given in --update ought to exist. Files given in --remove need not exist; if they don't they will be ignored. The lists may be empty but oughtn't contain duplicates or overlap.

NOTE: The list of files is lost when the daemon is restarted.

Parameters
args:`argparse.Namespace`	Undocumented

@action(restart_parser)
def do_restart(args): (source)

Restart daemon (it may or may not be running; but not hanging).

We first try to stop it politely if it's running. This also sets mypy flags from the command line (see do_start()).

Parameters
args:`argparse.Namespace`	Undocumented

@action(run_parser)
def do_run(args): (source)

Do a check, starting (or restarting) the daemon as necessary

Restarts the daemon if the running daemon reports that it is required (due to a configuration change, for example).

Setting flags is a bit awkward; you have to use e.g.:

dmypy run -- --strict a.py b.py ...

since we don't want to duplicate mypy's huge list of flags. (The -- is only necessary if flags are specified.)

Parameters
args:`argparse.Namespace`	Undocumented

@action(start_parser)
def do_start(args): (source)

Start daemon (it must not already be running).

This is where mypy flags are set from the command line.

Setting flags is a bit awkward; you have to use e.g.:

dmypy start -- --strict

since we don't want to duplicate mypy's huge list of flags.

Parameters
args:`argparse.Namespace`	Undocumented

@action(status_parser)
def do_status(args): (source)

Print daemon status.

This verifies that it is responsive to requests.

Parameters
args:`argparse.Namespace`	Undocumented

@action(stop_parser)
def do_stop(args): (source)

Stop daemon via a 'stop' request.

Parameters
args:`argparse.Namespace`	Undocumented

@action(suggest_parser)
def do_suggest(args): (source)

Ask the daemon for a suggested signature.

This just prints whatever the daemon reports as output. For now it may be closer to a list of call sites.

Parameters
args:`argparse.Namespace`	Undocumented

def fail(msg): (source)

Undocumented

Parameters
msg:`str`	Undocumented

def get_status(status_file): (source)

Read status file and check if the process is alive.

Return (pid, connection_name) on success.

Raise BadStatus if something's wrong.

Parameters
status_file:`str`	Undocumented
Returns
`Tuple[int, str]`	Undocumented

def is_running(status_file): (source)

Check if the server is running cleanly

Parameters
status_file:`str`	Undocumented
Returns
`bool`	Undocumented

def main(argv): (source)

The code is top-down.

Parameters
argv:`List[str]`	Undocumented

def read_status(status_file): (source)

Read status file.

Raise BadStatus if the status file doesn't exist or contains invalid JSON or the JSON is not a dict.

Parameters
status_file:`str`	Undocumented
Returns
`Dict[str, object]`	Undocumented

def request(status_file, command, *, timeout=None, **kwds): (source)

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': <message>} if an IPC operation or receive() raised OSError. This covers cases such as connection refused or closed prematurely as well as invalid JSON received.

Parameters
status_file:`str`	Undocumented
command:`str`	Undocumented
timeout:`Optional[int]`	Undocumented
**kwds:`object`	Undocumented
Returns
`Dict[str, Any]`	Undocumented

def restart_server(args, allow_sources=False): (source)

Restart daemon (it may or may not be running; but not hanging).

Parameters
args:`argparse.Namespace`	Undocumented
allow_sources:`bool`	Undocumented

def show_stats(response): (source)

Undocumented

Parameters
response:`Mapping[str, object]`	Undocumented

def start_server(args, allow_sources=False): (source)

Start the server from command arguments and wait for it.

Parameters
args:`argparse.Namespace`	Undocumented
allow_sources:`bool`	Undocumented

def wait_for_server(status_file, timeout=5.0): (source)

Wait until the server is up.

Exit if it doesn't happen within the timeout.

Parameters
status_file:`str`	Undocumented
timeout:`float`	Undocumented

ActionFunction = (source)

Undocumented

check_parser = (source)

Undocumented

daemon_parser = (source)

Undocumented

hang_parser = (source)

Undocumented

help_parser = (source)

Undocumented

kill_parser = (source)

Undocumented

p = (source)

Undocumented

parser = (source)

Undocumented

recheck_parser = (source)

Undocumented

restart_parser = (source)

Undocumented

run_parser = (source)

Undocumented

start_parser = (source)

Undocumented

status_parser = (source)

Undocumented

stop_parser = (source)

Undocumented

subparsers = (source)

Undocumented

suggest_parser = (source)

Undocumented