Revert #2205. Getting rid of the ugly Union types of pre 3.10 Python …

…was nice, but it is just too weird to have code that cant be validated with mypy on all supported Pythons.

locust/clients.py

@@ -2,7 +2,7 @@
 import re
 import time
 from contextlib import contextmanager
-from typing import Generator, Optional
+from typing import Generator, Optional, Union
 from urllib.parse import urlparse, urlunparse
 
 import requests
@@ -198,7 +198,7 @@ class ResponseContextManager(LocustResponse):
     :py:meth:`failure <locust.clients.ResponseContextManager.failure>`.
     """
 
-    _manual_result: bool | Exception | None = None
+    _manual_result: Optional[Union[bool, Exception]] = None
     _entered = False
 
     def __init__(self, response, request_event, request_meta):

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, Optional, Tuple, Dict, Any, Generator, cast, Union
 
 from http.cookiejar import CookieJar
 
@@ -149,17 +149,17 @@ def request(
         self,
         method: str,
         url: str,
-        name: str | None = None,
-        data: str | dict | None = None,
+        name: str = None,
+        data: Union[str, dict] = None,
         catch_response: bool = False,
         stream: bool = False,
-        headers: dict | None = None,
+        headers: dict = None,
         auth=None,
-        json: dict | None = None,
+        json: dict = None,
         allow_redirects=True,
         context: dict = {},
         **kwargs,
-    ) -> ResponseContextManager | FastResponse:
+    ) -> Union[ResponseContextManager, FastResponse]:
         """
         Send and HTTP request
         Returns :py:class:`locust.contrib.fasthttp.FastResponse` object.

locust/env.py

@@ -6,6 +6,7 @@
     Type,
     TypeVar,
     Optional,
+    Union,
 )
 
 from configargparse import Namespace
@@ -244,7 +245,7 @@ def assign_equal_weights(self) -> None:
         """
         for u in self.user_classes:
             u.weight = 1
-            user_tasks: List[TaskSet | Callable] = []
+            user_tasks: List[Union[TaskSet, Callable]] = []
             tasks_frontier = u.tasks
             while len(tasks_frontier) != 0:
                 t = tasks_frontier.pop()

locust/runners.py

@@ -30,6 +30,7 @@
     cast,
     Callable,
     TypedDict,
+    Union,
 )
 from uuid import uuid4
 import gevent
@@ -108,7 +109,7 @@ def __init__(self, environment: "Environment") -> None:
         self.state = STATE_INIT
         self.spawning_greenlet: Optional[gevent.Greenlet] = None
         self.shape_greenlet: Optional[gevent.Greenlet] = None
-        self.shape_last_tick: Tuple[int, float] | Tuple[int, float, Optional[List[Type[User]]]] | None = None
+        self.shape_last_tick: Union[Tuple[int, float], Tuple[int, float, Optional[List[Type[User]]]], None] = None
         self.current_cpu_usage: int = 0
         self.cpu_warning_emitted: bool = False
         self.worker_cpu_warning_emitted: bool = False

locust/shape.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 import time
-from typing import ClassVar, Optional, Tuple, List, Type
+from typing import ClassVar, Optional, Tuple, List, Type, Union
 from abc import ABCMeta, abstractmethod
 
 from . import User
@@ -52,7 +52,7 @@ def get_current_user_count(self):
         return self.runner.user_count
 
     @abstractmethod
-    def tick(self) -> Tuple[int, float] | Tuple[int, float, Optional[List[Type[User]]]] | None:
+    def tick(self) -> Union[Tuple[int, float], Tuple[int, float, Optional[List[Type[User]]]], None]:
         """
         Returns a tuple with 2 elements to control the running load test:
 

locust/stats.py

@@ -30,6 +30,7 @@
     cast,
     Protocol,
     TypedDict,
+    Union,
 )
 
 from types import FrameType
@@ -56,7 +57,7 @@
 
 class CSVWriter(Protocol):
     @abstractmethod
-    def writerow(self, columns: Iterable[str | int | float]) -> None:
+    def writerow(self, columns: Iterable[Union[str, int, float]]) -> None:
         ...
 
 
@@ -241,7 +242,7 @@ def log_request(self, method: str, name: str, response_time: int, content_length
         self.total.log(response_time, content_length)
         self.entries[(name, method)].log(response_time, content_length)
 
-    def log_error(self, method: str, name: str, error: Exception | str | None) -> None:
+    def log_error(self, method: str, name: str, error: Optional[Union[Exception, str]]) -> None:
         self.total.log_error(error)
         self.entries[(name, method)].log_error(error)
 
@@ -456,7 +457,7 @@ def current_rps(self) -> float:
             return 0
         slice_start_time = max(int(self.stats.last_request_timestamp) - 12, int(self.stats.start_time or 0))
 
-        reqs: List[int | float] = [
+        reqs: List[Union[int, float]] = [
             self.num_reqs_per_sec.get(t, 0) for t in range(slice_start_time, int(self.stats.last_request_timestamp) - 2)
         ]
         return avg(reqs)
@@ -704,14 +705,14 @@ def to_dict(self, escape_string_values=False):
 
 
 class StatsError:
-    def __init__(self, method: str, name: str, error: Exception | str | None, occurrences: int = 0):
+    def __init__(self, method: str, name: str, error: Optional[Union[Exception, str]], occurrences: int = 0):
         self.method = method
         self.name = name
         self.error = error
         self.occurrences = occurrences
 
     @classmethod
-    def parse_error(cls, error: Exception | str | None) -> str:
+    def parse_error(cls, error: Optional[Union[Exception, str]]) -> str:
         string_error = repr(error)
         target = "object at 0x"
         target_index = string_error.find(target)
@@ -725,7 +726,7 @@ def parse_error(cls, error: Exception | str | None) -> str:
         return string_error.replace(hex_address, "0x....")
 
     @classmethod
-    def create_key(cls, method: str, name: str, error: Exception | str | None) -> str:
+    def create_key(cls, method: str, name: str, error: Optional[Union[Exception, str]]) -> str:
         key = f"{method}.{name}.{StatsError.parse_error(error)!r}"
         return hashlib.sha256(key.encode("utf-8")).hexdigest()
 
@@ -771,7 +772,7 @@ def to_dict(self, escape_string_values=False):
         }
 
 
-def avg(values: List[float | int]) -> float:
+def avg(values: List[Union[float, int]]) -> float:
     return sum(values, 0.0) / max(len(values), 1)
 
 
@@ -977,7 +978,7 @@ def __init__(self, environment: "Environment", percentiles_to_report: List[float
             "Nodes",
         ]
 
-    def _percentile_fields(self, stats_entry: StatsEntry, use_current: bool = False) -> List[str] | List[int]:
+    def _percentile_fields(self, stats_entry: StatsEntry, use_current: bool = False) -> Union[List[str], List[int]]:
         if not stats_entry.num_requests:
             return self.percentiles_na
         elif use_current:

locust/user/task.py

@@ -16,6 +16,7 @@
     Protocol,
     final,
     runtime_checkable,
+    Union,
 )
 import gevent
 from gevent import GreenletExit
@@ -47,7 +48,7 @@ def task(weight: int) -> Callable[[TaskT], TaskT]:
     ...
 
 
-def task(weight: TaskT | int = 1) -> TaskT | Callable[[TaskT], TaskT]:
+def task(weight: Union[TaskT, int] = 1) -> Union[TaskT, Callable[[TaskT], TaskT]]:
     """
     Used as a convenience decorator to be able to declare tasks for a User or a TaskSet
     inline in the class. Example::
@@ -238,7 +239,7 @@ class TaskSet(metaclass=TaskSetMeta):
     will then continue in the first TaskSet).
     """
 
-    tasks: List[TaskSet | Callable] = []
+    tasks: List[Union["TaskSet", Callable]] = []
     """
     Collection of python callables and/or TaskSet classes that the User(s) will run.
 

locust/user/users.py

@@ -1,5 +1,5 @@
 from __future__ import annotations
-from typing import Callable, Dict, List, Optional, final
+from typing import Callable, Dict, List, Optional, final, Union
 
 import time
 from gevent import GreenletExit, greenlet
@@ -89,7 +89,7 @@ class MyUser(User):
     Method that returns the time between the execution of locust tasks in milliseconds
     """
 
-    tasks: List[TaskSet | Callable] = []
+    tasks: List[Union[TaskSet, Callable]] = []
     """
     Collection of python callables and/or TaskSet classes that the Locust user(s) will run.
 

locust/web.py

@@ -9,7 +9,7 @@
 from json import dumps
 from itertools import chain
 from time import time
-from typing import TYPE_CHECKING, Optional, Any, Dict, List
+from typing import TYPE_CHECKING, Optional, Any, Dict, List, Union
 
 import gevent
 from flask import (
@@ -153,7 +153,7 @@ def send_assets(path):
 
         @app.route("/")
         @self.auth_required_if_enabled
-        def index() -> str | Response:
+        def index() -> Union[str, Response]:
             if not environment.runner:
                 return make_response("Error: Locust Environment does not have any runner", 500)
             self.update_template_args()