Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright (c) Meta Platforms, Inc. and affiliates.
2#
3# This source code is licensed under the MIT license found in the
4# LICENSE file in the root directory of this source tree.
6import sys
7from concurrent.futures import Executor, Future
8from types import TracebackType
9from typing import Callable, Optional, Type, TypeVar
11if sys.version_info >= (3, 10):
12 from typing import ParamSpec
13else:
14 from typing_extensions import ParamSpec
16Return = TypeVar("Return")
17Params = ParamSpec("Params")
20class DummyExecutor(Executor):
21 """
22 Synchronous dummy `concurrent.futures.Executor` analogue.
23 """
25 def submit(
26 self,
27 fn: Callable[Params, Return],
28 /,
29 *args: Params.args,
30 **kwargs: Params.kwargs,
31 ) -> Future[Return]:
32 future: Future[Return] = Future()
33 try:
34 result = fn(*args, **kwargs)
35 future.set_result(result)
36 except Exception as exc:
37 future.set_exception(exc)
38 return future
40 def __enter__(self) -> "DummyExecutor":
41 return self
43 def __exit__(
44 self,
45 exc_type: Optional[Type[BaseException]],
46 exc_val: Optional[BaseException],
47 exc_tb: Optional[TracebackType],
48 ) -> None:
49 pass