ref: require stronger types for sentry.runner.commands.* (#68130)

<!-- Describe your PR here. -->

Author: asottile-sentry

pyproject.toml

@@ -631,7 +631,7 @@ module = [
     "sentry.nodestore.models",
     "sentry.relay.config.metric_extraction",
     "sentry.reprocessing2",
-    "sentry.runner.commands.backup",
+    "sentry.runner.commands.*",
     "sentry.snuba.metrics.extraction",
     "sentry.tasks.on_demand_metrics",
     "sentry.tasks.reprocessing2",

src/sentry/runner/commands/config.py

@@ -4,14 +4,14 @@
 
 
 @click.group()
-def config():
+def config() -> None:
     "Manage runtime config options."
 
 
 @config.command()
 @click.argument("pattern", default="*", required=False)
 @configuration
-def list(pattern):
+def list(pattern: str) -> None:
     "List configuration options."
     from fnmatch import fnmatch
 
@@ -26,7 +26,7 @@ def list(pattern):
 @click.option("--silent", "-q", default=False, is_flag=True, help="Suppress extraneous output.")
 @click.argument("option")
 @configuration
-def get(option, silent):
+def get(option: str, silent: bool) -> None:
     "Get a configuration option."
     from django.conf import settings
 
@@ -58,7 +58,7 @@ def get(option, silent):
 @click.argument("key")
 @click.argument("value", required=False)
 @configuration
-def set(key, value, secret):
+def set(key: str, value: str | None, secret: bool) -> None:
     "Set a configuration option to a new value."
     from sentry import options
     from sentry.options import UpdateChannel
@@ -82,7 +82,7 @@ def set(key, value, secret):
 @click.option("--no-input", default=False, is_flag=True, help="Do not show confirmation.")
 @click.argument("option")
 @configuration
-def delete(option, no_input):
+def delete(option: str, no_input: bool) -> None:
     "Delete/unset a configuration option."
     from sentry import options
     from sentry.options.manager import UnknownOption
@@ -158,15 +158,15 @@ def dump(flags: int, only_set: bool, pretty_print: bool) -> None:
 
 
 @config.command(name="generate-secret-key")
-def generate_secret_key():
+def generate_secret_key() -> None:
     "Generate a new cryptographically secure secret key value."
     from sentry.runner.settings import generate_secret_key
 
     click.echo(generate_secret_key())
 
 
 @config.command()
-def discover():
+def discover() -> None:
     "Print paths to config files."
     from sentry.runner.settings import discover_configs
 

src/sentry/runner/commands/configoptions.py

@@ -80,7 +80,7 @@ def _attempt_update(
 @log_options()
 @click.pass_context
 @configuration
-def configoptions(ctx, dry_run: bool, file: str | None, hide_drift: bool) -> None:
+def configoptions(ctx: click.Context, dry_run: bool, file: str | None, hide_drift: bool) -> None:
     """
     Makes changes to options in bulk starting from a yaml file.
     Contrarily to the `config` command, this is meant to perform
@@ -155,7 +155,7 @@ def configoptions(ctx, dry_run: bool, file: str | None, hide_drift: bool) -> Non
 @configoptions.command()
 @click.pass_context
 @configuration
-def patch(ctx) -> None:
+def patch(ctx: click.Context) -> None:
     """
     Applies to the DB the option values found in the config file.
     Only the options present in the file are updated. No deletions
@@ -217,7 +217,7 @@ def patch(ctx) -> None:
 @configoptions.command()
 @click.pass_context
 @configuration
-def sync(ctx):
+def sync(ctx: click.Context) -> None:
     """
     Synchronizes the content of the file with the DB. The source of
     truth is the config file, not the DB. If an option is missing in

src/sentry/runner/commands/createuser.py

@@ -1,15 +1,28 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
 import click
 
 from sentry.runner.decorators import configuration
 
+if TYPE_CHECKING:
+    from django.db.models.fields import Field
+
+    from sentry.models.user import User
+
+
+def _get_field(field_name: str) -> Field[str, str]:
+    from django.db.models.fields import Field
 
-def _get_field(field_name):
     from sentry.models.user import User
 
-    return User._meta.get_field(field_name)
+    ret = User._meta.get_field(field_name)
+    assert isinstance(ret, Field), ret
+    return ret
 
 
-def _get_email():
+def _get_email() -> list[str]:
     from django.core.exceptions import ValidationError
 
     rv = click.prompt("Email")
@@ -20,7 +33,7 @@ def _get_email():
         raise click.ClickException("; ".join(e.messages))
 
 
-def _get_password():
+def _get_password() -> str:
     from django.core.exceptions import ValidationError
 
     rv = click.prompt("Password", hide_input=True, confirmation_prompt=True)
@@ -31,11 +44,11 @@ def _get_password():
         raise click.ClickException("; ".join(e.messages))
 
 
-def _get_superuser():
+def _get_superuser() -> bool:
     return click.confirm("Should this user be a superuser?", default=False)
 
 
-def _set_superadmin(user):
+def _set_superadmin(user: User) -> None:
     """
     superadmin role approximates superuser (model attribute) but leveraging
     Sentry's role system.
@@ -78,7 +91,16 @@ def _set_superadmin(user):
     "--force-update", default=False, is_flag=True, help="If true, will update existing users."
 )
 @configuration
-def createuser(emails, org_id, password, superuser, staff, no_password, no_input, force_update):
+def createuser(
+    emails: list[str] | None,
+    org_id: str | None,
+    password: str | None,
+    superuser: bool | None,
+    staff: bool | None,
+    no_password: bool,
+    no_input: bool,
+    force_update: bool,
+) -> None:
     "Create a new user."
 
     from django.conf import settings

src/sentry/runner/commands/devservices.py

@@ -166,7 +166,7 @@ def ensure_interface(ports: dict[str, int | tuple[str, int]]) -> dict[str, tuple
     return rv
 
 
-def ensure_docker_cli_context(context: str):
+def ensure_docker_cli_context(context: str) -> None:
     # this is faster than running docker context use ...
     config_file = os.path.expanduser("~/.docker/config.json")
     config = {}

src/sentry/runner/commands/django.py

@@ -7,8 +7,8 @@
 @click.argument("management_args", nargs=-1, type=click.UNPROCESSED)
 @configuration
 @click.pass_context
-def django(ctx, management_args):
+def django(ctx: click.Context, management_args: tuple[str, ...]) -> None:
     "Execute Django subcommands."
     from django.core.management import execute_from_command_line
 
-    execute_from_command_line(argv=[ctx.command_path] + list(management_args))
+    execute_from_command_line(argv=[ctx.command_path, *management_args])

src/sentry/runner/commands/exec.py

@@ -20,7 +20,7 @@
 )
 @click.option("-c", default="", help="Read script from string.")
 @click.argument("file", default=None, required=False)
-def exec_(c, file):
+def exec_(c: str, file: str | None) -> None:
     """
     Execute a script.
 

src/sentry/runner/commands/execfile.py

@@ -8,7 +8,7 @@
     name="execfile", context_settings=dict(ignore_unknown_options=True, allow_extra_args=True)
 )
 @click.argument("filename", required=True)
-def execfile(filename):
+def execfile(filename: str) -> None:
     """Execute a script.
 
     This is very similar to `exec`, with the following differences:
@@ -27,7 +27,7 @@ def execfile(filename):
 
     - __file__ is set to the filename of the script.
     """
-    filename = pathlib.Path(filename)
+    filename_src = pathlib.Path(filename).read_text()
     preamble = "\n".join(
         [
             "from sentry.runner import configure; configure()",
@@ -39,5 +39,5 @@ def execfile(filename):
     preamble_code = compile(preamble, filename, "exec")
     exec(preamble_code, script_globals, script_globals)
     sys.argv = sys.argv[1:]
-    script_code = compile(filename.read_text(), filename, "exec")
+    script_code = compile(filename_src, filename, "exec")
     exec(script_code, script_globals, script_globals)

src/sentry/runner/commands/files.py

@@ -6,14 +6,14 @@
 
 
 @click.group()
-def files():
+def files() -> None:
     """Manage files from filestore."""
 
 
 @files.command()
 @click.argument("id", type=click.INT, metavar="FILE_ID")
 @configuration
-def get(id):
+def get(id: int) -> None:
     """Fetch a file's contents by id."""
     from sentry.models.files.file import File
 
@@ -33,7 +33,7 @@ def get(id):
 @click.argument("id", type=click.INT, metavar="FILE_ID")
 @click.option("--format", default="json", type=click.Choice(("json", "yaml")))
 @configuration
-def info(id, format):
+def info(id: int, format: str) -> None:
     """Show a file's metadata by id."""
     from sentry.models.files.file import File
 

src/sentry/runner/commands/help.py

@@ -3,6 +3,7 @@
 
 @click.command()
 @click.pass_context
-def help(ctx):
+def help(ctx: click.Context) -> None:
     "Show this message and exit."
+    assert ctx.parent is not None
     click.echo(ctx.parent.get_help())

src/sentry/runner/commands/init.py

@@ -11,8 +11,7 @@
     "--no-clobber", default=False, is_flag=True, help="Don't ask to overwrite existing config."
 )
 @click.argument("directory", required=False)
-@click.pass_context
-def init(ctx, dev, no_clobber, directory):
+def init(dev: bool, no_clobber: bool, directory: str) -> None:
     "Initialize new configuration directory."
     from sentry.runner.settings import discover_configs, generate_settings
 

src/sentry/runner/commands/killswitches.py

@@ -1,5 +1,6 @@
 import itertools
 import textwrap
+from typing import IO, Any
 
 import click
 import yaml
@@ -8,11 +9,11 @@
 
 
 @click.group()
-def killswitches():
+def killswitches() -> None:
     "Manage killswitches for ingestion pipeline."
 
 
-def _get_edit_template(killswitch_name: str, option_value) -> str:
+def _get_edit_template(killswitch_name: str, option_value: Any) -> str:
     from sentry import killswitches
 
     comments = [
@@ -46,7 +47,7 @@ def _get_edit_template(killswitch_name: str, option_value) -> str:
 @click.argument("killswitch_name", required=True)
 @click.argument("outfile", type=click.File("w"), required=True)
 @configuration
-def _pull(killswitch_name, outfile):
+def _pull(killswitch_name: str, outfile: IO[str]) -> None:
     """
     Save the current state of the given killswitch in a file.
 
@@ -71,7 +72,7 @@ def _pull(killswitch_name, outfile):
 @click.argument("infile", type=click.File("r"), required=True)
 @click.option("--yes", is_flag=True, help="skip confirmation prompts, very dangerous")
 @configuration
-def _push(killswitch_name, infile, yes):
+def _push(killswitch_name: str, infile: IO[str], yes: bool) -> None:
     """
     Write back a killswitch into the DB.
 
@@ -118,7 +119,7 @@ def _push(killswitch_name, infile, yes):
 
 @killswitches.command("list")
 @configuration
-def _list():
+def _list() -> None:
     """
     List all killswitches and whether they are enabled (and how).
     """

src/sentry/runner/commands/migrations.py

@@ -8,7 +8,7 @@
 
 @click.group()
 @configuration
-def migrations():
+def migrations() -> None:
     from sentry.runner.initializer import monkeypatch_django_migrations
 
     # Include our monkeypatches for migrations.
@@ -21,11 +21,8 @@ def migrations():
 @migrations.command()
 @click.argument("app_name")
 @click.argument("migration_name")
-@click.pass_context
-def run(ctx, app_name: str, migration_name: str) -> None:
+def run(app_name: str, migration_name: str) -> None:
     "Manually run a single data migration. Will error if migration is not post-deploy/dangerous"
-    del ctx  # assertion: unused argument
-
     for connection_name in settings.DATABASES.keys():
         if settings.DATABASES[connection_name].get("REPLICA_OF", False):
             continue

src/sentry/runner/commands/openai.py

@@ -1,5 +1,6 @@
 # The sentry utils json cannot pretty print
-import json  # noqa
+import json  # noqa: S003
+from typing import IO
 
 import click
 
@@ -8,7 +9,7 @@
 @click.option("--event", type=click.File("r"))
 @click.option("--model", default="gpt-3.5-turbo")
 @click.option("--dump-prompt", is_flag=True)
-def openai(event, model, dump_prompt):
+def openai(event: IO[str], model: str, dump_prompt: bool) -> None:
     """
     Runs the OpenAI assistent against a JSON event payload.
     """

src/sentry/runner/commands/performance.py

@@ -23,7 +23,7 @@ def performance() -> None:
 )
 @click.option("-v", "--verbose", count=True)
 @configuration
-def detect(filename, detector_class, verbose):
+def detect(filename: str, detector_class: str | None, verbose: int) -> None:
     """
     Runs performance problem detection on event data in the supplied filename
     using default detector settings with every detector. Filename should be a
@@ -80,7 +80,7 @@ def detect(filename, detector_class, verbose):
     "-n", required=False, type=int, default=1000, help="Number of times to run detection."
 )
 @configuration
-def timeit(filename, detector_class, n):
+def timeit(filename: str, detector_class: str, n: int) -> None:
     """
     Runs timing on performance problem detection on event data in the supplied
     filename and report results.
@@ -99,7 +99,7 @@ def timeit(filename, detector_class, n):
 
     detector = performance_detection.__dict__[detector_class](settings, data)
 
-    def detect():
+    def detect() -> None:
         performance_detection.run_detector_on_data(detector, data)
 
     result = timeit.timeit(stmt=detect, number=n)