Fix mypy configuration & upgrade typing

.git-blame-ignore-revs

@@ -5,3 +5,5 @@
 6ec972f4dbb880bf0c7a11809e6c1ba194c9784c
 # upgrade code style to use f-strings using flynt
 313b80f27f525441c449593a3aeaf38389f63c13
+# upgrade typing annotations using fix-future-annotations
+b5324820b299b1fe7da0608f0cc8ec47f58b1e40

locust/argument_parser.py

@@ -1,8 +1,10 @@
+from __future__ import annotations
+
 import os
 import platform
 import sys
 import textwrap
-from typing import Dict, List, NamedTuple, Optional, Any
+from typing import NamedTuple, Any
 import configargparse
 
 import locust
@@ -38,11 +40,11 @@ def add_argument(self, *args, **kwargs) -> configargparse.Action:
         return action
 
     @property
-    def args_included_in_web_ui(self) -> Dict[str, configargparse.Action]:
+    def args_included_in_web_ui(self) -> dict[str, configargparse.Action]:
         return {a.dest: a for a in self._actions if hasattr(a, "include_in_web_ui") and a.include_in_web_ui}
 
     @property
-    def secret_args_included_in_web_ui(self) -> Dict[str, configargparse.Action]:
+    def secret_args_included_in_web_ui(self) -> dict[str, configargparse.Action]:
         return {
             a.dest: a
             for a in self._actions
@@ -91,7 +93,7 @@ def find_locustfile(locustfile):
     # Implicit 'return None' if nothing was found
 
 
-def find_locustfiles(locustfiles: List[str], is_directory: bool) -> List[str]:
+def find_locustfiles(locustfiles: list[str], is_directory: bool) -> list[str]:
     """
     Returns a list of relative file paths for the Locustfile Picker. If is_directory is True,
     locustfiles is expected to have a single index which is a directory that will be searched for
@@ -176,7 +178,7 @@ def get_empty_argument_parser(add_help=True, default_config_files=DEFAULT_CONFIG
     return parser
 
 
-def parse_locustfile_option(args=None) -> List[str]:
+def parse_locustfile_option(args=None) -> list[str]:
     """
     Construct a command line parser that is only used to parse the -f argument so that we can
     import the test scripts in case any of them adds additional command line arguments to the
@@ -667,10 +669,10 @@ class UIExtraArgOptions(NamedTuple):
     default_value: str
     is_secret: bool
     help_text: str
-    choices: Optional[List[str]] = None
+    choices: list[str] | None = None
 
 
-def ui_extra_args_dict(args=None) -> Dict[str, Dict[str, Any]]:
+def ui_extra_args_dict(args=None) -> dict[str, dict[str, Any]]:
     """Get all the UI visible arguments"""
     locust_args = default_args_dict()
 
@@ -691,7 +693,7 @@ def ui_extra_args_dict(args=None) -> Dict[str, Dict[str, Any]]:
     return extra_args
 
 
-def locustfile_is_directory(locustfiles: List[str]) -> bool:
+def locustfile_is_directory(locustfiles: list[str]) -> bool:
     """
     If a user passes in a locustfile without a file extension and there is a directory with the same name,
     this function defaults to using the file and will raise a warning.

locust/clients.py

@@ -2,7 +2,7 @@
 import re
 import time
 from contextlib import contextmanager
-from typing import Generator, Optional
+from typing import Generator
 from urllib.parse import urlparse, urlunparse
 
 import requests
@@ -48,7 +48,7 @@ class HttpSession(requests.Session):
                            and then mark it as successful even if the response code was not (i.e 500 or 404).
     """
 
-    def __init__(self, base_url, request_event, user, *args, pool_manager: Optional[PoolManager] = None, **kwargs):
+    def __init__(self, base_url, request_event, user, *args, pool_manager: PoolManager | None = None, **kwargs):
         super().__init__(*args, **kwargs)
 
         self.base_url = base_url
@@ -57,7 +57,7 @@ def __init__(self, base_url, request_event, user, *args, pool_manager: Optional[
 
         # User can group name, or use the group context manager to gather performance statistics under a specific name
         # This is an alternative to passing in the "name" parameter to the requests function
-        self.request_name: Optional[str] = None
+        self.request_name: str | None = None
 
         # Check for basic authentication
         parsed_url = urlparse(self.base_url)
@@ -301,7 +301,7 @@ def failure(self, exc):
 
 
 class LocustHttpAdapter(HTTPAdapter):
-    def __init__(self, pool_manager: Optional[PoolManager], *args, **kwargs):
+    def __init__(self, pool_manager: PoolManager | None, *args, **kwargs):
         self.poolmanager = pool_manager
         super().__init__(*args, **kwargs)
 
@@ -323,5 +323,5 @@ def _failure(self):
     )
 
 
-Response.success = _success
-Response.failure = _failure
+Response.success = _success  # type: ignore[attr-defined]
+Response.failure = _failure  # type: ignore[attr-defined]

locust/contrib/fasthttp.py

@@ -10,7 +10,7 @@
 from ssl import SSLError
 import time
 import traceback
-from typing import Callable, Optional, Tuple, Dict, Any, Generator, cast
+from typing import Callable, Any, Generator, cast
 
 from http.cookiejar import CookieJar
 
@@ -79,10 +79,10 @@ def __init__(
         self,
         environment: Environment,
         base_url: str,
-        user: Optional[User],
+        user: User | None,
         insecure=True,
-        client_pool: Optional[HTTPClientPool] = None,
-        ssl_context_factory: Optional[Callable] = None,
+        client_pool: HTTPClientPool | None = None,
+        ssl_context_factory: Callable | None = None,
         **kwargs,
     ):
         self.environment = environment
@@ -313,18 +313,18 @@ class FastHttpUser(User):
     insecure: bool = True
     """Parameter passed to FastHttpSession. Default True, meaning no SSL verification."""
 
-    default_headers: Optional[dict] = None
+    default_headers: dict | None = None
     """Parameter passed to FastHttpSession. Adds the listed headers to every request."""
 
     concurrency: int = 10
     """Parameter passed to FastHttpSession. Describes number of concurrent requests allowed by the FastHttpSession. Default 10.
     Note that setting this value has no effect when custom client_pool was given, and you need to spawn a your own gevent pool
     to use it (as Users only have one greenlet). See test_fasthttp.py / test_client_pool_concurrency for an example."""
 
-    client_pool: Optional[HTTPClientPool] = None
+    client_pool: HTTPClientPool | None = None
     """HTTP client pool to use. If not given, a new pool is created per single user."""
 
-    ssl_context_factory: Optional[Callable] = None
+    ssl_context_factory: Callable | None = None
     """A callable that return a SSLContext for overriding the default context created by the FastHttpSession."""
 
     abstract = True
@@ -360,7 +360,7 @@ def __init__(self, environment):
 
     @contextmanager
     def rest(
-        self, method, url, headers: Optional[dict] = None, **kwargs
+        self, method, url, headers: dict | None = None, **kwargs
     ) -> Generator[RestResponseContextManager, None, None]:
         """
         A wrapper for self.client.request that:
@@ -423,36 +423,36 @@ def rest_(self, method, url, name=None, **kwargs) -> Generator[RestResponseConte
 
 
 class FastRequest(CompatRequest):
-    payload: Optional[str] = None
+    payload: str | None = None
 
     @property
-    def body(self) -> Optional[str]:
+    def body(self) -> str | None:
         return self.payload
 
 
 class FastResponse(CompatResponse):
-    headers: Optional[Headers] = None
+    headers: Headers | None = None
     """Dict like object containing the response headers"""
 
-    _response: Optional[HTTPSocketPoolResponse] = None
+    _response: HTTPSocketPoolResponse | None = None
 
-    encoding: Optional[str] = None
+    encoding: str | None = None
     """In some cases setting the encoding explicitly is needed. If so, do it before calling .text"""
 
-    request: Optional[FastRequest] = None
+    request: FastRequest | None = None
 
     def __init__(
         self,
         ghc_response: HTTPSocketPoolResponse,
-        request: Optional[FastRequest] = None,
-        sent_request: Optional[str] = None,
+        request: FastRequest | None = None,
+        sent_request: str | None = None,
     ):
         super().__init__(ghc_response, request, sent_request)
 
         self.request = request
 
     @property
-    def text(self) -> Optional[str]:
+    def text(self) -> str | None:
         """
         Returns the text content of the response as a decoded string
         """
@@ -472,7 +472,7 @@ def text(self) -> Optional[str]:
         return str(self.content, str(self.encoding), errors="replace")
 
     @property
-    def url(self) -> Optional[str]:
+    def url(self) -> str | None:
         """
         Get "response" URL, which is the same as the request URL. This is a small deviation from HttpSession, which gets the final (possibly redirected) URL.
         """
@@ -527,11 +527,11 @@ class ErrorResponse:
     that doesn't have a real Response object attached. E.g. a socket error or similar
     """
 
-    headers: Optional[Headers] = None
+    headers: Headers | None = None
     content = None
     status_code = 0
-    error: Optional[Exception] = None
-    text: Optional[str] = None
+    error: Exception | None = None
+    text: str | None = None
     request: CompatRequest
 
     def __init__(self, url: str, request: CompatRequest):
@@ -547,7 +547,7 @@ class LocustUserAgent(UserAgent):
     request_type = FastRequest
     valid_response_codes = frozenset([200, 201, 202, 203, 204, 205, 206, 207, 208, 226, 301, 302, 303, 304, 307])
 
-    def __init__(self, client_pool: Optional[HTTPClientPool] = None, **kwargs):
+    def __init__(self, client_pool: HTTPClientPool | None = None, **kwargs):
         super().__init__(**kwargs)
 
         if client_pool is not None:

locust/debug.py

@@ -1,10 +1,11 @@
+from __future__ import annotations
+
 from datetime import datetime, timezone
 import os
 import inspect
 import locust
 import locust.log
 from locust import User, argument_parser
-from typing import Type, Optional
 from locust.env import Environment
 from locust.exception import CatchResponseError, RescheduleTask
 
@@ -94,16 +95,16 @@ def on_request(
         print()
 
 
-_env: Optional[Environment] = None  # minimal Environment for debugging
+_env: Environment | None = None  # minimal Environment for debugging
 
 
 def run_single_user(
-    user_class: Type[User],
+    user_class: type[User],
     include_length=False,
     include_time=False,
     include_context=False,
     include_payload=False,
-    loglevel: Optional[str] = "WARNING",
+    loglevel: str | None = "WARNING",
 ):
     """
     Runs a single User. Useful when you want to run a debugger.