Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-01-07 23:15
📁 Deprecated-1.2.18.dist-info	-	2025-05-25 12:26
📁 PyMySQL-1.1.1.dist-info	-	2025-05-25 12:26
📁 __pycache__	-	2026-01-07 23:15
📁 _distutils_hack	-	2026-03-27 02:58
📁 aiosignal	-	2025-05-25 12:26
📁 aiosignal-1.3.2.dist-info	-	2025-05-25 12:26
📁 async_timeout	-	2025-05-25 12:26
📁 async_timeout-4.0.2.dist-info	-	2025-05-25 12:26
📁 attr	-	2025-05-25 12:26
📁 attrs	-	2025-05-25 12:26
📁 attrs-25.3.0.dist-info	-	2025-05-25 12:26
📁 boto3	-	2025-05-25 12:26
📁 boto3-1.38.23.dist-info	-	2025-05-25 12:26
📁 botocore	-	2025-05-25 12:26
📁 botocore-1.38.23.dist-info	-	2025-05-25 12:26
📁 cachetools	-	2025-05-25 12:26
📁 cachetools-5.5.2.dist-info	-	2025-05-25 12:26
📁 certifi	-	2025-05-25 12:26
📁 certifi-2025.4.26.dist-info	-	2025-05-25 12:26
📁 chardet	-	2025-05-25 12:26
📁 chardet-4.0.0.dist-info	-	2025-05-25 12:26
📁 click	-	2025-05-25 12:26
📁 click-8.1.8.dist-info	-	2025-05-25 12:26
📁 dateutil	-	2025-05-25 12:26
📁 deprecated	-	2025-05-25 12:26
📄 distutils-precedence.pth import os; var = 'SETUPTOOLS_USE_DISTUTILS'; enabled = os.environ.get(var, 'local') == 'local'; enabled and __import__('_distutils_hack').add_shim();	151 B	2023-11-13 22:05
📄 easy_install.py """Run the EasyInstall command""" if __name__ == '__main__': from setuptools.command.easy_install import main main()	126 B	2025-05-25 12:25
📁 elastic_transport	-	2025-05-25 12:26
📁 elastic_transport-8.17.1.dist-info	-	2025-05-25 12:26
📁 elasticsearch	-	2025-05-25 12:26
📁 elasticsearch-9.0.1.dist-info	-	2025-05-25 12:26
📁 fluent	-	2025-05-25 12:26
📁 fluent_logger-0.11.1.dist-info	-	2025-05-25 12:26
📁 google	-	2025-05-25 12:26
📁 google_api_core-2.24.2.dist-info	-	2025-05-25 12:26
📁 google_auth-2.40.2.dist-info	-	2025-05-25 12:26
📁 google_cloud_pubsub-2.29.0.dist-info	-	2025-05-25 12:26
📁 googleapis_common_protos-1.70.0.dist-info	-	2025-05-25 12:26
📁 graypy	-	2025-05-25 12:26
📁 graypy-2.1.0.dist-info	-	2025-05-25 12:26
📁 grpc_google_iam_v1-0.14.2.dist-info	-	2025-05-25 12:26
📁 grpc_status	-	2025-05-25 12:26
📁 grpcio_status-1.71.0.dist-info	-	2025-05-25 12:26
📁 idna	-	2025-05-25 12:26
📁 idna-2.10.dist-info	-	2025-05-25 12:26
📁 importlib_metadata	-	2025-05-25 12:26
📁 importlib_metadata-8.6.1.dist-info	-	2025-05-25 12:26
📁 jmespath	-	2025-05-25 12:26
📁 jmespath-1.0.1.dist-info	-	2025-05-25 12:26
📁 jsons	-	2025-05-25 12:26
📁 jsons-1.6.3.dist-info	-	2025-05-25 12:26
📁 loguru	-	2025-05-25 12:26
📁 loguru-0.7.3.dist-info	-	2025-05-25 12:26
📁 markdown_it	-	2025-05-25 12:26
📁 markdown_it_py-3.0.0.dist-info	-	2025-05-25 12:26
📁 mdurl	-	2025-05-25 12:26
📁 mdurl-0.1.2.dist-info	-	2025-05-25 12:26
📁 opentelemetry	-	2025-05-25 12:26
📁 opentelemetry_api-1.33.1.dist-info	-	2025-05-25 12:26
📁 opentelemetry_sdk-1.33.1.dist-info	-	2025-05-25 12:26
📁 opentelemetry_semantic_conventions-0.54b1.dist-info	-	2025-05-25 12:26
📁 packaging	-	2025-05-25 12:26
📁 packaging-25.0.dist-info	-	2025-05-25 12:26
📁 pip	-	2026-03-21 15:53
📁 pip-21.3.1.dist-info	-	2026-03-21 15:53
📁 pkg_resources	-	2026-03-27 02:58
📁 prometheus_client	-	2025-05-25 12:26
📁 prometheus_client-0.22.0.dist-info	-	2025-05-25 12:26
📁 proto	-	2025-05-25 12:26
📁 proto_plus-1.26.1.dist-info	-	2025-05-25 12:26
📁 pyasn1	-	2025-05-25 12:26
📁 pyasn1-0.6.1.dist-info	-	2025-05-25 12:26
📁 pyasn1_modules	-	2025-05-25 12:26
📁 pyasn1_modules-0.4.2.dist-info	-	2025-05-25 12:26
📁 pydig	-	2025-05-25 12:26
📁 pydig-0.4.0.dist-info	-	2025-05-25 12:26
📁 pygments	-	2025-05-25 12:26
📁 pygments-2.19.1.dist-info	-	2025-05-25 12:26
📁 pymysql	-	2025-05-25 12:26
📁 pyroute2	-	2025-05-25 12:26
📁 pyroute2-0.9.2.dist-info	-	2025-05-25 12:26
📁 python_dateutil-2.9.0.post0.dist-info	-	2025-05-25 12:26
📁 redis	-	2025-05-25 12:26
📁 redis-4.1.4.dist-info	-	2025-05-25 12:26
📁 requests	-	2025-05-25 12:26
📁 requests-2.25.1.dist-info	-	2025-05-25 12:26
📁 rich	-	2025-05-25 12:26
📁 rich-14.0.0.dist-info	-	2025-05-25 12:26
📁 rsa	-	2025-05-25 12:26
📁 rsa-4.9.1.dist-info	-	2025-05-25 12:26
📁 s3transfer	-	2025-05-25 12:26
📁 s3transfer-0.13.0.dist-info	-	2025-05-25 12:26
📁 setuptools	-	2026-03-27 02:58
📁 setuptools-44.1.0.dist-info	-	2025-05-25 12:25
📁 setuptools-65.6.3.dist-info	-	2026-03-27 02:58
📁 sh-2.2.2.dist-info	-	2025-05-25 12:26
📄 sh.py """ https://sh.readthedocs.io/en/latest/ https://github.com/amoffat/sh """ # =============================================================================== # Copyright (C) 2011-2025 by Andrew Moffat # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN # THE SOFTWARE. # =============================================================================== import asyncio import platform from collections import deque from collections.abc import Mapping from importlib import metadata try: __version__ = metadata.version("sh") except metadata.PackageNotFoundError: # pragma: no cover __version__ = "unknown" if "windows" in platform.system().lower(): # pragma: no cover raise ImportError( f"sh {__version__} is currently only supported on Linux and macOS." ) import errno import fcntl import gc import getpass import glob as glob_module import inspect import logging import os import pty import pwd import re import select import signal import stat import struct import sys import termios import textwrap import threading import time import traceback import tty import warnings import weakref from asyncio import Queue as AQueue from contextlib import contextmanager from functools import partial from io import BytesIO, StringIO, UnsupportedOperation from io import open as fdopen from locale import getpreferredencoding from queue import Empty, Queue from shlex import quote as shlex_quote from types import GeneratorType, ModuleType from typing import Any, Dict, Type, Union __project_url__ = "https://github.com/amoffat/sh" TEE_STDOUT = {True, "out", 1} TEE_STDERR = {"err", 2} DEFAULT_ENCODING = getpreferredencoding() or "UTF-8" IS_MACOS = platform.system() in ("AIX", "Darwin") SH_LOGGER_NAME = __name__ # normally i would hate this idea of using a global to signify whether we are # running tests, because it breaks the assumption that what is running in the # tests is what will run live, but we ONLY use this in a place that has no # serious side-effects that could change anything. as long as we do that, it # should be ok RUNNING_TESTS = bool(int(os.environ.get("SH_TESTS_RUNNING", "0"))) FORCE_USE_SELECT = bool(int(os.environ.get("SH_TESTS_USE_SELECT", "0"))) # a re-entrant lock for pushd. this way, multiple threads that happen to use # pushd will all see the current working directory for the duration of the # with-context PUSHD_LOCK = threading.RLock() def get_num_args(fn): return len(inspect.getfullargspec(fn).args) _unicode_methods = set(dir("")) HAS_POLL = hasattr(select, "poll") POLLER_EVENT_READ = 1 POLLER_EVENT_WRITE = 2 POLLER_EVENT_HUP = 4 POLLER_EVENT_ERROR = 8 class PollPoller: def __init__(self): self._poll = select.poll() # file descriptor <-> file object bidirectional maps self.fd_lookup = {} self.fo_lookup = {} def __nonzero__(self): return len(self.fd_lookup) != 0 def __len__(self): return len(self.fd_lookup) def _set_fileobject(self, f): if hasattr(f, "fileno"): fd = f.fileno() self.fd_lookup[fd] = f self.fo_lookup[f] = fd else: self.fd_lookup[f] = f self.fo_lookup[f] = f def _remove_fileobject(self, f): if hasattr(f, "fileno"): fd = f.fileno() del self.fd_lookup[fd] del self.fo_lookup[f] else: del self.fd_lookup[f] del self.fo_lookup[f] def _get_file_descriptor(self, f): return self.fo_lookup.get(f) def _get_file_object(self, fd): return self.fd_lookup.get(fd) def _register(self, f, events): # f can be a file descriptor or file object self._set_fileobject(f) fd = self._get_file_descriptor(f) self._poll.register(fd, events) def register_read(self, f): self._register(f, select.POLLIN \| select.POLLPRI) def register_write(self, f): self._register(f, select.POLLOUT) def register_error(self, f): self._register(f, select.POLLERR \| select.POLLHUP \| select.POLLNVAL) def unregister(self, f): fd = self._get_file_descriptor(f) self._poll.unregister(fd) self._remove_fileobject(f) def poll(self, timeout): if timeout is not None: # convert from seconds to milliseconds timeout = 1000 changes = self._poll.poll(timeout) results = [] for fd, events in changes: f = self._get_file_object(fd) if events & (select.POLLIN \| select.POLLPRI): results.append((f, POLLER_EVENT_READ)) elif events & select.POLLOUT: results.append((f, POLLER_EVENT_WRITE)) elif events & select.POLLHUP: results.append((f, POLLER_EVENT_HUP)) elif events & (select.POLLERR \| select.POLLNVAL): results.append((f, POLLER_EVENT_ERROR)) return results class SelectPoller: def __init__(self): self.rlist = [] self.wlist = [] self.xlist = [] def __nonzero__(self): return len(self.rlist) + len(self.wlist) + len(self.xlist) != 0 def __len__(self): return len(self.rlist) + len(self.wlist) + len(self.xlist) @staticmethod def _register(f, events): if f not in events: events.append(f) @staticmethod def _unregister(f, events): if f in events: events.remove(f) def register_read(self, f): self._register(f, self.rlist) def register_write(self, f): self._register(f, self.wlist) def register_error(self, f): self._register(f, self.xlist) def unregister(self, f): self._unregister(f, self.rlist) self._unregister(f, self.wlist) self._unregister(f, self.xlist) def poll(self, timeout): _in, _out, _err = select.select(self.rlist, self.wlist, self.xlist, timeout) results = [] for f in _in: results.append((f, POLLER_EVENT_READ)) for f in _out: results.append((f, POLLER_EVENT_WRITE)) for f in _err: results.append((f, POLLER_EVENT_ERROR)) return results # here we use an use a poller interface that transparently selects the most # capable poller (out of either select.select or select.poll). this was added # by zhangyafeikimi when he discovered that if the fds created internally by sh # numbered > 1024, select.select failed (a limitation of select.select). this # can happen if your script opens a lot of files Poller: Union[Type[SelectPoller], Type[PollPoller]] = SelectPoller if HAS_POLL and not FORCE_USE_SELECT: Poller = PollPoller class ForkException(Exception): def __init__(self, orig_exc): msg = f""" Original exception: =================== {textwrap.indent(orig_exc, " ")} """ Exception.__init__(self, msg) class ErrorReturnCodeMeta(type): """a metaclass which provides the ability for an ErrorReturnCode (or derived) instance, imported from one sh module, to be considered the subclass of ErrorReturnCode from another module. this is mostly necessary in the tests, where we do assertRaises, but the ErrorReturnCode that the program we're testing throws may not be the same class that we pass to assertRaises """ def __subclasscheck__(self, o): other_bases = {b.__name__ for b in o.__bases__} return self.__name__ in other_bases or o.__name__ == self.__name__ class ErrorReturnCode(Exception): __metaclass__ = ErrorReturnCodeMeta """ base class for all exceptions as a result of a command's exit status being deemed an error. this base class is dynamically subclassed into derived classes with the format: ErrorReturnCode_NNN where NNN is the exit code number. the reason for this is it reduces boiler plate code when testing error return codes: try: some_cmd() except ErrorReturnCode_12: print("couldn't do X") vs: try: some_cmd() except ErrorReturnCode as e: if e.exit_code == 12: print("couldn't do X") it's not much of a savings, but i believe it makes the code easier to read """ truncate_cap = 750 def __reduce__(self): return self.__class__, (self.full_cmd, self.stdout, self.stderr, self.truncate) def __init__(self, full_cmd, stdout, stderr, truncate=True): self.exit_code = self.exit_code # makes pylint happy self.full_cmd = full_cmd self.stdout = stdout self.stderr = stderr self.truncate = truncate exc_stdout = self.stdout if truncate: exc_stdout = exc_stdout[: self.truncate_cap] out_delta = len(self.stdout) - len(exc_stdout) if out_delta: exc_stdout += (f"... ({out_delta} more, please see e.stdout)").encode() exc_stderr = self.stderr if truncate: exc_stderr = exc_stderr[: self.truncate_cap] err_delta = len(self.stderr) - len(exc_stderr) if err_delta: exc_stderr += (f"... ({err_delta} more, please see e.stderr)").encode() msg = ( f"\n\n RAN: {self.full_cmd}" f"\n\n STDOUT:\n{exc_stdout.decode(DEFAULT_ENCODING, 'replace')}" f"\n\n STDERR:\n{exc_stderr.decode(DEFAULT_ENCODING, 'replace')}" ) super().__init__(msg) class SignalException(ErrorReturnCode): pass class TimeoutException(Exception): """the exception thrown when a command is killed because a specified timeout (via _timeout or .wait(timeout)) was hit""" def __init__(self, exit_code, full_cmd): self.exit_code = exit_code self.full_cmd = full_cmd super(Exception, self).__init__() SIGNALS_THAT_SHOULD_THROW_EXCEPTION = { signal.SIGABRT, signal.SIGBUS, signal.SIGFPE, signal.SIGILL, signal.SIGINT, signal.SIGKILL, signal.SIGPIPE, signal.SIGQUIT, signal.SIGSEGV, signal.SIGTERM, signal.SIGSYS, } # we subclass AttributeError because: # https://github.com/ipython/ipython/issues/2577 # https://github.com/amoffat/sh/issues/97#issuecomment-10610629 class CommandNotFound(AttributeError): pass rc_exc_regex = re.compile(r"(ErrorReturnCode\|SignalException)_((\d+)\|SIG[a-zA-Z]+)") rc_exc_cache: Dict[str, Type[ErrorReturnCode]] = {} SIGNAL_MAPPING = { v: k for k, v in signal.__dict__.items() if re.match(r"SIG[a-zA-Z]+", k) } def get_exc_from_name(name): """takes an exception name, like: ErrorReturnCode_1 SignalException_9 SignalException_SIGHUP and returns the corresponding exception. this is primarily used for importing exceptions from sh into user code, for instance, to capture those exceptions""" exc = None try: return rc_exc_cache[name] except KeyError: m = rc_exc_regex.match(name) if m: base = m.group(1) rc_or_sig_name = m.group(2) if base == "SignalException": try: rc = -int(rc_or_sig_name) except ValueError: rc = -getattr(signal, rc_or_sig_name) else: rc = int(rc_or_sig_name) exc = get_rc_exc(rc) return exc def get_rc_exc(rc): """takes a exit code or negative signal number and produces an exception that corresponds to that return code. positive return codes yield ErrorReturnCode exception, negative return codes yield SignalException we also cache the generated exception so that only one signal of that type exists, preserving identity""" try: return rc_exc_cache[rc] except KeyError: pass if rc >= 0: name = f"ErrorReturnCode_{rc}" base = ErrorReturnCode else: name = f"SignalException_{SIGNAL_MAPPING[abs(rc)]}" base = SignalException exc = ErrorReturnCodeMeta(name, (base,), {"exit_code": rc}) rc_exc_cache[rc] = exc return exc # we monkey patch glob. i'm normally generally against monkey patching, but i # decided to do this really un-intrusive patch because we need a way to detect # if a list that we pass into an sh command was generated from glob. the reason # being that glob returns an empty list if a pattern is not found, and so # commands will treat the empty list as no arguments, which can be a problem, # ie: # # ls(glob(".ojfawe")) # # ^ will show the contents of your home directory, because it's essentially # running ls([]) which, as a process, is just "ls". # # so we subclass list and monkey patch the glob function. nobody should be the # wiser, but we'll have results that we can make some determinations on _old_glob = glob_module.glob class GlobResults(list): def __init__(self, path, results): self.path = path list.__init__(self, results) def glob(path, args, kwargs): expanded = GlobResults(path, _old_glob(path, args, kwargs)) return expanded glob_module.glob = glob # type: ignore def canonicalize(path): return os.path.abspath(os.path.expanduser(path)) def _which(program, paths=None): """takes a program name or full path, plus an optional collection of search paths, and returns the full path of the requested executable. if paths is specified, it is the entire list of search paths, and the PATH env is not used at all. otherwise, PATH env is used to look for the program""" def is_exe(file_path): return ( os.path.exists(file_path) and os.access(file_path, os.X_OK) and os.path.isfile(os.path.realpath(file_path)) ) found_path = None fpath, fname = os.path.split(program) # if there's a path component, then we've specified a path to the program, # and we should just test if that program is executable. if it is, return if fpath: program = canonicalize(program) if is_exe(program): found_path = program # otherwise, we've just passed in the program name, and we need to search # the paths to find where it actually lives else: paths_to_search = [] if isinstance(paths, (tuple, list)): paths_to_search.extend(paths) else: env_paths = os.environ.get("PATH", "").split(os.pathsep) paths_to_search.extend(env_paths) for path in paths_to_search: exe_file = os.path.join(canonicalize(path), program) if is_exe(exe_file): found_path = exe_file break return found_path def resolve_command_path(program): path = _which(program) if not path: # our actual command might have a dash in it, but we can't call # that from python (we have to use underscores), so we'll check # if a dash version of our underscore command exists and use that # if it does if "_" in program: path = _which(program.replace("_", "-")) if not path: return None return path def resolve_command(name, command_cls, baked_args=None): path = resolve_command_path(name) cmd = None if path: cmd = command_cls(path) if baked_args: cmd = cmd.bake(baked_args) return cmd class Logger: """provides a memory-inexpensive logger. a gotcha about python's builtin logger is that logger objects are never garbage collected. if you create a thousand loggers with unique names, they'll sit there in memory until your script is done. with sh, it's easy to create loggers with unique names if we want our loggers to include our command arguments. for example, these are all unique loggers: ls -l ls -l /tmp ls /tmp so instead of creating unique loggers, and without sacrificing logging output, we use this class, which maintains as part of its state, the logging "context", which will be the very unique name. this allows us to get a logger with a very general name, eg: "command", and have a unique name appended to it via the context, eg: "ls -l /tmp" """ def __init__(self, name, context=None): self.name = name self.log = logging.getLogger(f"{SH_LOGGER_NAME}.{name}") self.context = self.sanitize_context(context) def _format_msg(self, msg, a): if self.context: msg = f"{self.context}: {msg}" return msg % a @staticmethod def sanitize_context(context): if context: context = context.replace("%", "%%") return context or "" def get_child(self, name, context): new_name = self.name + "." + name new_context = self.context + "." + context return Logger(new_name, new_context) def info(self, msg, a): self.log.info(self._format_msg(msg, a)) def debug(self, msg, a): self.log.debug(self._format_msg(msg, a)) def error(self, msg, a): self.log.error(self._format_msg(msg, a)) def exception(self, msg, a): self.log.exception(self._format_msg(msg, a)) def default_logger_str(cmd, call_args, pid=None): if pid: s = f"<Command {cmd!r}, pid {pid}>" else: s = f"<Command {cmd!r}>" return s class RunningCommand: """this represents an executing Command object. it is returned as the result of __call__() being executed on a Command instance. this creates a reference to a OProc instance, which is a low-level wrapper around the process that was exec'd this is the class that gets manipulated the most by user code, and so it implements various convenience methods and logical mechanisms for the underlying process. for example, if a user tries to access a backgrounded-process's stdout/err, the RunningCommand object is smart enough to know to wait() on the process to finish first. and when the process finishes, RunningCommand is smart enough to translate exit codes to exceptions.""" # these are attributes that we allow to pass through to OProc _OProc_attr_allowlist = { "signal", "terminate", "kill", "kill_group", "signal_group", "pid", "sid", "pgid", "ctty", "input_thread_exc", "output_thread_exc", "bg_thread_exc", } def __init__(self, cmd, call_args, stdin, stdout, stderr): # self.ran is used for auditing what actually ran. for example, in # exceptions, or if you just want to know what was ran after the # command ran self.ran = " ".join([shlex_quote(str(arg)) for arg in cmd]) self.call_args = call_args self.cmd = cmd self.process = None self._waited_until_completion = False should_wait = True spawn_process = True # if we're using an async for loop on this object, we need to put the underlying # iterable in no-block mode. however, we will only know if we're using an async # for loop after this object is constructed. so we'll set it to False now, but # then later set it to True if we need it self._force_noblock_iter = False # this event is used when we want to `await` a RunningCommand. see how it gets # used in self.__await__ try: asyncio.get_running_loop() except RuntimeError: self.aio_output_complete = None else: self.aio_output_complete = asyncio.Event() # this is used to track if we've already raised StopIteration, and if we # have, raise it immediately again if the user tries to call next() on # us. https://github.com/amoffat/sh/issues/273 self._stopped_iteration = False # with contexts shouldn't run at all yet, they prepend # to every command in the context if call_args["with"]: spawn_process = False get_prepend_stack().append(self) if call_args["piped"] or call_args["iter"] or call_args["iter_noblock"]: should_wait = False if call_args["async"]: should_wait = False # we're running in the background, return self and let us lazily # evaluate if call_args["bg"]: should_wait = False # redirection if call_args["err_to_out"]: stderr = OProc.STDOUT done_callback = call_args["done"] if done_callback: call_args["done"] = partial(done_callback, self) # set up which stream should write to the pipe # TODO, make pipe None by default and limit the size of the Queue # in oproc.OProc pipe = OProc.STDOUT if call_args["iter"] == "out" or call_args["iter"] is True: pipe = OProc.STDOUT elif call_args["iter"] == "err": pipe = OProc.STDERR if call_args["iter_noblock"] == "out" or call_args["iter_noblock"] is True: pipe = OProc.STDOUT elif call_args["iter_noblock"] == "err": pipe = OProc.STDERR # there's currently only one case where we wouldn't spawn a child # process, and that's if we're using a with-context with our command self._spawned_and_waited = False if spawn_process: log_str_factory = call_args["log_msg"] or default_logger_str logger_str = log_str_factory(self.ran, call_args) self.log = Logger("command", logger_str) self.log.debug("starting process") if should_wait: self._spawned_and_waited = True # this lock is needed because of a race condition where a background # thread, created in the OProc constructor, may try to access # self.process, but it has not been assigned yet process_assign_lock = threading.Lock() with process_assign_lock: self.process = OProc( self, self.log, cmd, stdin, stdout, stderr, self.call_args, pipe, process_assign_lock, ) logger_str = log_str_factory(self.ran, call_args, self.process.pid) self.log.context = self.log.sanitize_context(logger_str) self.log.info("process started") if should_wait: self.wait() def wait(self, timeout=None): """waits for the running command to finish. this is called on all running commands, eventually, except for ones that run in the background if timeout is a number, it is the number of seconds to wait for the process to resolve. otherwise block on wait. this function can raise a TimeoutException, either because of a `_timeout` on the command itself as it was launched, or because of a timeout passed into this method. """ if not self._waited_until_completion: # if we've been given a timeout, we need to poll is_alive() if timeout is not None: waited_for = 0 sleep_amt = 0.1 alive = False exit_code = None if timeout < 0: raise RuntimeError("timeout cannot be negative") # while we still have time to wait, run this loop # notice that alive and exit_code are only defined in this loop, but # the loop is also guaranteed to run, defining them, given the # constraints that timeout is non-negative while waited_for <= timeout: alive, exit_code = self.process.is_alive() # if we're alive, we need to wait some more, but let's sleep # before we poll again if alive: time.sleep(sleep_amt) waited_for += sleep_amt # but if we're not alive, we're done waiting else: break # if we've made it this far, and we're still alive, then it means we # timed out waiting if alive: raise TimeoutException(None, self.ran) # if we didn't time out, we fall through and let the rest of the code # handle exit_code. notice that we set _waited_until_completion here, # only if we didn't time out. this allows us to re-wait again on # timeout, if we catch the TimeoutException in the parent frame self._waited_until_completion = True else: exit_code = self.process.wait() self._waited_until_completion = True if self.process.timed_out: # if we timed out, our exit code represents a signal, which is # negative, so let's make it positive to store in our # TimeoutException raise TimeoutException(-exit_code, self.ran) else: self.handle_command_exit_code(exit_code) # if an iterable command is using an instance of OProc for its stdin, # wait on it. the process is probably set to "piped", which means it # won't be waited on, which means exceptions won't propagate up to the # main thread. this allows them to bubble up if self.process._stdin_process: self.process._stdin_process.command.wait() self.log.debug("process completed") return self def is_alive(self): """returns whether or not we're still alive. this call has side-effects on OProc""" return self.process.is_alive()[0] def handle_command_exit_code(self, code): """here we determine if we had an exception, or an error code that we weren't expecting to see. if we did, we create and raise an exception """ ca = self.call_args exc_class = get_exc_exit_code_would_raise(code, ca["ok_code"], ca["piped"]) if exc_class: exc = exc_class( self.ran, self.process.stdout, self.process.stderr, ca["truncate_exc"] ) raise exc @property def stdout(self): self.wait() return self.process.stdout @property def stderr(self): self.wait() return self.process.stderr @property def exit_code(self): self.wait() return self.process.exit_code def __len__(self): return len(str(self)) def __enter__(self): """we don't actually do anything here because anything that should have been done would have been done in the Command.__call__ call. essentially all that has to happen is the command be pushed on the prepend stack.""" pass def __iter__(self): return self def __next__(self): """allow us to iterate over the output of our command""" if self._stopped_iteration: raise StopIteration() pq = self.process._pipe_queue # the idea with this is, if we're using regular `_iter` (non-asyncio), then we # want to have blocking be True when we read from the pipe queue, so our cpu # doesn't spin too fast. however, if we are* using asyncio (an async for loop), # then we want non-blocking pipe queue reads, because we'll do an asyncio.sleep, # in the coroutine that is doing the iteration, this way coroutines have better # yielding (see queue_connector in __aiter__). block_pq_read = not self._force_noblock_iter # we do this because if get blocks, we can't catch a KeyboardInterrupt # so the slight timeout allows for that. while True: try: chunk = pq.get(block_pq_read, self.call_args["iter_poll_time"]) except Empty: if self.call_args["iter_noblock"] or self._force_noblock_iter: return errno.EWOULDBLOCK else: if chunk is None: self.wait() self._stopped_iteration = True raise StopIteration() try: return chunk.decode( self.call_args["encoding"], self.call_args["decode_errors"] ) except UnicodeDecodeError: return chunk def __await__(self): async def wait_for_completion(): await self.aio_output_complete.wait() if self.call_args["return_cmd"]: # We know the command has completed already, # but need to catch exceptions self.wait() return self else: return str(self) return wait_for_completion().__await__() def __aiter__(self): # maxsize is critical to making sure our queue_connector function below yields # when it awaits _aio_queue.put(chunk). if we didn't have a maxsize, our loop # would happily iterate through `chunk in self` and put onto the queue without # any blocking, and therefore no yielding, which would prevent other coroutines # from running. self._aio_queue = AQueue(maxsize=1) self._force_noblock_iter = True # the sole purpose of this coroutine is to connect our pipe_queue (which is # being populated by a thread) to an asyncio-friendly queue. then, in __anext__, # we can iterate over that asyncio queue. async def queue_connector(): try: # this will spin as fast as possible if there's no data to read, # thanks to self._force_noblock_iter. so we sleep below. for chunk in self: if chunk == errno.EWOULDBLOCK: # let us have better coroutine yielding. await asyncio.sleep(0.01) else: await self._aio_queue.put(chunk) finally: await self._aio_queue.put(None) task = asyncio.create_task(queue_connector()) self._aio_task = task return self async def __anext__(self): chunk = await self._aio_queue.get() if chunk is not None: return chunk else: exc = self._aio_task.exception() if exc is not None: raise exc raise StopAsyncIteration def __exit__(self, exc_type, exc_val, exc_tb): if self.call_args["with"] and get_prepend_stack(): get_prepend_stack().pop() def __str__(self): if self.process and self.stdout: return self.stdout.decode( self.call_args["encoding"], self.call_args["decode_errors"] ) return "" def __eq__(self, other): return id(self) == id(other) def __contains__(self, item): return item in str(self) def __getattr__(self, p): # let these three attributes pass through to the OProc object if p in self._OProc_attr_allowlist: if self.process: return getattr(self.process, p) else: raise AttributeError # see if strings have what we're looking for if p in _unicode_methods: return getattr(str(self), p) raise AttributeError def __repr__(self): try: return str(self) except UnicodeDecodeError: if self.process: if self.stdout: return repr(self.stdout) return repr("") def __long__(self): return int(str(self).strip()) def __float__(self): return float(str(self).strip()) def __int__(self): return int(str(self).strip()) def output_redirect_is_filename(out): return isinstance(out, str) or hasattr(out, "__fspath__") def get_prepend_stack(): tl = Command.thread_local if not hasattr(tl, "_prepend_stack"): tl._prepend_stack = [] return tl._prepend_stack def special_kwarg_validator(passed_kwargs, merged_kwargs, invalid_list): s1 = set(passed_kwargs.keys()) invalid_args = [] for elem in invalid_list: if callable(elem): fn = elem ret = fn(passed_kwargs, merged_kwargs) invalid_args.extend(ret) else: elem, error_msg = elem if s1.issuperset(elem): invalid_args.append((elem, error_msg)) return invalid_args def get_fileno(ob): # in py2, this will return None. in py3, it will return an method that # raises when called fileno_meth = getattr(ob, "fileno", None) fileno = None if fileno_meth: # py3 StringIO objects will report a fileno, but calling it will raise # an exception try: fileno = fileno_meth() except UnsupportedOperation: pass elif isinstance(ob, (int,)) and ob >= 0: fileno = ob return fileno def ob_is_fd_based(ob): return get_fileno(ob) is not None def ob_is_tty(ob): """checks if an object (like a file-like object) is a tty.""" fileno = get_fileno(ob) is_tty = False if fileno is not None: is_tty = os.isatty(fileno) return is_tty def ob_is_pipe(ob): fileno = get_fileno(ob) is_pipe = False if fileno: fd_stat = os.fstat(fileno) is_pipe = stat.S_ISFIFO(fd_stat.st_mode) return is_pipe def output_iterator_validator(passed_kwargs, merged_kwargs): invalid = [] if passed_kwargs.get("no_out") and passed_kwargs.get("iter") in (True, "out"): error = "You cannot iterate over output if there is no output" invalid.append((("no_out", "iter"), error)) return invalid def tty_in_validator(passed_kwargs, merged_kwargs): # here we'll validate that people aren't randomly shotgun-debugging different tty # options and hoping that they'll work, without understanding what they do pairs = (("tty_in", "in"), ("tty_out", "out")) invalid = [] for tty_type, std in pairs: if tty_type in passed_kwargs and ob_is_tty(passed_kwargs.get(std, None)): error = ( f"`_{std}` is a TTY already, so so it doesn't make sense to set up a" f" TTY with `_{tty_type}`" ) invalid.append(((tty_type, std), error)) # if unify_ttys is set, then both tty_in and tty_out must both be True if merged_kwargs["unify_ttys"] and not ( merged_kwargs["tty_in"] and merged_kwargs["tty_out"] ): invalid.append( ( ("unify_ttys", "tty_in", "tty_out"), "`_tty_in` and `_tty_out` must both be True if `_unify_ttys` is True", ) ) return invalid def fg_validator(passed_kwargs, merged_kwargs): """fg is not valid with basically every other option""" invalid = [] msg = """\ _fg is invalid with nearly every other option, see warning and workaround here: https://sh.readthedocs.io/en/latest/sections/special_arguments.html#fg""" allowlist = {"env", "fg", "cwd", "ok_code"} offending = set(passed_kwargs.keys()) - allowlist if "fg" in passed_kwargs and passed_kwargs["fg"] and offending: invalid.append(("fg", msg)) return invalid def bufsize_validator(passed_kwargs, merged_kwargs): """a validator to prevent a user from saying that they want custom buffering when they're using an in/out object that will be os.dup'ed to the process, and has its own buffering. an example is a pipe or a tty. it doesn't make sense to tell them to have a custom buffering, since the os controls this.""" invalid = [] in_ob = passed_kwargs.get("in", None) out_ob = passed_kwargs.get("out", None) in_buf = passed_kwargs.get("in_bufsize", None) out_buf = passed_kwargs.get("out_bufsize", None) in_no_buf = ob_is_fd_based(in_ob) out_no_buf = ob_is_fd_based(out_ob) err = "Can't specify an {target} bufsize if the {target} target is a pipe or TTY" if in_no_buf and in_buf is not None: invalid.append((("in", "in_bufsize"), err.format(target="in"))) if out_no_buf and out_buf is not None: invalid.append((("out", "out_bufsize"), err.format(target="out"))) return invalid def env_validator(passed_kwargs, merged_kwargs): """a validator to check that env is a dictionary and that all environment variable keys and values are strings. Otherwise, we would exit with a confusing exit code 255.""" invalid = [] env = passed_kwargs.get("env", None) if env is None: return invalid if not isinstance(env, Mapping): invalid.append(("env", f"env must be dict-like. Got {env!r}")) return invalid for k, v in passed_kwargs["env"].items(): if not isinstance(k, str): invalid.append(("env", f"env key {k!r} must be a str")) if not isinstance(v, str): invalid.append(("env", f"value {v!r} of env key {k!r} must be a str")) return invalid class Command: """represents an un-run system program, like "ls" or "cd". because it represents the program itself (and not a running instance of it), it should hold very little state. in fact, the only state it does hold is baked arguments. when a Command object is called, the result that is returned is a RunningCommand object, which represents the Command put into an execution state.""" thread_local = threading.local() RunningCommandCls = RunningCommand _call_args: Dict[str, Any] = { "fg": False, # run command in foreground # run a command in the background. commands run in the background # ignore SIGHUP and do not automatically exit when the parent process # ends "bg": False, # automatically report exceptions for background commands "bg_exc": True, "with": False, # prepend the command to every command after it "in": None, "out": None, # redirect STDOUT "err": None, # redirect STDERR "err_to_out": None, # redirect STDERR to STDOUT # stdin buffer size # 1 for line, 0 for unbuffered, any other number for that amount "in_bufsize": 0, # stdout buffer size, same values as above "out_bufsize": 1, "err_bufsize": 1, # this is how big the output buffers will be for stdout and stderr. # this is essentially how much output they will store from the process. # we use a deque, so if it overflows past this amount, the first items # get pushed off as each new item gets added. # # NOTICE # this is not a BYTE size, this is a CHUNK size...meaning, that if # you're buffering out/err at 1024 bytes, the internal buffer size will # be "internal_bufsize" CHUNKS of 1024 bytes "internal_bufsize": 3 * 1024*2, "env": None, "piped": None, "iter": None, "iter_noblock": None, # the amount of time to sleep between polling for the iter output queue "iter_poll_time": 0.1, "ok_code": 0, "cwd": None, # the separator delimiting between a long-argument's name and its value # setting this to None will cause name and value to be two separate # arguments, like for short options # for example, --arg=derp, '=' is the long_sep "long_sep": "=", # the prefix used for long arguments "long_prefix": "--", # this is for programs that expect their input to be from a terminal. # ssh is one of those programs "tty_in": False, "tty_out": True, "unify_ttys": False, "encoding": DEFAULT_ENCODING, "decode_errors": "strict", # how long the process should run before it is auto-killed "timeout": None, "timeout_signal": signal.SIGKILL, # TODO write some docs on "long-running processes" # these control whether or not stdout/err will get aggregated together # as the process runs. this has memory usage implications, so sometimes # with long-running processes with a lot of data, it makes sense to # set these to true "no_out": False, "no_err": False, "no_pipe": False, # if any redirection is used for stdout or stderr, internal buffering # of that data is not stored. this forces it to be stored, as if # the output is being T'd to both the redirected destination and our # internal buffers "tee": None, # will be called when a process terminates regardless of exception "done": None, # a tuple (rows, columns) of the desired size of both the stdout and # stdin ttys, if ttys are being used "tty_size": (24, 80), # whether or not our exceptions should be truncated "truncate_exc": True, # a function to call after the child forks but before the process execs "preexec_fn": None, # UID to set after forking. Requires root privileges. Not supported on # Windows. "uid": None, # put the forked process in its own process session? "new_session": False, # put the forked process in its own process group? "new_group": False, # pre-process args passed into __call__. only really useful when used # in .bake() "arg_preprocess": None, # a callable that produces a log message from an argument tuple of the # command and the args "log_msg": None, # whether or not to close all inherited fds. typically, this should be True, # as inheriting fds can be a security vulnerability "close_fds": True, # a allowlist of the integer fds to pass through to the child process. setting # this forces close_fds to be True "pass_fds": set(), # return an instance of RunningCommand always. if this isn't True, then # sometimes we may return just a plain unicode string "return_cmd": False, "async": False, } # this is a collection of validators to make sure the special kwargs make # sense _kwarg_validators = ( (("err", "err_to_out"), "Stderr is already being redirected"), (("piped", "iter"), "You cannot iterate when this command is being piped"), ( ("piped", "no_pipe"), "Using a pipe doesn't make sense if you've disabled the pipe", ), output_iterator_validator, (("close_fds", "pass_fds"), "Passing `pass_fds` forces `close_fds` to be True"), tty_in_validator, bufsize_validator, env_validator, fg_validator, ) def __init__(self, path, search_paths=None): found = _which(path, search_paths) self._path = "" # is the command baked (aka, partially applied)? self._partial = False self._partial_baked_args = [] self._partial_call_args = {} # bugfix for functools.wraps. issue #121 self.__name__ = str(self) if not found: raise CommandNotFound(path) # the reason why we set the values early in the constructor, and again # here, is for people who have tools that inspect the stack on # exception. if CommandNotFound is raised, we need self._path and the # other attributes to be set correctly, so repr() works when they're # inspecting the stack. issue #304 self._path = found self.__name__ = str(self) def __getattribute__(self, name): # convenience get_attr = partial(object.__getattribute__, self) val = None if name.startswith("_"): val = get_attr(name) elif name == "bake": val = get_attr("bake") # here we have a way of getting past shadowed subcommands. for example, # if "git bake" was a thing, we wouldn't be able to do `git.bake()` # because `.bake()` is already a method. so we allow `git.bake_()` elif name.endswith("_"): name = name[:-1] if val is None: val = get_attr("bake")(name) return val @classmethod def _extract_call_args(cls, kwargs): """takes kwargs that were passed to a command's __call__ and extracts out the special keyword arguments, we return a tuple of special keyword args, and kwargs that will go to the exec'ed command""" kwargs = kwargs.copy() call_args = {} for parg, default in cls._call_args.items(): key = "_" + parg if key in kwargs: call_args[parg] = kwargs[key] del kwargs[key] merged_args = cls._call_args.copy() merged_args.update(call_args) invalid_kwargs = special_kwarg_validator( call_args, merged_args, cls._kwarg_validators ) if invalid_kwargs: exc_msg = [] for kwarg, error_msg in invalid_kwargs: exc_msg.append(f" {kwarg!r}: {error_msg}") exc_msg = "\n".join(exc_msg) raise TypeError(f"Invalid special arguments:\n\n{exc_msg}\n") return call_args, kwargs def bake(self, args, *kwargs): """returns a new Command object after baking(freezing) the given command arguments which are used automatically when its exec'ed special keyword arguments can be temporary baked and additionally be overridden in __call__ or in subsequent bakes (basically setting defaults)""" # construct the base Command fn = type(self)(self._path) fn._partial = True call_args, kwargs = self._extract_call_args(kwargs) fn._partial_call_args.update(self._partial_call_args) fn._partial_call_args.update(call_args) fn._partial_baked_args.extend(self._partial_baked_args) sep = call_args.get("long_sep", self._call_args["long_sep"]) prefix = call_args.get("long_prefix", self._call_args["long_prefix"]) fn._partial_baked_args.extend(compile_args(args, kwargs, sep, prefix)) return fn def __str__(self): baked_args = " ".join(self._partial_baked_args) if baked_args: baked_args = " " + baked_args return self._path + baked_args def __eq__(self, other): return str(self) == str(other) def __repr__(self): return f"<Command {str(self)!r}>" def __enter__(self): self(_with=True) def __exit__(self, exc_type, exc_val, exc_tb): get_prepend_stack().pop() def __call__(self, args, *kwargs): kwargs = kwargs.copy() args = list(args) # this will hold our final command, including arguments, that will be # exec'ed cmd = [] # this will hold a complete mapping of all our special keyword arguments # and their values call_args = self.__class__._call_args.copy() # aggregate any 'with' contexts for prepend in get_prepend_stack(): pcall_args = prepend.call_args.copy() # don't pass the 'with' call arg pcall_args.pop("with", None) call_args.update(pcall_args) # we do not prepend commands used as a 'with' context as they will # be prepended to any nested commands if not kwargs.get("_with", False): cmd.extend(prepend.cmd) cmd.append(self._path) # do we have an argument pre-processor? if so, run it. we need to do # this early, so that args, kwargs are accurate preprocessor = self._partial_call_args.get("arg_preprocess", None) if preprocessor: args, kwargs = preprocessor(args, kwargs) # here we extract the special kwargs and override any # special kwargs from the possibly baked command extracted_call_args, kwargs = self._extract_call_args(kwargs) call_args.update(self._partial_call_args) call_args.update(extracted_call_args) # handle a None. this is added back only to not break the api in the # 1. version. TODO remove this in 2.0, as "ok_code", if specified, # should always be a definitive value or list of values, and None is # ambiguous if call_args["ok_code"] is None: call_args["ok_code"] = 0 if not getattr(call_args["ok_code"], "__iter__", None): call_args["ok_code"] = [call_args["ok_code"]] # determine what our real STDIN is. is it something explicitly passed into # _in? stdin = call_args["in"] # now that we have our stdin, let's figure out how we should handle it if isinstance(stdin, RunningCommand): if stdin.call_args["piped"]: stdin = stdin.process else: stdin = stdin.process._pipe_queue processed_args = compile_args( args, kwargs, call_args["long_sep"], call_args["long_prefix"] ) # makes sure our arguments are broken up correctly split_args = self._partial_baked_args + processed_args final_args = split_args cmd.extend(final_args) # if we're running in foreground mode, we need to completely bypass # launching a RunningCommand and OProc and just do a spawn if call_args["fg"]: cwd = call_args["cwd"] or os.getcwd() with pushd(cwd): if call_args["env"] is None: exit_code = os.spawnv(os.P_WAIT, cmd[0], cmd) else: exit_code = os.spawnve(os.P_WAIT, cmd[0], cmd, call_args["env"]) exc_class = get_exc_exit_code_would_raise( exit_code, call_args["ok_code"], call_args["piped"] ) if exc_class: ran = " ".join(cmd) exc = exc_class(ran, b"", b"", call_args["truncate_exc"]) raise exc return None # stdout redirection stdout = call_args["out"] if output_redirect_is_filename(stdout): stdout = open(str(stdout), "wb") # stderr redirection stderr = call_args["err"] if output_redirect_is_filename(stderr): stderr = open(str(stderr), "wb") rc = self.__class__.RunningCommandCls(cmd, call_args, stdin, stdout, stderr) if rc._spawned_and_waited and not call_args["return_cmd"]: return str(rc) else: return rc def compile_args(a, kwargs, sep, prefix): """takes args and kwargs, as they were passed into the command instance being executed with __call__, and compose them into a flat list that will eventually be fed into exec. example: with this call: sh.ls("-l", "/tmp", color="never") this function receives args = ['-l', '/tmp'] kwargs = {'color': 'never'} and produces ['-l', '/tmp', '--color=geneticnever'] """ processed_args = [] # aggregate positional args for arg in a: if isinstance(arg, (list, tuple)): if isinstance(arg, GlobResults) and not arg: arg = [arg.path] for sub_arg in arg: processed_args.append(sub_arg) elif isinstance(arg, dict): processed_args += _aggregate_keywords(arg, sep, prefix, raw=True) # see https://github.com/amoffat/sh/issues/522 elif arg is None or arg is False: pass else: processed_args.append(str(arg)) # aggregate the keyword arguments processed_args += _aggregate_keywords(kwargs, sep, prefix) return processed_args def _aggregate_keywords(keywords, sep, prefix, raw=False): """take our keyword arguments, and a separator, and compose the list of flat long (and short) arguments. example {'color': 'never', 't': True, 'something': True} with sep '=' becomes ['--color=never', '-t', '--something'] the `raw` argument indicates whether or not we should leave the argument name alone, or whether we should replace "_" with "-". if we pass in a dictionary, like this: sh.command({"some_option": 12}) then `raw` gets set to True, because we want to leave the key as-is, to produce: ['--some_option=12'] but if we just use a command's kwargs, `raw` is False, which means this: sh.command(some_option=12) becomes: ['--some-option=12'] essentially, using kwargs is a convenience, but it lacks the ability to put a '-' in the name, so we do the replacement of '_' to '-' for you. but when you really don't want that to happen, you should use a dictionary instead with the exact names you want """ processed = [] for k, maybe_list_of_v in keywords.items(): # turn our value(s) into a list of values so that we can process them # all individually under the same key list_of_v = [maybe_list_of_v] if isinstance(maybe_list_of_v, (list, tuple)): list_of_v = maybe_list_of_v for v in list_of_v: # we're passing a short arg as a kwarg, example: # cut(d="\t") if len(k) == 1: if v is not False: processed.append("-" + k) if v is not True: processed.append(str(v)) # we're doing a long arg else: if not raw: k = k.replace("_", "-") # if it's true, it has no value, just pass the name if v is True: processed.append(prefix + k) # if it's false, skip passing it elif v is False: pass # we may need to break the argument up into multiple arguments elif sep is None or sep == " ": processed.append(prefix + k) processed.append(str(v)) # otherwise just join it together into a single argument else: arg = f"{prefix}{k}{sep}{v}" processed.append(arg) return processed def _start_daemon_thread(fn, name, exc_queue, a): def wrap(rgs, *kwargs): try: fn(rgs, *kwargs) except Exception as e: exc_queue.put(e) raise thread = threading.Thread(target=wrap, name=name, args=a) thread.daemon = True thread.start() return thread def setwinsize(fd, rows_cols): """set the terminal size of a tty file descriptor. borrowed logic from pexpect.py""" rows, cols = rows_cols winsize = getattr(termios, "TIOCSWINSZ", -2146929561) s = struct.pack("HHHH", rows, cols, 0, 0) fcntl.ioctl(fd, winsize, s) def construct_streamreader_callback(process, handler): """here we're constructing a closure for our streamreader callback. this is used in the case that we pass a callback into _out or _err, meaning we want to our callback to handle each bit of output we construct the closure based on how many arguments it takes. the reason for this is to make it as easy as possible for people to use, without limiting them. a new user will assume the callback takes 1 argument (the data). as they get more advanced, they may want to terminate the process, or pass some stdin back, and will realize that they can pass a callback of more args""" # implied arg refers to the "self" that methods will pass in. we need to # account for this implied arg when figuring out what function the user # passed in based on number of args implied_arg = 0 partial_args = 0 handler_to_inspect = handler if isinstance(handler, partial): partial_args = len(handler.args) handler_to_inspect = handler.func if inspect.ismethod(handler_to_inspect): implied_arg = 1 num_args = get_num_args(handler_to_inspect) else: if inspect.isfunction(handler_to_inspect): num_args = get_num_args(handler_to_inspect) # is an object instance with __call__ method else: implied_arg = 1 num_args = get_num_args(handler_to_inspect.__call__) net_args = num_args - implied_arg - partial_args handler_args = () # just the chunk if net_args == 1: handler_args = () # chunk, stdin if net_args == 2: handler_args = (process.stdin,) # chunk, stdin, process elif net_args == 3: # notice we're only storing a weakref, to prevent cyclic references # (where the process holds a streamreader, and a streamreader holds a # handler-closure with a reference to the process handler_args = (process.stdin, weakref.ref(process)) def fn(chunk): # this is pretty ugly, but we're evaluating the process at call-time, # because it's a weakref a = handler_args if len(a) == 2: a = (handler_args[0], handler_args[1]()) return handler(chunk, a) return fn def get_exc_exit_code_would_raise(exit_code, ok_codes, sigpipe_ok): exc = None success = exit_code in ok_codes signals_that_should_throw_exception = [ sig for sig in SIGNALS_THAT_SHOULD_THROW_EXCEPTION if -sig not in ok_codes ] bad_sig = -exit_code in signals_that_should_throw_exception # if this is a piped command, SIGPIPE must be ignored by us and not raise an # exception, since it's perfectly normal for the consumer of a process's # pipe to terminate early if sigpipe_ok and -exit_code == signal.SIGPIPE: bad_sig = False success = True if not success or bad_sig: exc = get_rc_exc(exit_code) return exc def handle_process_exit_code(exit_code): """this should only ever be called once for each child process""" # if we exited from a signal, let our exit code reflect that if os.WIFSIGNALED(exit_code): exit_code = -os.WTERMSIG(exit_code) # otherwise just give us a normal exit code elif os.WIFEXITED(exit_code): exit_code = os.WEXITSTATUS(exit_code) else: raise RuntimeError("Unknown child exit status!") return exit_code def no_interrupt(syscall, args, kwargs): """a helper for making system calls immune to EINTR""" ret = None while True: try: ret = syscall(args, *kwargs) except OSError as e: if e.errno == errno.EINTR: continue else: raise else: break return ret class OProc: """this class is instantiated by RunningCommand for a command to be exec'd. it handles all the nasty business involved with correctly setting up the input/output to the child process. it gets its name for subprocess.Popen (process open) but we're calling ours OProc (open process)""" _default_window_size = (24, 80) # used in redirecting STDOUT = -1 STDERR = -2 def __init__( self, command, parent_log, cmd, stdin, stdout, stderr, call_args, pipe, process_assign_lock, ): """ cmd is the full list of arguments that will be exec'd. it includes the program name and all its arguments. stdin, stdout, stderr are what the child will use for standard input/output/err. call_args is a mapping of all the special keyword arguments to apply to the child process. """ self.command = command self.call_args = call_args # convenience ca = self.call_args if ca["uid"] is not None: if os.getuid() != 0: raise RuntimeError("UID setting requires root privileges") target_uid = ca["uid"] pwrec = pwd.getpwuid(ca["uid"]) target_gid = pwrec.pw_gid else: target_uid, target_gid = None, None # I had issues with getting 'Input/Output error reading stdin' from dd, # until I set _tty_out=False if ca["piped"]: ca["tty_out"] = False self._stdin_process = None # if the objects that we are passing to the OProc happen to be a # file-like object that is a tty, for example `sys.stdin`, then, later # on in this constructor, we're going to skip out on setting up pipes # and pseudoterminals for those endpoints stdin_is_fd_based = ob_is_fd_based(stdin) stdout_is_fd_based = ob_is_fd_based(stdout) stderr_is_fd_based = ob_is_fd_based(stderr) if isinstance(ca["tee"], (str, bool, int)) or ca["tee"] is None: tee = {ca["tee"]} else: tee = set(ca["tee"]) tee_out = TEE_STDOUT.intersection(tee) tee_err = TEE_STDERR.intersection(tee) single_tty = ca["tty_in"] and ca["tty_out"] and ca["unify_ttys"] # this logic is a little convoluted, but basically this top-level # if/else is for consolidating input and output TTYs into a single # TTY. this is the only way some secure programs like ssh will # output correctly (is if stdout and stdin are both the same TTY) if single_tty: # master_fd, slave_fd = pty.openpty() # # Anything that is written on the master end is provided to the process on # the slave end as though it was # input typed on a terminal. -"man 7 pty" # # later, in the child process, we're going to do this, so keep it in mind: # # os.dup2(self._stdin_child_fd, 0) # os.dup2(self._stdout_child_fd, 1) # os.dup2(self._stderr_child_fd, 2) self._stdin_parent_fd, self._stdin_child_fd = pty.openpty() # this makes our parent fds behave like a terminal. it says that the very # same fd that we "type" to (for stdin) is the same one that we see output # printed to (for stdout) self._stdout_parent_fd = os.dup(self._stdin_parent_fd) # this line is what makes stdout and stdin attached to the same pty. in # other words the process will write to the same underlying fd as stdout # as it uses to read from for stdin. this makes programs like ssh happy self._stdout_child_fd = os.dup(self._stdin_child_fd) self._stderr_parent_fd = os.dup(self._stdin_parent_fd) self._stderr_child_fd = os.dup(self._stdin_child_fd) # do not consolidate stdin and stdout. this is the most common use- # case else: # this check here is because we may be doing piping and so our stdin # might be an instance of OProc if isinstance(stdin, OProc) and stdin.call_args["piped"]: self._stdin_child_fd = stdin._pipe_fd self._stdin_parent_fd = None self._stdin_process = stdin elif stdin_is_fd_based: self._stdin_child_fd = os.dup(get_fileno(stdin)) self._stdin_parent_fd = None elif ca["tty_in"]: self._stdin_parent_fd, self._stdin_child_fd = pty.openpty() # tty_in=False is the default else: self._stdin_child_fd, self._stdin_parent_fd = os.pipe() if stdout_is_fd_based and not tee_out: self._stdout_child_fd = os.dup(get_fileno(stdout)) self._stdout_parent_fd = None # tty_out=True is the default elif ca["tty_out"]: self._stdout_parent_fd, self._stdout_child_fd = pty.openpty() else: self._stdout_parent_fd, self._stdout_child_fd = os.pipe() # unless STDERR is going to STDOUT, it ALWAYS needs to be a pipe, # and never a PTY. the reason for this is not totally clear to me, # but it has to do with the fact that if STDERR isn't set as the # CTTY (because STDOUT is), the STDERR buffer won't always flush # by the time the process exits, and the data will be lost. # i've only seen this on OSX. if stderr is OProc.STDOUT: # if stderr is going to stdout, but stdout is a tty or a pipe, # we should not specify a read_fd, because stdout is os.dup'ed # directly to the stdout fd (no pipe), and so stderr won't have # a slave end of a pipe either to dup if stdout_is_fd_based and not tee_out: self._stderr_parent_fd = None else: self._stderr_parent_fd = os.dup(self._stdout_parent_fd) self._stderr_child_fd = os.dup(self._stdout_child_fd) elif stderr_is_fd_based and not tee_err: self._stderr_child_fd = os.dup(get_fileno(stderr)) self._stderr_parent_fd = None else: self._stderr_parent_fd, self._stderr_child_fd = os.pipe() piped = ca["piped"] self._pipe_fd = None if piped: fd_to_use = self._stdout_parent_fd if piped == "err": fd_to_use = self._stderr_parent_fd self._pipe_fd = os.dup(fd_to_use) new_session = ca["new_session"] new_group = ca["new_group"] needs_ctty = ca["tty_in"] # if we need a controlling terminal, we have to be in a new session where we # are the session leader, otherwise we would need to take over the existing # process session, and we can't do that(?) if needs_ctty: new_session = True self.ctty = None if needs_ctty: self.ctty = os.ttyname(self._stdin_child_fd) gc_enabled = gc.isenabled() if gc_enabled: gc.disable() # for synchronizing session_pipe_read, session_pipe_write = os.pipe() exc_pipe_read, exc_pipe_write = os.pipe() # this pipe is for synchronizing with the child that the parent has # closed its in/out/err fds. this is a bug on OSX (but not linux), # where we can lose output sometimes, due to a race, if we do # os.close(self._stdout_child_fd) in the parent after the child starts # writing. if IS_MACOS: close_pipe_read, close_pipe_write = os.pipe() else: close_pipe_read, close_pipe_write = None, None # session id, group id, process id self.sid = None self.pgid = None self.pid = os.fork() # child if self.pid == 0: # pragma: no cover if IS_MACOS: os.read(close_pipe_read, 1) os.close(close_pipe_read) os.close(close_pipe_write) # this is critical # our exc_pipe_write must have CLOEXEC enabled. the reason for this is # tricky: if our child (the block we're in now), has an exception, we need # to be able to write to exc_pipe_write, so that when the parent does # os.read(exc_pipe_read), it gets our traceback. however, # os.read(exc_pipe_read) in the parent blocks, so if our child doesn't* # have an exception, and doesn't close the writing end, it hangs forever. # not good! but obviously the child can't close the writing end until it # knows it's not going to have an exception, which is impossible to know # because but what if os.execv has an exception? so the answer is CLOEXEC, # so that the writing end of the pipe gets closed upon successful exec, # and the parent reading the read end won't block (close breaks the block). flags = fcntl.fcntl(exc_pipe_write, fcntl.F_GETFD) flags \|= fcntl.FD_CLOEXEC fcntl.fcntl(exc_pipe_write, fcntl.F_SETFD, flags) try: # ignoring SIGHUP lets us persist even after the controlling terminal # is closed if ca["bg"] is True: signal.signal(signal.SIGHUP, signal.SIG_IGN) # python ignores SIGPIPE by default. we must make sure to put # this behavior back to the default for spawned processes, # otherwise SIGPIPE won't kill piped processes, which is what we # need, so that we can check the error code of the killed # process to see that SIGPIPE killed it signal.signal(signal.SIGPIPE, signal.SIG_DFL) # put our forked process in a new session? this will relinquish # any control of our inherited CTTY and also make our parent # process init if new_session: os.setsid() elif new_group: os.setpgid(0, 0) sid = os.getsid(0) pgid = os.getpgid(0) payload = (f"{sid},{pgid}").encode(DEFAULT_ENCODING) os.write(session_pipe_write, payload) if ca["tty_out"] and not stdout_is_fd_based and not single_tty: # set raw mode, so there isn't any weird translation of # newlines to \r\n and other oddities. we're not outputting # to a terminal anyways # # we HAVE to do this here, and not in the parent process, # because we have to guarantee that this is set before the # child process is run, and we can't do it twice. tty.setraw(self._stdout_child_fd) # if the parent-side fd for stdin exists, close it. the case # where it may not exist is if we're using piping if self._stdin_parent_fd: os.close(self._stdin_parent_fd) if self._stdout_parent_fd: os.close(self._stdout_parent_fd) if self._stderr_parent_fd: os.close(self._stderr_parent_fd) os.close(session_pipe_read) os.close(exc_pipe_read) cwd = ca["cwd"] if cwd: os.chdir(cwd) os.dup2(self._stdin_child_fd, 0) os.dup2(self._stdout_child_fd, 1) os.dup2(self._stderr_child_fd, 2) # set our controlling terminal, but only if we're using a tty # for stdin. it doesn't make sense to have a ctty otherwise if needs_ctty: tmp_fd = os.open(os.ttyname(0), os.O_RDWR) os.close(tmp_fd) if ca["tty_out"] and not stdout_is_fd_based: setwinsize(1, ca["tty_size"]) if ca["uid"] is not None: os.setgid(target_gid) os.setuid(target_uid) preexec_fn = ca["preexec_fn"] if callable(preexec_fn): preexec_fn() close_fds = ca["close_fds"] if ca["pass_fds"]: close_fds = True if close_fds: pass_fds = {0, 1, 2, exc_pipe_write} pass_fds.update(ca["pass_fds"]) # don't inherit file descriptors try: inherited_fds = os.listdir("/dev/fd") except OSError: # Some systems don't have /dev/fd. Raises OSError in # Python2, FileNotFoundError on Python3. The latter doesn't # exist on Python2, but inherits from IOError, which does. inherited_fds = os.listdir("/proc/self/fd") inherited_fds = {int(fd) for fd in inherited_fds} - pass_fds for fd in inherited_fds: try: os.close(fd) except OSError: pass # python=3.6, locale=c will fail test_unicode_arg if we don't # explicitly encode to bytes via our desired encoding. this does # not seem to be the case in other python versions, even if locale=c bytes_cmd = [c.encode(ca["encoding"]) for c in cmd] # actually execute the process if ca["env"] is None: os.execv(bytes_cmd[0], bytes_cmd) else: os.execve(bytes_cmd[0], bytes_cmd, ca["env"]) # we must ensure that we carefully exit the child process on # exception, otherwise the parent process code will be executed # twice on exception https://github.com/amoffat/sh/issues/202 # # if your parent process experiences an exit code 255, it is most # likely that an exception occurred between the fork of the child # and the exec. this should be reported. except Exception: # noqa: E722 # some helpful debugging tb = traceback.format_exc().encode("utf8", "ignore") try: os.write(exc_pipe_write, tb) except Exception as e: # dump to stderr if we cannot save it to exc_pipe_write sys.stderr.write(f"\nFATAL SH ERROR: {e}\n") finally: os._exit(255) # parent else: if gc_enabled: gc.enable() os.close(self._stdin_child_fd) os.close(self._stdout_child_fd) os.close(self._stderr_child_fd) # tell our child process that we've closed our write_fds, so it is # ok to proceed towards exec. see the comment where this pipe is # opened, for why this is necessary if IS_MACOS: os.close(close_pipe_read) os.write(close_pipe_write, str(1).encode(DEFAULT_ENCODING)) os.close(close_pipe_write) os.close(exc_pipe_write) fork_exc = os.read(exc_pipe_read, 1024*2) os.close(exc_pipe_read) if fork_exc: fork_exc = fork_exc.decode(DEFAULT_ENCODING) raise ForkException(fork_exc) os.close(session_pipe_write) sid, pgid = ( os.read(session_pipe_read, 1024).decode(DEFAULT_ENCODING).split(",") ) os.close(session_pipe_read) self.sid = int(sid) self.pgid = int(pgid) # used to determine what exception to raise. if our process was # killed via a timeout counter, we'll raise something different than # a SIGKILL exception self.timed_out = False self.started = time.time() self.cmd = cmd # exit code should only be manipulated from within self._wait_lock # to prevent race conditions self.exit_code = None self.stdin = stdin # this accounts for when _out is a callable that is passed stdin. in that # case, if stdin is unspecified, we must set it to a queue, so callbacks can # put things on it if callable(ca["out"]) and self.stdin is None: self.stdin = Queue() # _pipe_queue is used internally to hand off stdout from one process # to another. by default, all stdout from a process gets dumped # into this pipe queue, to be consumed in real time (hence the # thread-safe Queue), or at a potentially later time self._pipe_queue = Queue() # this is used to prevent a race condition when we're waiting for # a process to end, and the OProc's internal threads are also checking # for the processes's end self._wait_lock = threading.Lock() # these are for aggregating the stdout and stderr. we use a deque # because we don't want to overflow self._stdout = deque(maxlen=ca["internal_bufsize"]) self._stderr = deque(maxlen=ca["internal_bufsize"]) if ca["tty_in"] and not stdin_is_fd_based: setwinsize(self._stdin_parent_fd, ca["tty_size"]) self.log = parent_log.get_child("process", repr(self)) self.log.debug("started process") # disable echoing, but only if it's a tty that we created ourselves if ca["tty_in"] and not stdin_is_fd_based: attr = termios.tcgetattr(self._stdin_parent_fd) attr[3] &= ~termios.ECHO termios.tcsetattr(self._stdin_parent_fd, termios.TCSANOW, attr) # this represents the connection from a Queue object (or whatever # we're using to feed STDIN) to the process's STDIN fd self._stdin_stream = None if self._stdin_parent_fd: log = self.log.get_child("streamwriter", "stdin") self._stdin_stream = StreamWriter( log, self._stdin_parent_fd, self.stdin, ca["in_bufsize"], ca["encoding"], ca["tty_in"], ) stdout_pipe = None if pipe is OProc.STDOUT and not ca["no_pipe"]: stdout_pipe = self._pipe_queue # this represents the connection from a process's STDOUT fd to # wherever it has to go, sometimes a pipe Queue (that we will use # to pipe data to other processes), and also an internal deque # that we use to aggregate all the output save_stdout = not ca["no_out"] and (tee_out or stdout is None) pipe_out = ca["piped"] in ("out", True) pipe_err = ca["piped"] in ("err",) # if we're piping directly into another process's file descriptor, we # bypass reading from the stdout stream altogether, because we've # already hooked up this processes's stdout fd to the other # processes's stdin fd self._stdout_stream = None if not pipe_out and self._stdout_parent_fd: if callable(stdout): stdout = construct_streamreader_callback(self, stdout) self._stdout_stream = StreamReader( self.log.get_child("streamreader", "stdout"), self._stdout_parent_fd, stdout, self._stdout, ca["out_bufsize"], ca["encoding"], ca["decode_errors"], stdout_pipe, save_data=save_stdout, ) elif self._stdout_parent_fd: os.close(self._stdout_parent_fd) # if stderr is going to one place (because it's grouped with stdout, # or we're dealing with a single tty), then we don't actually need a # stream reader for stderr, because we've already set one up for # stdout above self._stderr_stream = None if ( stderr is not OProc.STDOUT and not single_tty and not pipe_err and self._stderr_parent_fd ): stderr_pipe = None if pipe is OProc.STDERR and not ca["no_pipe"]: stderr_pipe = self._pipe_queue save_stderr = not ca["no_err"] and (tee_err or stderr is None) if callable(stderr): stderr = construct_streamreader_callback(self, stderr) self._stderr_stream = StreamReader( Logger("streamreader"), self._stderr_parent_fd, stderr, self._stderr, ca["err_bufsize"], ca["encoding"], ca["decode_errors"], stderr_pipe, save_data=save_stderr, ) elif self._stderr_parent_fd: os.close(self._stderr_parent_fd) def timeout_fn(): self.timed_out = True self.signal(ca["timeout_signal"]) self._timeout_event = None self._timeout_timer = None if ca["timeout"]: self._timeout_event = threading.Event() self._timeout_timer = threading.Timer( ca["timeout"], self._timeout_event.set ) self._timeout_timer.start() # this is for cases where we know that the RunningCommand that was # launched was not .wait()ed on to complete. in those unique cases, # we allow the thread that processes output to report exceptions in # that thread. it's important that we only allow reporting of the # exception, and nothing else (like the additional stuff that # RunningCommand.wait() does), because we want the exception to be # re-raised in the future, if we DO call .wait() handle_exit_code = None if ( not self.command._spawned_and_waited and ca["bg_exc"] # we don't want background exceptions if we're doing async stuff, # because we want those to bubble up. and not ca["async"] ): def fn(exit_code): with process_assign_lock: return self.command.handle_command_exit_code(exit_code) handle_exit_code = fn self._quit_threads = threading.Event() thread_name = f"background thread for pid {self.pid}" self._bg_thread_exc_queue = Queue(1) self._background_thread = _start_daemon_thread( background_thread, thread_name, self._bg_thread_exc_queue, timeout_fn, self._timeout_event, handle_exit_code, self.is_alive, self._quit_threads, ) # start the main io threads. stdin thread is not needed if we are # connecting from another process's stdout pipe self._input_thread = None self._input_thread_exc_queue = Queue(1) if self._stdin_stream: close_before_term = not needs_ctty thread_name = f"STDIN thread for pid {self.pid}" self._input_thread = _start_daemon_thread( input_thread, thread_name, self._input_thread_exc_queue, self.log, self._stdin_stream, self.is_alive, self._quit_threads, close_before_term, ) # this event is for cases where the subprocess that we launch # launches its OWN subprocess and os.dup's the stdout/stderr fds to that # new subprocess. in that case, stdout and stderr will never EOF, # so our output_thread will never finish and will hang. this event # prevents that hanging self._stop_output_event = threading.Event() # we need to set up a callback to fire when our `output_thread` is about # to exit. this callback will set an asyncio Event, so that coroutiens can # be notified that our output is finished. # if the `sh` command was launched from within a thread (so we're not in # the main thread), then we won't have an event loop. try: loop = asyncio.get_running_loop() except RuntimeError: def output_complete(): pass else: def output_complete(): loop.call_soon_threadsafe(self.command.aio_output_complete.set) self._output_thread_exc_queue = Queue(1) thread_name = f"STDOUT/ERR thread for pid {self.pid}" self._output_thread = _start_daemon_thread( output_thread, thread_name, self._output_thread_exc_queue, self.log, self._stdout_stream, self._stderr_stream, self._timeout_event, self.is_alive, self._quit_threads, self._stop_output_event, output_complete, ) def __repr__(self): return f"<Process {self.pid} {self.cmd[:500]!r}>" def change_in_bufsize(self, buf): self._stdin_stream.stream_bufferer.change_buffering(buf) def change_out_bufsize(self, buf): self._stdout_stream.stream_bufferer.change_buffering(buf) def change_err_bufsize(self, buf): self._stderr_stream.stream_bufferer.change_buffering(buf) @property def stdout(self): return "".encode(self.call_args["encoding"]).join(self._stdout) @property def stderr(self): return "".encode(self.call_args["encoding"]).join(self._stderr) def get_pgid(self): """return the CURRENT group id of the process. this differs from self.pgid in that this reflects the current state of the process, where self.pgid is the group id at launch""" return os.getpgid(self.pid) def get_sid(self): """return the CURRENT session id of the process. this differs from self.sid in that this reflects the current state of the process, where self.sid is the session id at launch""" return os.getsid(self.pid) def signal_group(self, sig): self.log.debug("sending signal %d to group", sig) os.killpg(self.get_pgid(), sig) def signal(self, sig): self.log.debug("sending signal %d", sig) os.kill(self.pid, sig) def kill_group(self): self.log.debug("killing group") self.signal_group(signal.SIGKILL) def kill(self): self.log.debug("killing") self.signal(signal.SIGKILL) def terminate(self): self.log.debug("terminating") self.signal(signal.SIGTERM) def is_alive(self): """polls if our child process has completed, without blocking. this method has side-effects, such as setting our exit_code, if we happen to see our child exit while this is running""" if self.exit_code is not None: return False, self.exit_code # what we're doing here essentially is making sure that the main thread # (or another thread), isn't calling .wait() on the process. because # .wait() calls os.waitpid(self.pid, 0), we can't do an os.waitpid # here...because if we did, and the process exited while in this # thread, the main thread's os.waitpid(self.pid, 0) would raise OSError # (because the process ended in another thread). # # so essentially what we're doing is, using this lock, checking if # we're calling .wait(), and if we are, let .wait() get the exit code # and handle the status, otherwise let us do it. # # Using a small timeout provides backpressure against code that spams # calls to .is_alive() which may block the main thread from acquiring # the lock otherwise. acquired = self._wait_lock.acquire(timeout=0.00001) if not acquired: if self.exit_code is not None: return False, self.exit_code return True, self.exit_code witnessed_end = False try: # WNOHANG is just that...we're calling waitpid without hanging... # essentially polling the process. the return result is (0, 0) if # there's no process status, so we check that pid == self.pid below # in order to determine how to proceed pid, exit_code = no_interrupt(os.waitpid, self.pid, os.WNOHANG) if pid == self.pid: self.exit_code = handle_process_exit_code(exit_code) witnessed_end = True return False, self.exit_code # no child process except OSError: return False, self.exit_code else: return True, self.exit_code finally: self._wait_lock.release() if witnessed_end: self._process_just_ended() def _process_just_ended(self): if self._timeout_timer: self._timeout_timer.cancel() done_callback = self.call_args["done"] if done_callback: success = self.exit_code in self.call_args["ok_code"] done_callback(success, self.exit_code) # this can only be closed at the end of the process, because it might be # the CTTY, and closing it prematurely will send a SIGHUP. we also # don't want to close it if there's a self._stdin_stream, because that # is in charge of closing it also if self._stdin_parent_fd and not self._stdin_stream: os.close(self._stdin_parent_fd) def wait(self): """waits for the process to complete, handles the exit code""" self.log.debug("acquiring wait lock to wait for completion") # using the lock in a with-context blocks, which is what we want if # we're running wait() with self._wait_lock: self.log.debug("got wait lock") witnessed_end = False if self.exit_code is None: self.log.debug("exit code not set, waiting on pid") pid, exit_code = no_interrupt(os.waitpid, self.pid, 0) # blocks self.exit_code = handle_process_exit_code(exit_code) witnessed_end = True else: self.log.debug( "exit code already set (%d), no need to wait", self.exit_code ) self._process_exit_cleanup(witnessed_end=witnessed_end) return self.exit_code def _process_exit_cleanup(self, witnessed_end): self._quit_threads.set() # we may not have a thread for stdin, if the pipe has been connected # via _piped="direct" if self._input_thread: self._input_thread.join() # wait, then signal to our output thread that the child process is # done, and we should have finished reading all the stdout/stderr # data that we can by now timer = threading.Timer(2.0, self._stop_output_event.set) timer.start() # wait for our stdout and stderr streamreaders to finish reading and # aggregating the process output self._output_thread.join() timer.cancel() self._background_thread.join() if witnessed_end: self._process_just_ended() def input_thread(log, stdin, is_alive, quit_thread, close_before_term): """this is run in a separate thread. it writes into our process's stdin (a streamwriter) and waits the process to end AND everything that can be written to be written""" closed = False alive = True poller = Poller() poller.register_write(stdin) while poller and alive: changed = poller.poll(1) for fd, events in changed: if events & (POLLER_EVENT_WRITE \| POLLER_EVENT_HUP): log.debug("%r ready for more input", stdin) done = stdin.write() if done: poller.unregister(stdin) if close_before_term: stdin.close() closed = True alive, _ = is_alive() while alive: quit_thread.wait(1) alive, _ = is_alive() if not closed: stdin.close() def event_wait(ev, timeout=None): triggered = ev.wait(timeout) return triggered def background_thread( timeout_fn, timeout_event, handle_exit_code, is_alive, quit_thread ): """handles the timeout logic""" # if there's a timeout event, loop if timeout_event: while not quit_thread.is_set(): timed_out = event_wait(timeout_event, 0.1) if timed_out: timeout_fn() break # handle_exit_code will be a function ONLY if our command was NOT waited on # as part of its spawning. in other words, it's probably a background # command # # this reports the exit code exception in our thread. it's purely for the # user's awareness, and cannot be caught or used in any way, so it's ok to # suppress this during the tests if handle_exit_code and not RUNNING_TESTS: # pragma: no cover alive = True exit_code = None while alive: quit_thread.wait(1) alive, exit_code = is_alive() handle_exit_code(exit_code) def output_thread( log, stdout, stderr, timeout_event, is_alive, quit_thread, stop_output_event, output_complete, ): """this function is run in a separate thread. it reads from the process's stdout stream (a streamreader), and waits for it to claim that its done""" poller = Poller() if stdout is not None: poller.register_read(stdout) if stderr is not None: poller.register_read(stderr) # this is our poll loop for polling stdout or stderr that is ready to # be read and processed. if one of those streamreaders indicate that it # is done altogether being read from, we remove it from our list of # things to poll. when no more things are left to poll, we leave this # loop and clean up while poller: changed = no_interrupt(poller.poll, 0.1) for f, events in changed: if events & (POLLER_EVENT_READ \| POLLER_EVENT_HUP): log.debug("%r ready to be read from", f) done = f.read() if done: poller.unregister(f) elif events & POLLER_EVENT_ERROR: # for some reason, we have to just ignore streams that have had an # error. i'm not exactly sure why, but don't remove this until we # figure that out, and create a test for it pass if timeout_event and timeout_event.is_set(): break if stop_output_event.is_set(): break # we need to wait until the process is guaranteed dead before closing our # outputs, otherwise SIGPIPE alive, _ = is_alive() while alive: quit_thread.wait(1) alive, _ = is_alive() if stdout: stdout.close() if stderr: stderr.close() output_complete() class DoneReadingForever(Exception): pass class NotYetReadyToRead(Exception): pass def determine_how_to_read_input(input_obj): """given some kind of input object, return a function that knows how to read chunks of that input object. each reader function should return a chunk and raise a DoneReadingForever exception, or return None, when there's no more data to read NOTE: the function returned does not need to care much about the requested buffering type (eg, unbuffered vs newline-buffered). the StreamBufferer will take care of that. these functions just need to return a reasonably-sized chunk of data.""" if isinstance(input_obj, Queue): log_msg = "queue" get_chunk = get_queue_chunk_reader(input_obj) elif callable(input_obj): log_msg = "callable" get_chunk = get_callable_chunk_reader(input_obj) # also handles stringio elif hasattr(input_obj, "read"): log_msg = "file descriptor" get_chunk = get_file_chunk_reader(input_obj) elif isinstance(input_obj, str): log_msg = "string" get_chunk = get_iter_string_reader(input_obj) elif isinstance(input_obj, bytes): log_msg = "bytes" get_chunk = get_iter_string_reader(input_obj) elif isinstance(input_obj, GeneratorType): log_msg = "generator" get_chunk = get_iter_chunk_reader(iter(input_obj)) elif input_obj is None: log_msg = "None" def raise_(): raise DoneReadingForever get_chunk = raise_ else: try: it = iter(input_obj) except TypeError: raise Exception("unknown input object") else: log_msg = "general iterable" get_chunk = get_iter_chunk_reader(it) return get_chunk, log_msg def get_queue_chunk_reader(stdin): def fn(): try: chunk = stdin.get(True, 0.1) except Empty: raise NotYetReadyToRead if chunk is None: raise DoneReadingForever return chunk return fn def get_callable_chunk_reader(stdin): def fn(): try: data = stdin() except DoneReadingForever: raise if not data: raise DoneReadingForever return data return fn def get_iter_string_reader(stdin): """return an iterator that returns a chunk of a string every time it is called. notice that even though bufsize_type might be line buffered, we're not doing any line buffering here. that's because our StreamBufferer handles all buffering. we just need to return a reasonable-sized chunk.""" bufsize = 1024 iter_str = (stdin[i : i + bufsize] for i in range(0, len(stdin), bufsize)) return get_iter_chunk_reader(iter_str) def get_iter_chunk_reader(stdin): def fn(): try: chunk = stdin.__next__() return chunk except StopIteration: raise DoneReadingForever return fn def get_file_chunk_reader(stdin): bufsize = 1024 def fn(): # python 3. includes a fileno on stringios, but accessing it throws an # exception. that exception is how we'll know we can't do a poll on # stdin is_real_file = True try: stdin.fileno() except UnsupportedOperation: is_real_file = False # this poll is for files that may not yet be ready to read. we test # for fileno because StringIO/BytesIO cannot be used in a poll if is_real_file and hasattr(stdin, "fileno"): poller = Poller() poller.register_read(stdin) changed = poller.poll(0.1) ready = False for fd, events in changed: if events & (POLLER_EVENT_READ \| POLLER_EVENT_HUP): ready = True if not ready: raise NotYetReadyToRead chunk = stdin.read(bufsize) if not chunk: raise DoneReadingForever else: return chunk return fn def bufsize_type_to_bufsize(bf_type): """for a given bufsize type, return the actual bufsize we will read. notice that although 1 means "newline-buffered", we're reading a chunk size of 1024. this is because we have to read something. we let a StreamBufferer instance handle splitting our chunk on newlines""" # newlines if bf_type == 1: bufsize = 1024 # unbuffered elif bf_type == 0: bufsize = 1 # or buffered by specific amount else: bufsize = bf_type return bufsize class StreamWriter: """StreamWriter reads from some input (the stdin param) and writes to a fd (the stream param). the stdin may be a Queue, a callable, something with the "read" method, a string, or an iterable""" def __init__(self, log, stream, stdin, bufsize_type, encoding, tty_in): self.stream = stream self.stdin = stdin self.log = log self.encoding = encoding self.tty_in = tty_in self.stream_bufferer = StreamBufferer(bufsize_type, self.encoding) self.get_chunk, log_msg = determine_how_to_read_input(stdin) self.log.debug("parsed stdin as a %s", log_msg) def fileno(self): """defining this allows us to do poll on an instance of this class""" return self.stream def write(self): """attempt to get a chunk of data to write to our child process's stdin, then write it. the return value answers the questions "are we done writing forever?" """ # get_chunk may sometimes return bytes, and sometimes return strings # because of the nature of the different types of STDIN objects we # support try: chunk = self.get_chunk() if chunk is None: raise DoneReadingForever except DoneReadingForever: self.log.debug("done reading") if self.tty_in: # EOF time try: char = termios.tcgetattr(self.stream)[6][termios.VEOF] except: # noqa: E722 char = chr(4).encode() # normally, one EOF should be enough to signal to an program # that is read()ing, to return 0 and be on your way. however, # some programs are misbehaved, like python3.1 and python3.2. # they don't stop reading sometimes after read() returns 0. # this can be demonstrated with the following program: # # import sys # sys.stdout.write(sys.stdin.read()) # # then type 'a' followed by ctrl-d 3 times. in python # 2.6,2.7,3.3,3.4,3.5,3.6, it only takes 2 ctrl-d to terminate. # however, in python 3.1 and 3.2, it takes all 3. # # so here we send an extra EOF along, just in case. i don't # believe it can hurt anything os.write(self.stream, char) os.write(self.stream, char) return True except NotYetReadyToRead: self.log.debug("received no data") return False # if we're not bytes, make us bytes if not isinstance(chunk, bytes): chunk = chunk.encode(self.encoding) for proc_chunk in self.stream_bufferer.process(chunk): self.log.debug("got chunk size %d: %r", len(proc_chunk), proc_chunk[:30]) self.log.debug("writing chunk to process") try: os.write(self.stream, proc_chunk) except OSError: self.log.debug("OSError writing stdin chunk") return True def close(self): self.log.debug("closing, but flushing first") chunk = self.stream_bufferer.flush() self.log.debug("got chunk size %d to flush: %r", len(chunk), chunk[:30]) try: if chunk: os.write(self.stream, chunk) except OSError: pass os.close(self.stream) def determine_how_to_feed_output(handler, encoding, decode_errors): if callable(handler): process, finish = get_callback_chunk_consumer(handler, encoding, decode_errors) # in py3, this is used for bytes elif isinstance(handler, BytesIO): process, finish = get_cstringio_chunk_consumer(handler) # in py3, this is used for unicode elif isinstance(handler, StringIO): process, finish = get_stringio_chunk_consumer(handler, encoding, decode_errors) elif hasattr(handler, "write"): process, finish = get_file_chunk_consumer(handler, decode_errors) else: try: handler = int(handler) except (ValueError, TypeError): def process(chunk): return False # noqa: E731 def finish(): return None # noqa: E731 else: process, finish = get_fd_chunk_consumer(handler, decode_errors) return process, finish def get_fd_chunk_consumer(handler, decode_errors): handler = fdopen(handler, "w", closefd=False) return get_file_chunk_consumer(handler, decode_errors) def get_file_chunk_consumer(handler, decode_errors): if getattr(handler, "encoding", None): def encode(chunk): return chunk.decode(handler.encoding, decode_errors) # noqa: E731 else: def encode(chunk): return chunk # noqa: E731 if hasattr(handler, "flush"): flush = handler.flush else: def flush(): return None # noqa: E731 def process(chunk): handler.write(encode(chunk)) # we should flush on an fd. chunk is already the correctly-buffered # size, so we don't need the fd buffering as well flush() return False def finish(): flush() return process, finish def get_callback_chunk_consumer(handler, encoding, decode_errors): def process(chunk): # try to use the encoding first, if that doesn't work, send # the bytes, because it might be binary try: chunk = chunk.decode(encoding, decode_errors) except UnicodeDecodeError: pass return handler(chunk) def finish(): pass return process, finish def get_cstringio_chunk_consumer(handler): def process(chunk): handler.write(chunk) return False def finish(): pass return process, finish def get_stringio_chunk_consumer(handler, encoding, decode_errors): def process(chunk): handler.write(chunk.decode(encoding, decode_errors)) return False def finish(): pass return process, finish class StreamReader: """reads from some output (the stream) and sends what it just read to the handler.""" def __init__( self, log, stream, handler, buffer, bufsize_type, encoding, decode_errors, pipe_queue=None, save_data=True, ): self.stream = stream self.buffer = buffer self.save_data = save_data self.encoding = encoding self.decode_errors = decode_errors self.pipe_queue = None if pipe_queue: self.pipe_queue = weakref.ref(pipe_queue) self.log = log self.stream_bufferer = StreamBufferer( bufsize_type, self.encoding, self.decode_errors ) self.bufsize = bufsize_type_to_bufsize(bufsize_type) self.process_chunk, self.finish_chunk_processor = determine_how_to_feed_output( handler, encoding, decode_errors ) self.should_quit = False def fileno(self): """defining this allows us to do poll on an instance of this class""" return self.stream def close(self): chunk = self.stream_bufferer.flush() self.log.debug("got chunk size %d to flush: %r", len(chunk), chunk[:30]) if chunk: self.write_chunk(chunk) self.finish_chunk_processor() if self.pipe_queue and self.save_data: self.pipe_queue().put(None) os.close(self.stream) def write_chunk(self, chunk): # in PY3, the chunk coming in will be bytes, so keep that in mind if not self.should_quit: self.should_quit = self.process_chunk(chunk) if self.save_data: self.buffer.append(chunk) if self.pipe_queue: self.log.debug("putting chunk onto pipe: %r", chunk[:30]) self.pipe_queue().put(chunk) def read(self): # if we're PY3, we're reading bytes, otherwise we're reading # str try: chunk = no_interrupt(os.read, self.stream, self.bufsize) except OSError as e: self.log.debug("got errno %d, done reading", e.errno) return True if not chunk: self.log.debug("got no chunk, done reading") return True self.log.debug("got chunk size %d: %r", len(chunk), chunk[:30]) for chunk in self.stream_bufferer.process(chunk): self.write_chunk(chunk) class StreamBufferer: """this is used for feeding in chunks of stdout/stderr, and breaking it up into chunks that will actually be put into the internal buffers. for example, if you have two processes, one being piped to the other, and you want that, first process to feed lines of data (instead of the chunks however they come in), OProc will use an instance of this class to chop up the data and feed it as lines to be sent down the pipe""" def __init__(self, buffer_type, encoding=DEFAULT_ENCODING, decode_errors="strict"): # 0 for unbuffered, 1 for line, everything else for that amount self.type = buffer_type self.buffer = [] self.n_buffer_count = 0 self.encoding = encoding self.decode_errors = decode_errors # this is for if we change buffering types. if we change from line # buffered to unbuffered, its very possible that our self.buffer list # has data that was being saved up (while we searched for a newline). # we need to use that up, so we don't lose it self._use_up_buffer_first = False # the buffering lock is used because we might change the buffering # types from a different thread. for example, if we have a stdout # callback, we might use it to change the way stdin buffers. so we # lock self._buffering_lock = threading.RLock() self.log = Logger("stream_bufferer") def change_buffering(self, new_type): # TODO, when we stop supporting 2.6, make this a with context self.log.debug("acquiring buffering lock for changing buffering") self._buffering_lock.acquire() self.log.debug("got buffering lock for changing buffering") try: if new_type == 0: self._use_up_buffer_first = True self.type = new_type finally: self._buffering_lock.release() self.log.debug("released buffering lock for changing buffering") def process(self, chunk): # MAKE SURE THAT THE INPUT IS PY3 BYTES # THE OUTPUT IS ALWAYS PY3 BYTES # TODO, when we stop supporting 2.6, make this a with context self.log.debug( "acquiring buffering lock to process chunk (buffering: %d)", self.type ) self._buffering_lock.acquire() self.log.debug("got buffering lock to process chunk (buffering: %d)", self.type) try: # unbuffered if self.type == 0: if self._use_up_buffer_first: self._use_up_buffer_first = False to_write = self.buffer self.buffer = [] to_write.append(chunk) return to_write return [chunk] # line buffered elif self.type == 1: total_to_write = [] nl = "\n".encode(self.encoding) while True: newline = chunk.find(nl) if newline == -1: break chunk_to_write = chunk[: newline + 1] if self.buffer: chunk_to_write = b"".join(self.buffer) + chunk_to_write self.buffer = [] self.n_buffer_count = 0 chunk = chunk[newline + 1 :] total_to_write.append(chunk_to_write) if chunk: self.buffer.append(chunk) self.n_buffer_count += len(chunk) return total_to_write # N size buffered else: total_to_write = [] while True: overage = self.n_buffer_count + len(chunk) - self.type if overage >= 0: ret = "".encode(self.encoding).join(self.buffer) + chunk chunk_to_write = ret[: self.type] chunk = ret[self.type :] total_to_write.append(chunk_to_write) self.buffer = [] self.n_buffer_count = 0 else: self.buffer.append(chunk) self.n_buffer_count += len(chunk) break return total_to_write finally: self._buffering_lock.release() self.log.debug( "released buffering lock for processing chunk (buffering: %d)", self.type, ) def flush(self): self.log.debug("acquiring buffering lock for flushing buffer") self._buffering_lock.acquire() self.log.debug("got buffering lock for flushing buffer") try: ret = "".encode(self.encoding).join(self.buffer) self.buffer = [] return ret finally: self._buffering_lock.release() self.log.debug("released buffering lock for flushing buffer") def with_lock(lock): def wrapped(fn): fn = contextmanager(fn) @contextmanager def wrapped2(args, kwargs): with lock: with fn(args, kwargs): yield return wrapped2 return wrapped @with_lock(PUSHD_LOCK) def pushd(path): """pushd changes the actual working directory for the duration of the context, unlike the _cwd arg this will work with other built-ins such as sh.glob correctly""" orig_path = os.getcwd() os.chdir(path) try: yield finally: os.chdir(orig_path) @contextmanager def _args(kwargs): """allows us to temporarily override all the special keyword parameters in a with context""" kwargs_str = ",".join([f"{k}={v!r}" for k, v in kwargs.items()]) raise DeprecationWarning( f""" sh.args() has been deprecated because it was never thread safe. use the following instead: sh2 = sh({kwargs_str}) sh2.your_command() or sh2 = sh({kwargs_str}) from sh2 import your_command your_command() """ ) class Environment(dict): """this allows lookups to names that aren't found in the global scope to be searched for as a program name. for example, if "ls" isn't found in this module's scope, we consider it a system program and try to find it. we use a dict instead of just a regular object as the base class because the exec() statement used in the run_repl requires the "globals" argument to be a dictionary""" # this is a list of all of the names that the sh module exports that will # not resolve to functions. we don't want to accidentally shadow real # commands with functions/imports that we define in sh.py. for example, # "import time" may override the time system program allowlist = { "Command", "RunningCommand", "CommandNotFound", "DEFAULT_ENCODING", "DoneReadingForever", "ErrorReturnCode", "NotYetReadyToRead", "SignalException", "ForkException", "TimeoutException", "StreamBufferer", "_aggregate_keywords", "__project_url__", "__version__", "__file__", "_args", "pushd", "glob", "contrib", } def __init__(self, globs, baked_args=None): """baked_args are defaults for the 'sh' execution context. for example: tmp = sh(_out=StringIO()) 'out' would end up in here as an entry in the baked_args dict""" super(dict, self).__init__() self.globs = globs self.baked_args = baked_args or {} def __getitem__(self, k): if k == "args": # Let the deprecated '_args' context manager be imported as 'args' k = "_args" # if we're trying to import something real, see if it's in our global scope. # what defines "real" is that it's in our allowlist if k in self.allowlist: return self.globs[k] # somebody tried to be funny and do "from sh import " if k == "__all__": warnings.warn( "Cannot import from sh. Please import sh or import programs " "individually." ) return [] # check if we're naming a dynamically generated ReturnCode exception exc = get_exc_from_name(k) if exc: return exc # https://github.com/ipython/ipython/issues/2577 # https://github.com/amoffat/sh/issues/97#issuecomment-10610629 if k.startswith("__") and k.endswith("__"): raise AttributeError # is it a command? cmd = resolve_command(k, self.globs[Command.__name__], self.baked_args) if cmd: return cmd # is it a custom builtin? builtin = getattr(self, "b_" + k, None) if builtin: return builtin # how about an environment variable? # this check must come after testing if its a command, because on some # systems, there are an environment variables that can conflict with # command names. # https://github.com/amoffat/sh/issues/238 try: return os.environ[k] except KeyError: pass # nothing found, raise an exception raise CommandNotFound(k) # Methods that begin with "b_" are implementations of shell built-ins that # people are used to, but which may not have an executable equivalent. @staticmethod def b_which(program, paths=None): return _which(program, paths) class Contrib(ModuleType): # pragma: no cover @classmethod def __call__(cls, name): def wrapper1(fn): @property def cmd_getter(self): cmd = resolve_command(name, Command) if not cmd: raise CommandNotFound(name) new_cmd = fn(cmd) return new_cmd setattr(cls, name, cmd_getter) return fn return wrapper1 mod_name = __name__ + ".contrib" contrib = Contrib(mod_name) sys.modules[mod_name] = contrib @contrib("git") def git(orig): # pragma: no cover """most git commands play nicer without a TTY""" cmd = orig.bake(_tty_out=False) return cmd @contrib("bash") def bash(orig): cmd = orig.bake("-c") return cmd @contrib("sudo") def sudo(orig): # pragma: no cover """a nicer version of sudo that uses getpass to ask for a password, or allows the first argument to be a string password""" prompt = f"[sudo] password for {getpass.getuser()}: " def stdin(): pw = getpass.getpass(prompt=prompt) + "\n" yield pw def process(a, kwargs): password = kwargs.pop("password", None) if password is None: pass_getter = stdin() else: pass_getter = password.rstrip("\n") + "\n" kwargs["_in"] = pass_getter return a, kwargs cmd = orig.bake("-S", _arg_preprocess=process) return cmd @contrib("ssh") def ssh(orig): # pragma: no cover """An ssh command for automatic password login""" class SessionContent: def __init__(self): self.chars = deque(maxlen=50000) self.lines = deque(maxlen=5000) self.line_chars = [] self.last_line = "" self.cur_char = "" def append_char(self, char): if char == "\n": line = self.cur_line self.last_line = line self.lines.append(line) self.line_chars = [] else: self.line_chars.append(char) self.chars.append(char) self.cur_char = char @property def cur_line(self): line = "".join(self.line_chars) return line class SSHInteract: def __init__(self, prompt_match, pass_getter, out_handler, login_success): self.prompt_match = prompt_match self.pass_getter = pass_getter self.out_handler = out_handler self.login_success = login_success self.content = SessionContent() # some basic state self.pw_entered = False self.success = False def __call__(self, char, stdin): self.content.append_char(char) if self.pw_entered and not self.success: self.success = self.login_success(self.content) if self.success: return self.out_handler(self.content, stdin) if self.prompt_match(self.content): password = self.pass_getter() stdin.put(password + "\n") self.pw_entered = True def process(a, kwargs): real_out_handler = kwargs.pop("interact") password = kwargs.pop("password", None) login_success = kwargs.pop("login_success", None) prompt_match = kwargs.pop("prompt", None) prompt = "Please enter SSH password: " if prompt_match is None: def prompt_match(content): return content.cur_line.endswith("password: ") # noqa: E731 if password is None: def pass_getter(): return getpass.getpass(prompt=prompt) # noqa: E731 else: def pass_getter(): return password.rstrip("\n") # noqa: E731 if login_success is None: def login_success(content): return True # noqa: E731 kwargs["_out"] = SSHInteract( prompt_match, pass_getter, real_out_handler, login_success ) return a, kwargs cmd = orig.bake( _out_bufsize=0, _tty_in=True, _unify_ttys=True, _arg_preprocess=process ) return cmd def run_repl(env): # pragma: no cover print(f"\n>> sh v{__version__}\n>> https://github.com/amoffat/sh\n") while True: try: line = input("sh> ") except (ValueError, EOFError): break try: exec(compile(line, "<dummy>", "single"), env, env) except SystemExit: break except: # noqa: E722 print(traceback.format_exc()) # cleans up our last line print("") # this is a thin wrapper around THIS module (we patch sys.modules[__name__]). # this is in the case that the user does a "from sh import whatever" # in other words, they only want to import certain programs, not the whole # system PATH worth of commands. in this case, we just proxy the # import lookup to our Environment class class SelfWrapper(ModuleType): def __init__(self, self_module, baked_args=None): # this is super ugly to have to copy attributes like this, # but it seems to be the only way to make reload() behave # nicely. if i make these attributes dynamic lookups in # __getattr__, reload sometimes chokes in weird ways... super().__init__( name=getattr(self_module, "__name__", None), doc=getattr(self_module, "__doc__", None), ) for attr in ["__builtins__", "__file__", "__package__"]: setattr(self, attr, getattr(self_module, attr, None)) # python 3.2 (2.7 and 3.3 work fine) breaks on osx (not ubuntu) # if we set this to None. and 3.3 needs a value for __path__ self.__path__ = [] self.__self_module = self_module # Copy the Command class and add any baked call kwargs to it command_cls = Command cls_attrs = command_cls.__dict__.copy() cls_attrs.pop("__dict__", None) if baked_args: call_args, _ = command_cls._extract_call_args(baked_args) cls_attrs["_call_args"] = cls_attrs["_call_args"].copy() cls_attrs["_call_args"].update(call_args) globs = globals().copy() globs[command_cls.__name__] = type( command_cls.__name__, command_cls.__bases__, cls_attrs ) self.__env = Environment(globs, baked_args=baked_args) def __getattr__(self, name): return self.__env[name] def bake(self, **kwargs): baked_args = self.__env.baked_args.copy() baked_args.update(kwargs) new_sh = self.__class__(self.__self_module, baked_args) return new_sh if __name__ == "__main__": # pragma: no cover # we're being run as a stand-alone script env = Environment(globals()) run_repl(env) else: # we're being imported from somewhere sys.modules[__name__] = SelfWrapper(sys.modules[__name__])	124.14 KB	2025-05-25 12:26
📁 shellingham	-	2025-05-25 12:26
📁 shellingham-1.5.4.dist-info	-	2025-05-25 12:26
📁 six-1.17.0.dist-info	-	2025-05-25 12:26
📄 six.py # Copyright (c) 2010-2024 Benjamin Peterson # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in all # copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. """Utilities for writing code that runs on Python 2 and 3""" from __future__ import absolute_import import functools import itertools import operator import sys import types __author__ = "Benjamin Peterson <benjamin@python.org>" __version__ = "1.17.0" # Useful for very coarse version differentiation. PY2 = sys.version_info[0] == 2 PY3 = sys.version_info[0] == 3 PY34 = sys.version_info[0:2] >= (3, 4) if PY3: string_types = str, integer_types = int, class_types = type, text_type = str binary_type = bytes MAXSIZE = sys.maxsize else: string_types = basestring, integer_types = (int, long) class_types = (type, types.ClassType) text_type = unicode binary_type = str if sys.platform.startswith("java"): # Jython always uses 32 bits. MAXSIZE = int((1 << 31) - 1) else: # It's possible to have sizeof(long) != sizeof(Py_ssize_t). class X(object): def __len__(self): return 1 << 31 try: len(X()) except OverflowError: # 32-bit MAXSIZE = int((1 << 31) - 1) else: # 64-bit MAXSIZE = int((1 << 63) - 1) del X if PY34: from importlib.util import spec_from_loader else: spec_from_loader = None def _add_doc(func, doc): """Add documentation to a function.""" func.__doc__ = doc def _import_module(name): """Import module, returning the module after the last dot.""" __import__(name) return sys.modules[name] class _LazyDescr(object): def __init__(self, name): self.name = name def __get__(self, obj, tp): result = self._resolve() setattr(obj, self.name, result) # Invokes __set__. try: # This is a bit ugly, but it avoids running this again by # removing this descriptor. delattr(obj.__class__, self.name) except AttributeError: pass return result class MovedModule(_LazyDescr): def __init__(self, name, old, new=None): super(MovedModule, self).__init__(name) if PY3: if new is None: new = name self.mod = new else: self.mod = old def _resolve(self): return _import_module(self.mod) def __getattr__(self, attr): _module = self._resolve() value = getattr(_module, attr) setattr(self, attr, value) return value class _LazyModule(types.ModuleType): def __init__(self, name): super(_LazyModule, self).__init__(name) self.__doc__ = self.__class__.__doc__ def __dir__(self): attrs = ["__doc__", "__name__"] attrs += [attr.name for attr in self._moved_attributes] return attrs # Subclasses should override this _moved_attributes = [] class MovedAttribute(_LazyDescr): def __init__(self, name, old_mod, new_mod, old_attr=None, new_attr=None): super(MovedAttribute, self).__init__(name) if PY3: if new_mod is None: new_mod = name self.mod = new_mod if new_attr is None: if old_attr is None: new_attr = name else: new_attr = old_attr self.attr = new_attr else: self.mod = old_mod if old_attr is None: old_attr = name self.attr = old_attr def _resolve(self): module = _import_module(self.mod) return getattr(module, self.attr) class _SixMetaPathImporter(object): """ A meta path importer to import six.moves and its submodules. This class implements a PEP302 finder and loader. It should be compatible with Python 2.5 and all existing versions of Python3 """ def __init__(self, six_module_name): self.name = six_module_name self.known_modules = {} def _add_module(self, mod, fullnames): for fullname in fullnames: self.known_modules[self.name + "." + fullname] = mod def _get_module(self, fullname): return self.known_modules[self.name + "." + fullname] def find_module(self, fullname, path=None): if fullname in self.known_modules: return self return None def find_spec(self, fullname, path, target=None): if fullname in self.known_modules: return spec_from_loader(fullname, self) return None def __get_module(self, fullname): try: return self.known_modules[fullname] except KeyError: raise ImportError("This loader does not know module " + fullname) def load_module(self, fullname): try: # in case of a reload return sys.modules[fullname] except KeyError: pass mod = self.__get_module(fullname) if isinstance(mod, MovedModule): mod = mod._resolve() else: mod.__loader__ = self sys.modules[fullname] = mod return mod def is_package(self, fullname): """ Return true, if the named module is a package. We need this method to get correct spec objects with Python 3.4 (see PEP451) """ return hasattr(self.__get_module(fullname), "__path__") def get_code(self, fullname): """Return None Required, if is_package is implemented""" self.__get_module(fullname) # eventually raises ImportError return None get_source = get_code # same as get_code def create_module(self, spec): return self.load_module(spec.name) def exec_module(self, module): pass _importer = _SixMetaPathImporter(__name__) class _MovedItems(_LazyModule): """Lazy loading of moved objects""" __path__ = [] # mark as package _moved_attributes = [ MovedAttribute("cStringIO", "cStringIO", "io", "StringIO"), MovedAttribute("filter", "itertools", "builtins", "ifilter", "filter"), MovedAttribute("filterfalse", "itertools", "itertools", "ifilterfalse", "filterfalse"), MovedAttribute("input", "__builtin__", "builtins", "raw_input", "input"), MovedAttribute("intern", "__builtin__", "sys"), MovedAttribute("map", "itertools", "builtins", "imap", "map"), MovedAttribute("getcwd", "os", "os", "getcwdu", "getcwd"), MovedAttribute("getcwdb", "os", "os", "getcwd", "getcwdb"), MovedAttribute("getoutput", "commands", "subprocess"), MovedAttribute("range", "__builtin__", "builtins", "xrange", "range"), MovedAttribute("reload_module", "__builtin__", "importlib" if PY34 else "imp", "reload"), MovedAttribute("reduce", "__builtin__", "functools"), MovedAttribute("shlex_quote", "pipes", "shlex", "quote"), MovedAttribute("StringIO", "StringIO", "io"), MovedAttribute("UserDict", "UserDict", "collections", "IterableUserDict", "UserDict"), MovedAttribute("UserList", "UserList", "collections"), MovedAttribute("UserString", "UserString", "collections"), MovedAttribute("xrange", "__builtin__", "builtins", "xrange", "range"), MovedAttribute("zip", "itertools", "builtins", "izip", "zip"), MovedAttribute("zip_longest", "itertools", "itertools", "izip_longest", "zip_longest"), MovedModule("builtins", "__builtin__"), MovedModule("configparser", "ConfigParser"), MovedModule("collections_abc", "collections", "collections.abc" if sys.version_info >= (3, 3) else "collections"), MovedModule("copyreg", "copy_reg"), MovedModule("dbm_gnu", "gdbm", "dbm.gnu"), MovedModule("dbm_ndbm", "dbm", "dbm.ndbm"), MovedModule("_dummy_thread", "dummy_thread", "_dummy_thread" if sys.version_info < (3, 9) else "_thread"), MovedModule("http_cookiejar", "cookielib", "http.cookiejar"), MovedModule("http_cookies", "Cookie", "http.cookies"), MovedModule("html_entities", "htmlentitydefs", "html.entities"), MovedModule("html_parser", "HTMLParser", "html.parser"), MovedModule("http_client", "httplib", "http.client"), MovedModule("email_mime_base", "email.MIMEBase", "email.mime.base"), MovedModule("email_mime_image", "email.MIMEImage", "email.mime.image"), MovedModule("email_mime_multipart", "email.MIMEMultipart", "email.mime.multipart"), MovedModule("email_mime_nonmultipart", "email.MIMENonMultipart", "email.mime.nonmultipart"), MovedModule("email_mime_text", "email.MIMEText", "email.mime.text"), MovedModule("BaseHTTPServer", "BaseHTTPServer", "http.server"), MovedModule("CGIHTTPServer", "CGIHTTPServer", "http.server"), MovedModule("SimpleHTTPServer", "SimpleHTTPServer", "http.server"), MovedModule("cPickle", "cPickle", "pickle"), MovedModule("queue", "Queue"), MovedModule("reprlib", "repr"), MovedModule("socketserver", "SocketServer"), MovedModule("_thread", "thread", "_thread"), MovedModule("tkinter", "Tkinter"), MovedModule("tkinter_dialog", "Dialog", "tkinter.dialog"), MovedModule("tkinter_filedialog", "FileDialog", "tkinter.filedialog"), MovedModule("tkinter_scrolledtext", "ScrolledText", "tkinter.scrolledtext"), MovedModule("tkinter_simpledialog", "SimpleDialog", "tkinter.simpledialog"), MovedModule("tkinter_tix", "Tix", "tkinter.tix"), MovedModule("tkinter_ttk", "ttk", "tkinter.ttk"), MovedModule("tkinter_constants", "Tkconstants", "tkinter.constants"), MovedModule("tkinter_dnd", "Tkdnd", "tkinter.dnd"), MovedModule("tkinter_colorchooser", "tkColorChooser", "tkinter.colorchooser"), MovedModule("tkinter_commondialog", "tkCommonDialog", "tkinter.commondialog"), MovedModule("tkinter_tkfiledialog", "tkFileDialog", "tkinter.filedialog"), MovedModule("tkinter_font", "tkFont", "tkinter.font"), MovedModule("tkinter_messagebox", "tkMessageBox", "tkinter.messagebox"), MovedModule("tkinter_tksimpledialog", "tkSimpleDialog", "tkinter.simpledialog"), MovedModule("urllib_parse", __name__ + ".moves.urllib_parse", "urllib.parse"), MovedModule("urllib_error", __name__ + ".moves.urllib_error", "urllib.error"), MovedModule("urllib", __name__ + ".moves.urllib", __name__ + ".moves.urllib"), MovedModule("urllib_robotparser", "robotparser", "urllib.robotparser"), MovedModule("xmlrpc_client", "xmlrpclib", "xmlrpc.client"), MovedModule("xmlrpc_server", "SimpleXMLRPCServer", "xmlrpc.server"), ] # Add windows specific modules. if sys.platform == "win32": _moved_attributes += [ MovedModule("winreg", "_winreg"), ] for attr in _moved_attributes: setattr(_MovedItems, attr.name, attr) if isinstance(attr, MovedModule): _importer._add_module(attr, "moves." + attr.name) del attr _MovedItems._moved_attributes = _moved_attributes moves = _MovedItems(__name__ + ".moves") _importer._add_module(moves, "moves") class Module_six_moves_urllib_parse(_LazyModule): """Lazy loading of moved objects in six.moves.urllib_parse""" _urllib_parse_moved_attributes = [ MovedAttribute("ParseResult", "urlparse", "urllib.parse"), MovedAttribute("SplitResult", "urlparse", "urllib.parse"), MovedAttribute("parse_qs", "urlparse", "urllib.parse"), MovedAttribute("parse_qsl", "urlparse", "urllib.parse"), MovedAttribute("urldefrag", "urlparse", "urllib.parse"), MovedAttribute("urljoin", "urlparse", "urllib.parse"), MovedAttribute("urlparse", "urlparse", "urllib.parse"), MovedAttribute("urlsplit", "urlparse", "urllib.parse"), MovedAttribute("urlunparse", "urlparse", "urllib.parse"), MovedAttribute("urlunsplit", "urlparse", "urllib.parse"), MovedAttribute("quote", "urllib", "urllib.parse"), MovedAttribute("quote_plus", "urllib", "urllib.parse"), MovedAttribute("unquote", "urllib", "urllib.parse"), MovedAttribute("unquote_plus", "urllib", "urllib.parse"), MovedAttribute("unquote_to_bytes", "urllib", "urllib.parse", "unquote", "unquote_to_bytes"), MovedAttribute("urlencode", "urllib", "urllib.parse"), MovedAttribute("splitquery", "urllib", "urllib.parse"), MovedAttribute("splittag", "urllib", "urllib.parse"), MovedAttribute("splituser", "urllib", "urllib.parse"), MovedAttribute("splitvalue", "urllib", "urllib.parse"), MovedAttribute("uses_fragment", "urlparse", "urllib.parse"), MovedAttribute("uses_netloc", "urlparse", "urllib.parse"), MovedAttribute("uses_params", "urlparse", "urllib.parse"), MovedAttribute("uses_query", "urlparse", "urllib.parse"), MovedAttribute("uses_relative", "urlparse", "urllib.parse"), ] for attr in _urllib_parse_moved_attributes: setattr(Module_six_moves_urllib_parse, attr.name, attr) del attr Module_six_moves_urllib_parse._moved_attributes = _urllib_parse_moved_attributes _importer._add_module(Module_six_moves_urllib_parse(__name__ + ".moves.urllib_parse"), "moves.urllib_parse", "moves.urllib.parse") class Module_six_moves_urllib_error(_LazyModule): """Lazy loading of moved objects in six.moves.urllib_error""" _urllib_error_moved_attributes = [ MovedAttribute("URLError", "urllib2", "urllib.error"), MovedAttribute("HTTPError", "urllib2", "urllib.error"), MovedAttribute("ContentTooShortError", "urllib", "urllib.error"), ] for attr in _urllib_error_moved_attributes: setattr(Module_six_moves_urllib_error, attr.name, attr) del attr Module_six_moves_urllib_error._moved_attributes = _urllib_error_moved_attributes _importer._add_module(Module_six_moves_urllib_error(__name__ + ".moves.urllib.error"), "moves.urllib_error", "moves.urllib.error") class Module_six_moves_urllib_request(_LazyModule): """Lazy loading of moved objects in six.moves.urllib_request""" _urllib_request_moved_attributes = [ MovedAttribute("urlopen", "urllib2", "urllib.request"), MovedAttribute("install_opener", "urllib2", "urllib.request"), MovedAttribute("build_opener", "urllib2", "urllib.request"), MovedAttribute("pathname2url", "urllib", "urllib.request"), MovedAttribute("url2pathname", "urllib", "urllib.request"), MovedAttribute("getproxies", "urllib", "urllib.request"), MovedAttribute("Request", "urllib2", "urllib.request"), MovedAttribute("OpenerDirector", "urllib2", "urllib.request"), MovedAttribute("HTTPDefaultErrorHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPRedirectHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPCookieProcessor", "urllib2", "urllib.request"), MovedAttribute("ProxyHandler", "urllib2", "urllib.request"), MovedAttribute("BaseHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPPasswordMgr", "urllib2", "urllib.request"), MovedAttribute("HTTPPasswordMgrWithDefaultRealm", "urllib2", "urllib.request"), MovedAttribute("AbstractBasicAuthHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPBasicAuthHandler", "urllib2", "urllib.request"), MovedAttribute("ProxyBasicAuthHandler", "urllib2", "urllib.request"), MovedAttribute("AbstractDigestAuthHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPDigestAuthHandler", "urllib2", "urllib.request"), MovedAttribute("ProxyDigestAuthHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPSHandler", "urllib2", "urllib.request"), MovedAttribute("FileHandler", "urllib2", "urllib.request"), MovedAttribute("FTPHandler", "urllib2", "urllib.request"), MovedAttribute("CacheFTPHandler", "urllib2", "urllib.request"), MovedAttribute("UnknownHandler", "urllib2", "urllib.request"), MovedAttribute("HTTPErrorProcessor", "urllib2", "urllib.request"), MovedAttribute("urlretrieve", "urllib", "urllib.request"), MovedAttribute("urlcleanup", "urllib", "urllib.request"), MovedAttribute("proxy_bypass", "urllib", "urllib.request"), MovedAttribute("parse_http_list", "urllib2", "urllib.request"), MovedAttribute("parse_keqv_list", "urllib2", "urllib.request"), ] if sys.version_info[:2] < (3, 14): _urllib_request_moved_attributes.extend( [ MovedAttribute("URLopener", "urllib", "urllib.request"), MovedAttribute("FancyURLopener", "urllib", "urllib.request"), ] ) for attr in _urllib_request_moved_attributes: setattr(Module_six_moves_urllib_request, attr.name, attr) del attr Module_six_moves_urllib_request._moved_attributes = _urllib_request_moved_attributes _importer._add_module(Module_six_moves_urllib_request(__name__ + ".moves.urllib.request"), "moves.urllib_request", "moves.urllib.request") class Module_six_moves_urllib_response(_LazyModule): """Lazy loading of moved objects in six.moves.urllib_response""" _urllib_response_moved_attributes = [ MovedAttribute("addbase", "urllib", "urllib.response"), MovedAttribute("addclosehook", "urllib", "urllib.response"), MovedAttribute("addinfo", "urllib", "urllib.response"), MovedAttribute("addinfourl", "urllib", "urllib.response"), ] for attr in _urllib_response_moved_attributes: setattr(Module_six_moves_urllib_response, attr.name, attr) del attr Module_six_moves_urllib_response._moved_attributes = _urllib_response_moved_attributes _importer._add_module(Module_six_moves_urllib_response(__name__ + ".moves.urllib.response"), "moves.urllib_response", "moves.urllib.response") class Module_six_moves_urllib_robotparser(_LazyModule): """Lazy loading of moved objects in six.moves.urllib_robotparser""" _urllib_robotparser_moved_attributes = [ MovedAttribute("RobotFileParser", "robotparser", "urllib.robotparser"), ] for attr in _urllib_robotparser_moved_attributes: setattr(Module_six_moves_urllib_robotparser, attr.name, attr) del attr Module_six_moves_urllib_robotparser._moved_attributes = _urllib_robotparser_moved_attributes _importer._add_module(Module_six_moves_urllib_robotparser(__name__ + ".moves.urllib.robotparser"), "moves.urllib_robotparser", "moves.urllib.robotparser") class Module_six_moves_urllib(types.ModuleType): """Create a six.moves.urllib namespace that resembles the Python 3 namespace""" __path__ = [] # mark as package parse = _importer._get_module("moves.urllib_parse") error = _importer._get_module("moves.urllib_error") request = _importer._get_module("moves.urllib_request") response = _importer._get_module("moves.urllib_response") robotparser = _importer._get_module("moves.urllib_robotparser") def __dir__(self): return ['parse', 'error', 'request', 'response', 'robotparser'] _importer._add_module(Module_six_moves_urllib(__name__ + ".moves.urllib"), "moves.urllib") def add_move(move): """Add an item to six.moves.""" setattr(_MovedItems, move.name, move) def remove_move(name): """Remove item from six.moves.""" try: delattr(_MovedItems, name) except AttributeError: try: del moves.__dict__[name] except KeyError: raise AttributeError("no such move, %r" % (name,)) if PY3: _meth_func = "__func__" _meth_self = "__self__" _func_closure = "__closure__" _func_code = "__code__" _func_defaults = "__defaults__" _func_globals = "__globals__" else: _meth_func = "im_func" _meth_self = "im_self" _func_closure = "func_closure" _func_code = "func_code" _func_defaults = "func_defaults" _func_globals = "func_globals" try: advance_iterator = next except NameError: def advance_iterator(it): return it.next() next = advance_iterator try: callable = callable except NameError: def callable(obj): return any("__call__" in klass.__dict__ for klass in type(obj).__mro__) if PY3: def get_unbound_function(unbound): return unbound create_bound_method = types.MethodType def create_unbound_method(func, cls): return func Iterator = object else: def get_unbound_function(unbound): return unbound.im_func def create_bound_method(func, obj): return types.MethodType(func, obj, obj.__class__) def create_unbound_method(func, cls): return types.MethodType(func, None, cls) class Iterator(object): def next(self): return type(self).__next__(self) callable = callable _add_doc(get_unbound_function, """Get the function out of a possibly unbound function""") get_method_function = operator.attrgetter(_meth_func) get_method_self = operator.attrgetter(_meth_self) get_function_closure = operator.attrgetter(_func_closure) get_function_code = operator.attrgetter(_func_code) get_function_defaults = operator.attrgetter(_func_defaults) get_function_globals = operator.attrgetter(_func_globals) if PY3: def iterkeys(d, kw): return iter(d.keys(kw)) def itervalues(d, kw): return iter(d.values(kw)) def iteritems(d, kw): return iter(d.items(kw)) def iterlists(d, kw): return iter(d.lists(kw)) viewkeys = operator.methodcaller("keys") viewvalues = operator.methodcaller("values") viewitems = operator.methodcaller("items") else: def iterkeys(d, kw): return d.iterkeys(kw) def itervalues(d, kw): return d.itervalues(kw) def iteritems(d, kw): return d.iteritems(kw) def iterlists(d, kw): return d.iterlists(kw) viewkeys = operator.methodcaller("viewkeys") viewvalues = operator.methodcaller("viewvalues") viewitems = operator.methodcaller("viewitems") _add_doc(iterkeys, "Return an iterator over the keys of a dictionary.") _add_doc(itervalues, "Return an iterator over the values of a dictionary.") _add_doc(iteritems, "Return an iterator over the (key, value) pairs of a dictionary.") _add_doc(iterlists, "Return an iterator over the (key, [values]) pairs of a dictionary.") if PY3: def b(s): return s.encode("latin-1") def u(s): return s unichr = chr import struct int2byte = struct.Struct(">B").pack del struct byte2int = operator.itemgetter(0) indexbytes = operator.getitem iterbytes = iter import io StringIO = io.StringIO BytesIO = io.BytesIO del io _assertCountEqual = "assertCountEqual" if sys.version_info[1] <= 1: _assertRaisesRegex = "assertRaisesRegexp" _assertRegex = "assertRegexpMatches" _assertNotRegex = "assertNotRegexpMatches" else: _assertRaisesRegex = "assertRaisesRegex" _assertRegex = "assertRegex" _assertNotRegex = "assertNotRegex" else: def b(s): return s # Workaround for standalone backslash def u(s): return unicode(s.replace(r'\\', r'\\\\'), "unicode_escape") unichr = unichr int2byte = chr def byte2int(bs): return ord(bs[0]) def indexbytes(buf, i): return ord(buf[i]) iterbytes = functools.partial(itertools.imap, ord) import StringIO StringIO = BytesIO = StringIO.StringIO _assertCountEqual = "assertItemsEqual" _assertRaisesRegex = "assertRaisesRegexp" _assertRegex = "assertRegexpMatches" _assertNotRegex = "assertNotRegexpMatches" _add_doc(b, """Byte literal""") _add_doc(u, """Text literal""") def assertCountEqual(self, args, *kwargs): return getattr(self, _assertCountEqual)(args, *kwargs) def assertRaisesRegex(self, args, *kwargs): return getattr(self, _assertRaisesRegex)(args, *kwargs) def assertRegex(self, args, *kwargs): return getattr(self, _assertRegex)(args, *kwargs) def assertNotRegex(self, args, *kwargs): return getattr(self, _assertNotRegex)(args, *kwargs) if PY3: exec_ = getattr(moves.builtins, "exec") def reraise(tp, value, tb=None): try: if value is None: value = tp() if value.__traceback__ is not tb: raise value.with_traceback(tb) raise value finally: value = None tb = None else: def exec_(_code_, _globs_=None, _locs_=None): """Execute code in a namespace.""" if _globs_ is None: frame = sys._getframe(1) _globs_ = frame.f_globals if _locs_ is None: _locs_ = frame.f_locals del frame elif _locs_ is None: _locs_ = _globs_ exec("""exec _code_ in _globs_, _locs_""") exec_("""def reraise(tp, value, tb=None): try: raise tp, value, tb finally: tb = None """) if sys.version_info[:2] > (3,): exec_("""def raise_from(value, from_value): try: raise value from from_value finally: value = None """) else: def raise_from(value, from_value): raise value print_ = getattr(moves.builtins, "print", None) if print_ is None: def print_(args, *kwargs): """The new-style print function for Python 2.4 and 2.5.""" fp = kwargs.pop("file", sys.stdout) if fp is None: return def write(data): if not isinstance(data, basestring): data = str(data) # If the file has an encoding, encode unicode with it. if (isinstance(fp, file) and isinstance(data, unicode) and fp.encoding is not None): errors = getattr(fp, "errors", None) if errors is None: errors = "strict" data = data.encode(fp.encoding, errors) fp.write(data) want_unicode = False sep = kwargs.pop("sep", None) if sep is not None: if isinstance(sep, unicode): want_unicode = True elif not isinstance(sep, str): raise TypeError("sep must be None or a string") end = kwargs.pop("end", None) if end is not None: if isinstance(end, unicode): want_unicode = True elif not isinstance(end, str): raise TypeError("end must be None or a string") if kwargs: raise TypeError("invalid keyword arguments to print()") if not want_unicode: for arg in args: if isinstance(arg, unicode): want_unicode = True break if want_unicode: newline = unicode("\n") space = unicode(" ") else: newline = "\n" space = " " if sep is None: sep = space if end is None: end = newline for i, arg in enumerate(args): if i: write(sep) write(arg) write(end) if sys.version_info[:2] < (3, 3): _print = print_ def print_(args, *kwargs): fp = kwargs.get("file", sys.stdout) flush = kwargs.pop("flush", False) _print(args, *kwargs) if flush and fp is not None: fp.flush() _add_doc(reraise, """Reraise an exception.""") if sys.version_info[0:2] < (3, 4): # This does exactly the same what the :func:`py3:functools.update_wrapper` # function does on Python versions after 3.2. It sets the ``__wrapped__`` # attribute on ``wrapper`` object and it doesn't raise an error if any of # the attributes mentioned in ``assigned`` and ``updated`` are missing on # ``wrapped`` object. def _update_wrapper(wrapper, wrapped, assigned=functools.WRAPPER_ASSIGNMENTS, updated=functools.WRAPPER_UPDATES): for attr in assigned: try: value = getattr(wrapped, attr) except AttributeError: continue else: setattr(wrapper, attr, value) for attr in updated: getattr(wrapper, attr).update(getattr(wrapped, attr, {})) wrapper.__wrapped__ = wrapped return wrapper _update_wrapper.__doc__ = functools.update_wrapper.__doc__ def wraps(wrapped, assigned=functools.WRAPPER_ASSIGNMENTS, updated=functools.WRAPPER_UPDATES): return functools.partial(_update_wrapper, wrapped=wrapped, assigned=assigned, updated=updated) wraps.__doc__ = functools.wraps.__doc__ else: wraps = functools.wraps def with_metaclass(meta, bases): """Create a base class with a metaclass.""" # This requires a bit of explanation: the basic idea is to make a dummy # metaclass for one level of class instantiation that replaces itself with # the actual metaclass. class metaclass(type): def __new__(cls, name, this_bases, d): if sys.version_info[:2] >= (3, 7): # This version introduced PEP 560 that requires a bit # of extra care (we mimic what is done by __build_class__). resolved_bases = types.resolve_bases(bases) if resolved_bases is not bases: d['__orig_bases__'] = bases else: resolved_bases = bases return meta(name, resolved_bases, d) @classmethod def __prepare__(cls, name, this_bases): return meta.__prepare__(name, bases) return type.__new__(metaclass, 'temporary_class', (), {}) def add_metaclass(metaclass): """Class decorator for creating a class with a metaclass.""" def wrapper(cls): orig_vars = cls.__dict__.copy() slots = orig_vars.get('__slots__') if slots is not None: if isinstance(slots, str): slots = [slots] for slots_var in slots: orig_vars.pop(slots_var) orig_vars.pop('__dict__', None) orig_vars.pop('__weakref__', None) if hasattr(cls, '__qualname__'): orig_vars['__qualname__'] = cls.__qualname__ return metaclass(cls.__name__, cls.__bases__, orig_vars) return wrapper def ensure_binary(s, encoding='utf-8', errors='strict'): """Coerce s to six.binary_type. For Python 2: - `unicode` -> encoded to `str` - `str` -> `str` For Python 3: - `str` -> encoded to `bytes` - `bytes` -> `bytes` """ if isinstance(s, binary_type): return s if isinstance(s, text_type): return s.encode(encoding, errors) raise TypeError("not expecting type '%s'" % type(s)) def ensure_str(s, encoding='utf-8', errors='strict'): """Coerce s to `str`. For Python 2: - `unicode` -> encoded to `str` - `str` -> `str` For Python 3: - `str` -> `str` - `bytes` -> decoded to `str` """ # Optimization: Fast return for the common case. if type(s) is str: return s if PY2 and isinstance(s, text_type): return s.encode(encoding, errors) elif PY3 and isinstance(s, binary_type): return s.decode(encoding, errors) elif not isinstance(s, (text_type, binary_type)): raise TypeError("not expecting type '%s'" % type(s)) return s def ensure_text(s, encoding='utf-8', errors='strict'): """Coerce s to six.text_type. For Python 2: - `unicode` -> `unicode` - `str` -> `unicode` For Python 3: - `str` -> `str` - `bytes` -> decoded to `str` """ if isinstance(s, binary_type): return s.decode(encoding, errors) elif isinstance(s, text_type): return s else: raise TypeError("not expecting type '%s'" % type(s)) def python_2_unicode_compatible(klass): """ A class decorator that defines __unicode__ and __str__ methods under Python 2. Under Python 3 it does nothing. To support Python 2 and 3 with a single code base, define a __str__ method returning text and apply this decorator to the class. """ if PY2: if '__str__' not in klass.__dict__: raise ValueError("@python_2_unicode_compatible cannot be applied " "to %s because it doesn't define __str__()." % klass.__name__) klass.__unicode__ = klass.__str__ klass.__str__ = lambda self: self.__unicode__().encode('utf-8') return klass # Complete the moves implementation. # This code is at the end of this module to speed up module loading. # Turn this module into a package. __path__ = [] # required for PEP 302 and PEP 451 __package__ = __name__ # see PEP 366 @ReservedAssignment if globals().get("__spec__") is not None: __spec__.submodule_search_locations = [] # PEP 451 @UndefinedVariable # Remove other six meta path importers, since they cause problems. This can # happen if six is removed from sys.modules and then reloaded. (Setuptools does # this for some reason.) if sys.meta_path: for i, importer in enumerate(sys.meta_path): # Here's some real nastiness: Another "instance" of the six module might # be floating around. Therefore, we can't use isinstance() to check for # the six meta path importer, since the other six instance will have # inserted an importer with different class. if (type(importer).__name__ == "_SixMetaPathImporter" and importer.name == __name__): del sys.meta_path[i] break del i, importer # Finally, add the importer to the meta path import hook. sys.meta_path.append(_importer)	33.89 KB	2025-05-25 12:26
📁 tabulate	-	2025-05-25 12:26
📁 tabulate-0.9.0.dist-info	-	2025-05-25 12:26
📁 tests	-	2025-05-25 12:26
📁 typer	-	2025-05-25 12:26
📁 typer-0.15.4.dist-info	-	2025-05-25 12:26
📁 typing_extensions-4.13.2.dist-info	-	2025-05-25 12:26
📄 typing_extensions.py import abc import builtins import collections import collections.abc import contextlib import enum import functools import inspect import keyword import operator import sys import types as _types import typing import warnings __all__ = [ # Super-special typing primitives. 'Any', 'ClassVar', 'Concatenate', 'Final', 'LiteralString', 'ParamSpec', 'ParamSpecArgs', 'ParamSpecKwargs', 'Self', 'Type', 'TypeVar', 'TypeVarTuple', 'Unpack', # ABCs (from collections.abc). 'Awaitable', 'AsyncIterator', 'AsyncIterable', 'Coroutine', 'AsyncGenerator', 'AsyncContextManager', 'Buffer', 'ChainMap', # Concrete collection types. 'ContextManager', 'Counter', 'Deque', 'DefaultDict', 'NamedTuple', 'OrderedDict', 'TypedDict', # Structural checks, a.k.a. protocols. 'SupportsAbs', 'SupportsBytes', 'SupportsComplex', 'SupportsFloat', 'SupportsIndex', 'SupportsInt', 'SupportsRound', # One-off things. 'Annotated', 'assert_never', 'assert_type', 'clear_overloads', 'dataclass_transform', 'deprecated', 'Doc', 'evaluate_forward_ref', 'get_overloads', 'final', 'Format', 'get_annotations', 'get_args', 'get_origin', 'get_original_bases', 'get_protocol_members', 'get_type_hints', 'IntVar', 'is_protocol', 'is_typeddict', 'Literal', 'NewType', 'overload', 'override', 'Protocol', 'reveal_type', 'runtime', 'runtime_checkable', 'Text', 'TypeAlias', 'TypeAliasType', 'TypeForm', 'TypeGuard', 'TypeIs', 'TYPE_CHECKING', 'Never', 'NoReturn', 'ReadOnly', 'Required', 'NotRequired', 'NoDefault', 'NoExtraItems', # Pure aliases, have always been in typing 'AbstractSet', 'AnyStr', 'BinaryIO', 'Callable', 'Collection', 'Container', 'Dict', 'ForwardRef', 'FrozenSet', 'Generator', 'Generic', 'Hashable', 'IO', 'ItemsView', 'Iterable', 'Iterator', 'KeysView', 'List', 'Mapping', 'MappingView', 'Match', 'MutableMapping', 'MutableSequence', 'MutableSet', 'Optional', 'Pattern', 'Reversible', 'Sequence', 'Set', 'Sized', 'TextIO', 'Tuple', 'Union', 'ValuesView', 'cast', 'no_type_check', 'no_type_check_decorator', ] # for backward compatibility PEP_560 = True GenericMeta = type _PEP_696_IMPLEMENTED = sys.version_info >= (3, 13, 0, "beta") # Added with bpo-45166 to 3.10.1+ and some 3.9 versions _FORWARD_REF_HAS_CLASS = "__forward_is_class__" in typing.ForwardRef.__slots__ # The functions below are modified copies of typing internal helpers. # They are needed by _ProtocolMeta and they provide support for PEP 646. class _Sentinel: def __repr__(self): return "<sentinel>" _marker = _Sentinel() if sys.version_info >= (3, 10): def _should_collect_from_parameters(t): return isinstance( t, (typing._GenericAlias, _types.GenericAlias, _types.UnionType) ) elif sys.version_info >= (3, 9): def _should_collect_from_parameters(t): return isinstance(t, (typing._GenericAlias, _types.GenericAlias)) else: def _should_collect_from_parameters(t): return isinstance(t, typing._GenericAlias) and not t._special NoReturn = typing.NoReturn # Some unconstrained type variables. These are used by the container types. # (These are not for export.) T = typing.TypeVar('T') # Any type. KT = typing.TypeVar('KT') # Key type. VT = typing.TypeVar('VT') # Value type. T_co = typing.TypeVar('T_co', covariant=True) # Any type covariant containers. T_contra = typing.TypeVar('T_contra', contravariant=True) # Ditto contravariant. if sys.version_info >= (3, 11): from typing import Any else: class _AnyMeta(type): def __instancecheck__(self, obj): if self is Any: raise TypeError("typing_extensions.Any cannot be used with isinstance()") return super().__instancecheck__(obj) def __repr__(self): if self is Any: return "typing_extensions.Any" return super().__repr__() class Any(metaclass=_AnyMeta): """Special type indicating an unconstrained type. - Any is compatible with every type. - Any assumed to have all methods. - All values assumed to be instances of Any. Note that all the above statements are true from the point of view of static type checkers. At runtime, Any should not be used with instance checks. """ def __new__(cls, args, kwargs): if cls is Any: raise TypeError("Any cannot be instantiated") return super().__new__(cls, args, *kwargs) ClassVar = typing.ClassVar class _ExtensionsSpecialForm(typing._SpecialForm, _root=True): def __repr__(self): return 'typing_extensions.' + self._name Final = typing.Final if sys.version_info >= (3, 11): final = typing.final else: # @final exists in 3.8+, but we backport it for all versions # before 3.11 to keep support for the __final__ attribute. # See https://bugs.python.org/issue46342 def final(f): """This decorator can be used to indicate to type checkers that the decorated method cannot be overridden, and decorated class cannot be subclassed. For example: class Base: @final def done(self) -> None: ... class Sub(Base): def done(self) -> None: # Error reported by type checker ... @final class Leaf: ... class Other(Leaf): # Error reported by type checker ... There is no runtime checking of these properties. The decorator sets the ``__final__`` attribute to ``True`` on the decorated object to allow runtime introspection. """ try: f.__final__ = True except (AttributeError, TypeError): # Skip the attribute silently if it is not writable. # AttributeError happens if the object has __slots__ or a # read-only property, TypeError if it's a builtin class. pass return f def IntVar(name): return typing.TypeVar(name) # A Literal bug was fixed in 3.11.0, 3.10.1 and 3.9.8 if sys.version_info >= (3, 10, 1): Literal = typing.Literal else: def _flatten_literal_params(parameters): """An internal helper for Literal creation: flatten Literals among parameters""" params = [] for p in parameters: if isinstance(p, _LiteralGenericAlias): params.extend(p.__args__) else: params.append(p) return tuple(params) def _value_and_type_iter(params): for p in params: yield p, type(p) class _LiteralGenericAlias(typing._GenericAlias, _root=True): def __eq__(self, other): if not isinstance(other, _LiteralGenericAlias): return NotImplemented these_args_deduped = set(_value_and_type_iter(self.__args__)) other_args_deduped = set(_value_and_type_iter(other.__args__)) return these_args_deduped == other_args_deduped def __hash__(self): return hash(frozenset(_value_and_type_iter(self.__args__))) class _LiteralForm(_ExtensionsSpecialForm, _root=True): def __init__(self, doc: str): self._name = 'Literal' self._doc = self.__doc__ = doc def __getitem__(self, parameters): if not isinstance(parameters, tuple): parameters = (parameters,) parameters = _flatten_literal_params(parameters) val_type_pairs = list(_value_and_type_iter(parameters)) try: deduped_pairs = set(val_type_pairs) except TypeError: # unhashable parameters pass else: # similar logic to typing._deduplicate on Python 3.9+ if len(deduped_pairs) < len(val_type_pairs): new_parameters = [] for pair in val_type_pairs: if pair in deduped_pairs: new_parameters.append(pair[0]) deduped_pairs.remove(pair) assert not deduped_pairs, deduped_pairs parameters = tuple(new_parameters) return _LiteralGenericAlias(self, parameters) Literal = _LiteralForm(doc="""\ A type that can be used to indicate to type checkers that the corresponding value has a value literally equivalent to the provided parameter. For example: var: Literal[4] = 4 The type checker understands that 'var' is literally equal to the value 4 and no other value. Literal[...] cannot be subclassed. There is no runtime checking verifying that the parameter is actually a value instead of a type.""") _overload_dummy = typing._overload_dummy if hasattr(typing, "get_overloads"): # 3.11+ overload = typing.overload get_overloads = typing.get_overloads clear_overloads = typing.clear_overloads else: # {module: {qualname: {firstlineno: func}}} _overload_registry = collections.defaultdict( functools.partial(collections.defaultdict, dict) ) def overload(func): """Decorator for overloaded functions/methods. In a stub file, place two or more stub definitions for the same function in a row, each decorated with @overload. For example: @overload def utf8(value: None) -> None: ... @overload def utf8(value: bytes) -> bytes: ... @overload def utf8(value: str) -> bytes: ... In a non-stub file (i.e. a regular .py file), do the same but follow it with an implementation. The implementation should not* be decorated with @overload. For example: @overload def utf8(value: None) -> None: ... @overload def utf8(value: bytes) -> bytes: ... @overload def utf8(value: str) -> bytes: ... def utf8(value): # implementation goes here The overloads for a function can be retrieved at runtime using the get_overloads() function. """ # classmethod and staticmethod f = getattr(func, "__func__", func) try: _overload_registry[f.__module__][f.__qualname__][ f.__code__.co_firstlineno ] = func except AttributeError: # Not a normal function; ignore. pass return _overload_dummy def get_overloads(func): """Return all defined overloads for func as a sequence.""" # classmethod and staticmethod f = getattr(func, "__func__", func) if f.__module__ not in _overload_registry: return [] mod_dict = _overload_registry[f.__module__] if f.__qualname__ not in mod_dict: return [] return list(mod_dict[f.__qualname__].values()) def clear_overloads(): """Clear all overloads in the registry.""" _overload_registry.clear() # This is not a real generic class. Don't use outside annotations. Type = typing.Type # Various ABCs mimicking those in collections.abc. # A few are simply re-exported for completeness. Awaitable = typing.Awaitable Coroutine = typing.Coroutine AsyncIterable = typing.AsyncIterable AsyncIterator = typing.AsyncIterator Deque = typing.Deque DefaultDict = typing.DefaultDict OrderedDict = typing.OrderedDict Counter = typing.Counter ChainMap = typing.ChainMap Text = typing.Text TYPE_CHECKING = typing.TYPE_CHECKING if sys.version_info >= (3, 13, 0, "beta"): from typing import AsyncContextManager, AsyncGenerator, ContextManager, Generator else: def _is_dunder(attr): return attr.startswith('__') and attr.endswith('__') # Python <3.9 doesn't have typing._SpecialGenericAlias _special_generic_alias_base = getattr( typing, "_SpecialGenericAlias", typing._GenericAlias ) class _SpecialGenericAlias(_special_generic_alias_base, _root=True): def __init__(self, origin, nparams, , inst=True, name=None, defaults=()): if _special_generic_alias_base is typing._GenericAlias: # Python <3.9 self.__origin__ = origin self._nparams = nparams super().__init__(origin, nparams, special=True, inst=inst, name=name) else: # Python >= 3.9 super().__init__(origin, nparams, inst=inst, name=name) self._defaults = defaults def __setattr__(self, attr, val): allowed_attrs = {'_name', '_inst', '_nparams', '_defaults'} if _special_generic_alias_base is typing._GenericAlias: # Python <3.9 allowed_attrs.add("__origin__") if _is_dunder(attr) or attr in allowed_attrs: object.__setattr__(self, attr, val) else: setattr(self.__origin__, attr, val) @typing._tp_cache def __getitem__(self, params): if not isinstance(params, tuple): params = (params,) msg = "Parameters to generic types must be types." params = tuple(typing._type_check(p, msg) for p in params) if ( self._defaults and len(params) < self._nparams and len(params) + len(self._defaults) >= self._nparams ): params = (params, self._defaults[len(params) - self._nparams:]) actual_len = len(params) if actual_len != self._nparams: if self._defaults: expected = f"at least {self._nparams - len(self._defaults)}" else: expected = str(self._nparams) if not self._nparams: raise TypeError(f"{self} is not a generic class") raise TypeError( f"Too {'many' if actual_len > self._nparams else 'few'}" f" arguments for {self};" f" actual {actual_len}, expected {expected}" ) return self.copy_with(params) _NoneType = type(None) Generator = _SpecialGenericAlias( collections.abc.Generator, 3, defaults=(_NoneType, _NoneType) ) AsyncGenerator = _SpecialGenericAlias( collections.abc.AsyncGenerator, 2, defaults=(_NoneType,) ) ContextManager = _SpecialGenericAlias( contextlib.AbstractContextManager, 2, name="ContextManager", defaults=(typing.Optional[bool],) ) AsyncContextManager = _SpecialGenericAlias( contextlib.AbstractAsyncContextManager, 2, name="AsyncContextManager", defaults=(typing.Optional[bool],) ) _PROTO_ALLOWLIST = { 'collections.abc': [ 'Callable', 'Awaitable', 'Iterable', 'Iterator', 'AsyncIterable', 'Hashable', 'Sized', 'Container', 'Collection', 'Reversible', 'Buffer', ], 'contextlib': ['AbstractContextManager', 'AbstractAsyncContextManager'], 'typing_extensions': ['Buffer'], } _EXCLUDED_ATTRS = frozenset(typing.EXCLUDED_ATTRIBUTES) \| { "__match_args__", "__protocol_attrs__", "__non_callable_proto_members__", "__final__", } def _get_protocol_attrs(cls): attrs = set() for base in cls.__mro__[:-1]: # without object if base.__name__ in {'Protocol', 'Generic'}: continue annotations = getattr(base, '__annotations__', {}) for attr in (base.__dict__, annotations): if (not attr.startswith('_abc_') and attr not in _EXCLUDED_ATTRS): attrs.add(attr) return attrs def _caller(depth=2): try: return sys._getframe(depth).f_globals.get('__name__', '__main__') except (AttributeError, ValueError): # For platforms without _getframe() return None # `__match_args__` attribute was removed from protocol members in 3.13, # we want to backport this change to older Python versions. if sys.version_info >= (3, 13): Protocol = typing.Protocol else: def _allow_reckless_class_checks(depth=3): """Allow instance and class checks for special stdlib modules. The abc and functools modules indiscriminately call isinstance() and issubclass() on the whole MRO of a user class, which may contain protocols. """ return _caller(depth) in {'abc', 'functools', None} def _no_init(self, args, *kwargs): if type(self)._is_protocol: raise TypeError('Protocols cannot be instantiated') def _type_check_issubclass_arg_1(arg): """Raise TypeError if `arg` is not an instance of `type` in `issubclass(arg, <protocol>)`. In most cases, this is verified by type.__subclasscheck__. Checking it again unnecessarily would slow down issubclass() checks, so, we don't perform this check unless we absolutely have to. For various error paths, however, we want to ensure that this* error message is shown to the user where relevant, rather than a typing.py-specific error message. """ if not isinstance(arg, type): # Same error message as for issubclass(1, int). raise TypeError('issubclass() arg 1 must be a class') # Inheriting from typing._ProtocolMeta isn't actually desirable, # but is necessary to allow typing.Protocol and typing_extensions.Protocol # to mix without getting TypeErrors about "metaclass conflict" class _ProtocolMeta(type(typing.Protocol)): # This metaclass is somewhat unfortunate, # but is necessary for several reasons... # # NOTE: DO NOT call super() in any methods in this class # That would call the methods on typing._ProtocolMeta on Python 3.8-3.11 # and those are slow def __new__(mcls, name, bases, namespace, kwargs): if name == "Protocol" and len(bases) < 2: pass elif {Protocol, typing.Protocol} & set(bases): for base in bases: if not ( base in {object, typing.Generic, Protocol, typing.Protocol} or base.__name__ in _PROTO_ALLOWLIST.get(base.__module__, []) or is_protocol(base) ): raise TypeError( f"Protocols can only inherit from other protocols, " f"got {base!r}" ) return abc.ABCMeta.__new__(mcls, name, bases, namespace, kwargs) def __init__(cls, args, kwargs): abc.ABCMeta.__init__(cls, args, *kwargs) if getattr(cls, "_is_protocol", False): cls.__protocol_attrs__ = _get_protocol_attrs(cls) def __subclasscheck__(cls, other): if cls is Protocol: return type.__subclasscheck__(cls, other) if ( getattr(cls, '_is_protocol', False) and not _allow_reckless_class_checks() ): if not getattr(cls, '_is_runtime_protocol', False): _type_check_issubclass_arg_1(other) raise TypeError( "Instance and class checks can only be used with " "@runtime_checkable protocols" ) if ( # this attribute is set by @runtime_checkable: cls.__non_callable_proto_members__ and cls.__dict__.get("__subclasshook__") is _proto_hook ): _type_check_issubclass_arg_1(other) non_method_attrs = sorted(cls.__non_callable_proto_members__) raise TypeError( "Protocols with non-method members don't support issubclass()." f" Non-method members: {str(non_method_attrs)[1:-1]}." ) return abc.ABCMeta.__subclasscheck__(cls, other) def __instancecheck__(cls, instance): # We need this method for situations where attributes are # assigned in __init__. if cls is Protocol: return type.__instancecheck__(cls, instance) if not getattr(cls, "_is_protocol", False): # i.e., it's a concrete subclass of a protocol return abc.ABCMeta.__instancecheck__(cls, instance) if ( not getattr(cls, '_is_runtime_protocol', False) and not _allow_reckless_class_checks() ): raise TypeError("Instance and class checks can only be used with" " @runtime_checkable protocols") if abc.ABCMeta.__instancecheck__(cls, instance): return True for attr in cls.__protocol_attrs__: try: val = inspect.getattr_static(instance, attr) except AttributeError: break # this attribute is set by @runtime_checkable: if val is None and attr not in cls.__non_callable_proto_members__: break else: return True return False def __eq__(cls, other): # Hack so that typing.Generic.__class_getitem__ # treats typing_extensions.Protocol # as equivalent to typing.Protocol if abc.ABCMeta.__eq__(cls, other) is True: return True return cls is Protocol and other is typing.Protocol # This has to be defined, or the abc-module cache # complains about classes with this metaclass being unhashable, # if we define only __eq__! def __hash__(cls) -> int: return type.__hash__(cls) @classmethod def _proto_hook(cls, other): if not cls.__dict__.get('_is_protocol', False): return NotImplemented for attr in cls.__protocol_attrs__: for base in other.__mro__: # Check if the members appears in the class dictionary... if attr in base.__dict__: if base.__dict__[attr] is None: return NotImplemented break # ...or in annotations, if it is a sub-protocol. annotations = getattr(base, '__annotations__', {}) if ( isinstance(annotations, collections.abc.Mapping) and attr in annotations and is_protocol(other) ): break else: return NotImplemented return True class Protocol(typing.Generic, metaclass=_ProtocolMeta): __doc__ = typing.Protocol.__doc__ __slots__ = () _is_protocol = True _is_runtime_protocol = False def __init_subclass__(cls, args, *kwargs): super().__init_subclass__(args, *kwargs) # Determine if this is a protocol or a concrete subclass. if not cls.__dict__.get('_is_protocol', False): cls._is_protocol = any(b is Protocol for b in cls.__bases__) # Set (or override) the protocol subclass hook. if '__subclasshook__' not in cls.__dict__: cls.__subclasshook__ = _proto_hook # Prohibit instantiation for protocol classes if cls._is_protocol and cls.__init__ is Protocol.__init__: cls.__init__ = _no_init if sys.version_info >= (3, 13): runtime_checkable = typing.runtime_checkable else: def runtime_checkable(cls): """Mark a protocol class as a runtime protocol. Such protocol can be used with isinstance() and issubclass(). Raise TypeError if applied to a non-protocol class. This allows a simple-minded structural check very similar to one trick ponies in collections.abc such as Iterable. For example:: @runtime_checkable class Closable(Protocol): def close(self): ... assert isinstance(open('/some/file'), Closable) Warning: this will check only the presence of the required methods, not their type signatures! """ if not issubclass(cls, typing.Generic) or not getattr(cls, '_is_protocol', False): raise TypeError(f'@runtime_checkable can be only applied to protocol classes,' f' got {cls!r}') cls._is_runtime_protocol = True # typing.Protocol classes on <=3.11 break if we execute this block, # because typing.Protocol classes on <=3.11 don't have a # `__protocol_attrs__` attribute, and this block relies on the # `__protocol_attrs__` attribute. Meanwhile, typing.Protocol classes on 3.12.2+ # break if we don't* execute this block, because they assume that all # protocol classes have a `__non_callable_proto_members__` attribute # (which this block sets) if isinstance(cls, _ProtocolMeta) or sys.version_info >= (3, 12, 2): # PEP 544 prohibits using issubclass() # with protocols that have non-method members. # See gh-113320 for why we compute this attribute here, # rather than in `_ProtocolMeta.__init__` cls.__non_callable_proto_members__ = set() for attr in cls.__protocol_attrs__: try: is_callable = callable(getattr(cls, attr, None)) except Exception as e: raise TypeError( f"Failed to determine whether protocol member {attr!r} " "is a method member" ) from e else: if not is_callable: cls.__non_callable_proto_members__.add(attr) return cls # The "runtime" alias exists for backwards compatibility. runtime = runtime_checkable # Our version of runtime-checkable protocols is faster on Python 3.8-3.11 if sys.version_info >= (3, 12): SupportsInt = typing.SupportsInt SupportsFloat = typing.SupportsFloat SupportsComplex = typing.SupportsComplex SupportsBytes = typing.SupportsBytes SupportsIndex = typing.SupportsIndex SupportsAbs = typing.SupportsAbs SupportsRound = typing.SupportsRound else: @runtime_checkable class SupportsInt(Protocol): """An ABC with one abstract method __int__.""" __slots__ = () @abc.abstractmethod def __int__(self) -> int: pass @runtime_checkable class SupportsFloat(Protocol): """An ABC with one abstract method __float__.""" __slots__ = () @abc.abstractmethod def __float__(self) -> float: pass @runtime_checkable class SupportsComplex(Protocol): """An ABC with one abstract method __complex__.""" __slots__ = () @abc.abstractmethod def __complex__(self) -> complex: pass @runtime_checkable class SupportsBytes(Protocol): """An ABC with one abstract method __bytes__.""" __slots__ = () @abc.abstractmethod def __bytes__(self) -> bytes: pass @runtime_checkable class SupportsIndex(Protocol): __slots__ = () @abc.abstractmethod def __index__(self) -> int: pass @runtime_checkable class SupportsAbs(Protocol[T_co]): """ An ABC with one abstract method __abs__ that is covariant in its return type. """ __slots__ = () @abc.abstractmethod def __abs__(self) -> T_co: pass @runtime_checkable class SupportsRound(Protocol[T_co]): """ An ABC with one abstract method __round__ that is covariant in its return type. """ __slots__ = () @abc.abstractmethod def __round__(self, ndigits: int = 0) -> T_co: pass def _ensure_subclassable(mro_entries): def inner(func): if sys.implementation.name == "pypy" and sys.version_info < (3, 9): cls_dict = { "__call__": staticmethod(func), "__mro_entries__": staticmethod(mro_entries) } t = type(func.__name__, (), cls_dict) return functools.update_wrapper(t(), func) else: func.__mro_entries__ = mro_entries return func return inner _NEEDS_SINGLETONMETA = ( not hasattr(typing, "NoDefault") or not hasattr(typing, "NoExtraItems") ) if _NEEDS_SINGLETONMETA: class SingletonMeta(type): def __setattr__(cls, attr, value): # TypeError is consistent with the behavior of NoneType raise TypeError( f"cannot set {attr!r} attribute of immutable type {cls.__name__!r}" ) if hasattr(typing, "NoDefault"): NoDefault = typing.NoDefault else: class NoDefaultType(metaclass=SingletonMeta): """The type of the NoDefault singleton.""" __slots__ = () def __new__(cls): return globals().get("NoDefault") or object.__new__(cls) def __repr__(self): return "typing_extensions.NoDefault" def __reduce__(self): return "NoDefault" NoDefault = NoDefaultType() del NoDefaultType if hasattr(typing, "NoExtraItems"): NoExtraItems = typing.NoExtraItems else: class NoExtraItemsType(metaclass=SingletonMeta): """The type of the NoExtraItems singleton.""" __slots__ = () def __new__(cls): return globals().get("NoExtraItems") or object.__new__(cls) def __repr__(self): return "typing_extensions.NoExtraItems" def __reduce__(self): return "NoExtraItems" NoExtraItems = NoExtraItemsType() del NoExtraItemsType if _NEEDS_SINGLETONMETA: del SingletonMeta # Update this to something like >=3.13.0b1 if and when # PEP 728 is implemented in CPython _PEP_728_IMPLEMENTED = False if _PEP_728_IMPLEMENTED: # The standard library TypedDict in Python 3.8 does not store runtime information # about which (if any) keys are optional. See https://bugs.python.org/issue38834 # The standard library TypedDict in Python 3.9.0/1 does not honour the "total" # keyword with old-style TypedDict(). See https://bugs.python.org/issue42059 # The standard library TypedDict below Python 3.11 does not store runtime # information about optional and required keys when using Required or NotRequired. # Generic TypedDicts are also impossible using typing.TypedDict on Python <3.11. # Aaaand on 3.12 we add __orig_bases__ to TypedDict # to enable better runtime introspection. # On 3.13 we deprecate some odd ways of creating TypedDicts. # Also on 3.13, PEP 705 adds the ReadOnly[] qualifier. # PEP 728 (still pending) makes more changes. TypedDict = typing.TypedDict _TypedDictMeta = typing._TypedDictMeta is_typeddict = typing.is_typeddict else: # 3.10.0 and later _TAKES_MODULE = "module" in inspect.signature(typing._type_check).parameters def _get_typeddict_qualifiers(annotation_type): while True: annotation_origin = get_origin(annotation_type) if annotation_origin is Annotated: annotation_args = get_args(annotation_type) if annotation_args: annotation_type = annotation_args[0] else: break elif annotation_origin is Required: yield Required annotation_type, = get_args(annotation_type) elif annotation_origin is NotRequired: yield NotRequired annotation_type, = get_args(annotation_type) elif annotation_origin is ReadOnly: yield ReadOnly annotation_type, = get_args(annotation_type) else: break class _TypedDictMeta(type): def __new__(cls, name, bases, ns, , total=True, closed=None, extra_items=NoExtraItems): """Create new typed dict class object. This method is called when TypedDict is subclassed, or when TypedDict is instantiated. This way TypedDict supports all three syntax forms described in its docstring. Subclasses and instances of TypedDict return actual dictionaries. """ for base in bases: if type(base) is not _TypedDictMeta and base is not typing.Generic: raise TypeError('cannot inherit from both a TypedDict type ' 'and a non-TypedDict base class') if closed is not None and extra_items is not NoExtraItems: raise TypeError(f"Cannot combine closed={closed!r} and extra_items") if any(issubclass(b, typing.Generic) for b in bases): generic_base = (typing.Generic,) else: generic_base = () # typing.py generally doesn't let you inherit from plain Generic, unless # the name of the class happens to be "Protocol" tp_dict = type.__new__(_TypedDictMeta, "Protocol", (generic_base, dict), ns) tp_dict.__name__ = name if tp_dict.__qualname__ == "Protocol": tp_dict.__qualname__ = name if not hasattr(tp_dict, '__orig_bases__'): tp_dict.__orig_bases__ = bases annotations = {} if "__annotations__" in ns: own_annotations = ns["__annotations__"] elif "__annotate__" in ns: # TODO: Use inspect.VALUE here, and make the annotations lazily evaluated own_annotations = ns["__annotate__"](1) else: own_annotations = {} msg = "TypedDict('Name', {f0: t0, f1: t1, ...}); each t must be a type" if _TAKES_MODULE: own_annotations = { n: typing._type_check(tp, msg, module=tp_dict.__module__) for n, tp in own_annotations.items() } else: own_annotations = { n: typing._type_check(tp, msg) for n, tp in own_annotations.items() } required_keys = set() optional_keys = set() readonly_keys = set() mutable_keys = set() extra_items_type = extra_items for base in bases: base_dict = base.__dict__ annotations.update(base_dict.get('__annotations__', {})) required_keys.update(base_dict.get('__required_keys__', ())) optional_keys.update(base_dict.get('__optional_keys__', ())) readonly_keys.update(base_dict.get('__readonly_keys__', ())) mutable_keys.update(base_dict.get('__mutable_keys__', ())) # This was specified in an earlier version of PEP 728. Support # is retained for backwards compatibility, but only for Python # 3.13 and lower. if (closed and sys.version_info < (3, 14) and "__extra_items__" in own_annotations): annotation_type = own_annotations.pop("__extra_items__") qualifiers = set(_get_typeddict_qualifiers(annotation_type)) if Required in qualifiers: raise TypeError( "Special key __extra_items__ does not support " "Required" ) if NotRequired in qualifiers: raise TypeError( "Special key __extra_items__ does not support " "NotRequired" ) extra_items_type = annotation_type annotations.update(own_annotations) for annotation_key, annotation_type in own_annotations.items(): qualifiers = set(_get_typeddict_qualifiers(annotation_type)) if Required in qualifiers: required_keys.add(annotation_key) elif NotRequired in qualifiers: optional_keys.add(annotation_key) elif total: required_keys.add(annotation_key) else: optional_keys.add(annotation_key) if ReadOnly in qualifiers: mutable_keys.discard(annotation_key) readonly_keys.add(annotation_key) else: mutable_keys.add(annotation_key) readonly_keys.discard(annotation_key) tp_dict.__annotations__ = annotations tp_dict.__required_keys__ = frozenset(required_keys) tp_dict.__optional_keys__ = frozenset(optional_keys) tp_dict.__readonly_keys__ = frozenset(readonly_keys) tp_dict.__mutable_keys__ = frozenset(mutable_keys) tp_dict.__total__ = total tp_dict.__closed__ = closed tp_dict.__extra_items__ = extra_items_type return tp_dict __call__ = dict # static method def __subclasscheck__(cls, other): # Typed dicts are only for static structural subtyping. raise TypeError('TypedDict does not support instance and class checks') __instancecheck__ = __subclasscheck__ _TypedDict = type.__new__(_TypedDictMeta, 'TypedDict', (), {}) @_ensure_subclassable(lambda bases: (_TypedDict,)) def TypedDict( typename, fields=_marker, /, , total=True, closed=None, extra_items=NoExtraItems, kwargs ): """A simple typed namespace. At runtime it is equivalent to a plain dict. TypedDict creates a dictionary type such that a type checker will expect all instances to have a certain set of keys, where each key is associated with a value of a consistent type. This expectation is not checked at runtime. Usage:: class Point2D(TypedDict): x: int y: int label: str a: Point2D = {'x': 1, 'y': 2, 'label': 'good'} # OK b: Point2D = {'z': 3, 'label': 'bad'} # Fails type check assert Point2D(x=1, y=2, label='first') == dict(x=1, y=2, label='first') The type info can be accessed via the Point2D.__annotations__ dict, and the Point2D.__required_keys__ and Point2D.__optional_keys__ frozensets. TypedDict supports an additional equivalent form:: Point2D = TypedDict('Point2D', {'x': int, 'y': int, 'label': str}) By default, all keys must be present in a TypedDict. It is possible to override this by specifying totality:: class Point2D(TypedDict, total=False): x: int y: int This means that a Point2D TypedDict can have any of the keys omitted. A type checker is only expected to support a literal False or True as the value of the total argument. True is the default, and makes all items defined in the class body be required. The Required and NotRequired special forms can also be used to mark individual keys as being required or not required:: class Point2D(TypedDict): x: int # the "x" key must always be present (Required is the default) y: NotRequired[int] # the "y" key can be omitted See PEP 655 for more details on Required and NotRequired. """ if fields is _marker or fields is None: if fields is _marker: deprecated_thing = "Failing to pass a value for the 'fields' parameter" else: deprecated_thing = "Passing `None` as the 'fields' parameter" example = f"`{typename} = TypedDict({typename!r}, {{}})`" deprecation_msg = ( f"{deprecated_thing} is deprecated and will be disallowed in " "Python 3.15. To create a TypedDict class with 0 fields " "using the functional syntax, pass an empty dictionary, e.g. " ) + example + "." warnings.warn(deprecation_msg, DeprecationWarning, stacklevel=2) # Support a field called "closed" if closed is not False and closed is not True and closed is not None: kwargs["closed"] = closed closed = None # Or "extra_items" if extra_items is not NoExtraItems: kwargs["extra_items"] = extra_items extra_items = NoExtraItems fields = kwargs elif kwargs: raise TypeError("TypedDict takes either a dict or keyword arguments," " but not both") if kwargs: if sys.version_info >= (3, 13): raise TypeError("TypedDict takes no keyword arguments") warnings.warn( "The kwargs-based syntax for TypedDict definitions is deprecated " "in Python 3.11, will be removed in Python 3.13, and may not be " "understood by third-party type checkers.", DeprecationWarning, stacklevel=2, ) ns = {'__annotations__': dict(fields)} module = _caller() if module is not None: # Setting correct module is necessary to make typed dict classes pickleable. ns['__module__'] = module td = _TypedDictMeta(typename, (), ns, total=total, closed=closed, extra_items=extra_items) td.__orig_bases__ = (TypedDict,) return td if hasattr(typing, "_TypedDictMeta"): _TYPEDDICT_TYPES = (typing._TypedDictMeta, _TypedDictMeta) else: _TYPEDDICT_TYPES = (_TypedDictMeta,) def is_typeddict(tp): """Check if an annotation is a TypedDict class For example:: class Film(TypedDict): title: str year: int is_typeddict(Film) # => True is_typeddict(Union[list, str]) # => False """ # On 3.8, this would otherwise return True if hasattr(typing, "TypedDict") and tp is typing.TypedDict: return False return isinstance(tp, _TYPEDDICT_TYPES) if hasattr(typing, "assert_type"): assert_type = typing.assert_type else: def assert_type(val, typ, /): """Assert (to the type checker) that the value is of the given type. When the type checker encounters a call to assert_type(), it emits an error if the value is not of the specified type:: def greet(name: str) -> None: assert_type(name, str) # ok assert_type(name, int) # type checker error At runtime this returns the first argument unchanged and otherwise does nothing. """ return val if hasattr(typing, "ReadOnly"): # 3.13+ get_type_hints = typing.get_type_hints else: # <=3.13 # replaces _strip_annotations() def _strip_extras(t): """Strips Annotated, Required and NotRequired from a given type.""" if isinstance(t, _AnnotatedAlias): return _strip_extras(t.__origin__) if hasattr(t, "__origin__") and t.__origin__ in (Required, NotRequired, ReadOnly): return _strip_extras(t.__args__[0]) if isinstance(t, typing._GenericAlias): stripped_args = tuple(_strip_extras(a) for a in t.__args__) if stripped_args == t.__args__: return t return t.copy_with(stripped_args) if hasattr(_types, "GenericAlias") and isinstance(t, _types.GenericAlias): stripped_args = tuple(_strip_extras(a) for a in t.__args__) if stripped_args == t.__args__: return t return _types.GenericAlias(t.__origin__, stripped_args) if hasattr(_types, "UnionType") and isinstance(t, _types.UnionType): stripped_args = tuple(_strip_extras(a) for a in t.__args__) if stripped_args == t.__args__: return t return functools.reduce(operator.or_, stripped_args) return t def get_type_hints(obj, globalns=None, localns=None, include_extras=False): """Return type hints for an object. This is often the same as obj.__annotations__, but it handles forward references encoded as string literals, adds Optional[t] if a default value equal to None is set and recursively replaces all 'Annotated[T, ...]', 'Required[T]' or 'NotRequired[T]' with 'T' (unless 'include_extras=True'). The argument may be a module, class, method, or function. The annotations are returned as a dictionary. For classes, annotations include also inherited members. TypeError is raised if the argument is not of a type that can contain annotations, and an empty dictionary is returned if no annotations are present. BEWARE -- the behavior of globalns and localns is counterintuitive (unless you are familiar with how eval() and exec() work). The search order is locals first, then globals. - If no dict arguments are passed, an attempt is made to use the globals from obj (or the respective module's globals for classes), and these are also used as the locals. If the object does not appear to have globals, an empty dictionary is used. - If one dict argument is passed, it is used for both globals and locals. - If two dict arguments are passed, they specify globals and locals, respectively. """ if hasattr(typing, "Annotated"): # 3.9+ hint = typing.get_type_hints( obj, globalns=globalns, localns=localns, include_extras=True ) else: # 3.8 hint = typing.get_type_hints(obj, globalns=globalns, localns=localns) if sys.version_info < (3, 11): _clean_optional(obj, hint, globalns, localns) if sys.version_info < (3, 9): # In 3.8 eval_type does not flatten Optional[ForwardRef] correctly # This will recreate and and cache Unions. hint = { k: (t if get_origin(t) != Union else Union[t.__args__]) for k, t in hint.items() } if include_extras: return hint return {k: _strip_extras(t) for k, t in hint.items()} _NoneType = type(None) def _could_be_inserted_optional(t): """detects Union[..., None] pattern""" # 3.8+ compatible checking before _UnionGenericAlias if get_origin(t) is not Union: return False # Assume if last argument is not None they are user defined if t.__args__[-1] is not _NoneType: return False return True # < 3.11 def _clean_optional(obj, hints, globalns=None, localns=None): # reverts injected Union[..., None] cases from typing.get_type_hints # when a None default value is used. # see https://github.com/python/typing_extensions/issues/310 if not hints or isinstance(obj, type): return defaults = typing._get_defaults(obj) # avoid accessing __annotations___ if not defaults: return original_hints = obj.__annotations__ for name, value in hints.items(): # Not a Union[..., None] or replacement conditions not fullfilled if (not _could_be_inserted_optional(value) or name not in defaults or defaults[name] is not None ): continue original_value = original_hints[name] # value=NoneType should have caused a skip above but check for safety if original_value is None: original_value = _NoneType # Forward reference if isinstance(original_value, str): if globalns is None: if isinstance(obj, _types.ModuleType): globalns = obj.__dict__ else: nsobj = obj # Find globalns for the unwrapped object. while hasattr(nsobj, '__wrapped__'): nsobj = nsobj.__wrapped__ globalns = getattr(nsobj, '__globals__', {}) if localns is None: localns = globalns elif localns is None: localns = globalns if sys.version_info < (3, 9): original_value = ForwardRef(original_value) else: original_value = ForwardRef( original_value, is_argument=not isinstance(obj, _types.ModuleType) ) original_evaluated = typing._eval_type(original_value, globalns, localns) if sys.version_info < (3, 9) and get_origin(original_evaluated) is Union: # Union[str, None, "str"] is not reduced to Union[str, None] original_evaluated = Union[original_evaluated.__args__] # Compare if values differ. Note that even if equal # value might be cached by typing._tp_cache contrary to original_evaluated if original_evaluated != value or ( # 3.10: ForwardRefs of UnionType might be turned into _UnionGenericAlias hasattr(_types, "UnionType") and isinstance(original_evaluated, _types.UnionType) and not isinstance(value, _types.UnionType) ): hints[name] = original_evaluated # Python 3.9+ has PEP 593 (Annotated) if hasattr(typing, 'Annotated'): Annotated = typing.Annotated # Not exported and not a public API, but needed for get_origin() and get_args() # to work. _AnnotatedAlias = typing._AnnotatedAlias # 3.8 else: class _AnnotatedAlias(typing._GenericAlias, _root=True): """Runtime representation of an annotated type. At its core 'Annotated[t, dec1, dec2, ...]' is an alias for the type 't' with extra annotations. The alias behaves like a normal typing alias, instantiating is the same as instantiating the underlying type, binding it to types is also the same. """ def __init__(self, origin, metadata): if isinstance(origin, _AnnotatedAlias): metadata = origin.__metadata__ + metadata origin = origin.__origin__ super().__init__(origin, origin) self.__metadata__ = metadata def copy_with(self, params): assert len(params) == 1 new_type = params[0] return _AnnotatedAlias(new_type, self.__metadata__) def __repr__(self): return (f"typing_extensions.Annotated[{typing._type_repr(self.__origin__)}, " f"{', '.join(repr(a) for a in self.__metadata__)}]") def __reduce__(self): return operator.getitem, ( Annotated, (self.__origin__, self.__metadata__) ) def __eq__(self, other): if not isinstance(other, _AnnotatedAlias): return NotImplemented if self.__origin__ != other.__origin__: return False return self.__metadata__ == other.__metadata__ def __hash__(self): return hash((self.__origin__, self.__metadata__)) class Annotated: """Add context specific metadata to a type. Example: Annotated[int, runtime_check.Unsigned] indicates to the hypothetical runtime_check module that this type is an unsigned int. Every other consumer of this type can ignore this metadata and treat this type as int. The first argument to Annotated must be a valid type (and will be in the __origin__ field), the remaining arguments are kept as a tuple in the __extra__ field. Details: - It's an error to call `Annotated` with less than two arguments. - Nested Annotated are flattened:: Annotated[Annotated[T, Ann1, Ann2], Ann3] == Annotated[T, Ann1, Ann2, Ann3] - Instantiating an annotated type is equivalent to instantiating the underlying type:: Annotated[C, Ann1](5) == C(5) - Annotated can be used as a generic type alias:: Optimized = Annotated[T, runtime.Optimize()] Optimized[int] == Annotated[int, runtime.Optimize()] OptimizedList = Annotated[List[T], runtime.Optimize()] OptimizedList[int] == Annotated[List[int], runtime.Optimize()] """ __slots__ = () def __new__(cls, args, kwargs): raise TypeError("Type Annotated cannot be instantiated.") @typing._tp_cache def __class_getitem__(cls, params): if not isinstance(params, tuple) or len(params) < 2: raise TypeError("Annotated[...] should be used " "with at least two arguments (a type and an " "annotation).") allowed_special_forms = (ClassVar, Final) if get_origin(params[0]) in allowed_special_forms: origin = params[0] else: msg = "Annotated[t, ...]: t must be a type." origin = typing._type_check(params[0], msg) metadata = tuple(params[1:]) return _AnnotatedAlias(origin, metadata) def __init_subclass__(cls, args, *kwargs): raise TypeError( f"Cannot subclass {cls.__module__}.Annotated" ) # Python 3.8 has get_origin() and get_args() but those implementations aren't # Annotated-aware, so we can't use those. Python 3.9's versions don't support # ParamSpecArgs and ParamSpecKwargs, so only Python 3.10's versions will do. if sys.version_info[:2] >= (3, 10): get_origin = typing.get_origin get_args = typing.get_args # 3.8-3.9 else: try: # 3.9+ from typing import _BaseGenericAlias except ImportError: _BaseGenericAlias = typing._GenericAlias try: # 3.9+ from typing import GenericAlias as _typing_GenericAlias except ImportError: _typing_GenericAlias = typing._GenericAlias def get_origin(tp): """Get the unsubscripted version of a type. This supports generic types, Callable, Tuple, Union, Literal, Final, ClassVar and Annotated. Return None for unsupported types. Examples:: get_origin(Literal[42]) is Literal get_origin(int) is None get_origin(ClassVar[int]) is ClassVar get_origin(Generic) is Generic get_origin(Generic[T]) is Generic get_origin(Union[T, int]) is Union get_origin(List[Tuple[T, T]][int]) == list get_origin(P.args) is P """ if isinstance(tp, _AnnotatedAlias): return Annotated if isinstance(tp, (typing._GenericAlias, _typing_GenericAlias, _BaseGenericAlias, ParamSpecArgs, ParamSpecKwargs)): return tp.__origin__ if tp is typing.Generic: return typing.Generic return None def get_args(tp): """Get type arguments with all substitutions performed. For unions, basic simplifications used by Union constructor are performed. Examples:: get_args(Dict[str, int]) == (str, int) get_args(int) == () get_args(Union[int, Union[T, int], str][int]) == (int, str) get_args(Union[int, Tuple[T, int]][str]) == (int, Tuple[str, int]) get_args(Callable[[], T][int]) == ([], int) """ if isinstance(tp, _AnnotatedAlias): return (tp.__origin__, tp.__metadata__) if isinstance(tp, (typing._GenericAlias, _typing_GenericAlias)): if getattr(tp, "_special", False): return () res = tp.__args__ if get_origin(tp) is collections.abc.Callable and res[0] is not Ellipsis: res = (list(res[:-1]), res[-1]) return res return () # 3.10+ if hasattr(typing, 'TypeAlias'): TypeAlias = typing.TypeAlias # 3.9 elif sys.version_info[:2] >= (3, 9): @_ExtensionsSpecialForm def TypeAlias(self, parameters): """Special marker indicating that an assignment should be recognized as a proper type alias definition by type checkers. For example:: Predicate: TypeAlias = Callable[..., bool] It's invalid when used anywhere except as in the example above. """ raise TypeError(f"{self} is not subscriptable") # 3.8 else: TypeAlias = _ExtensionsSpecialForm( 'TypeAlias', doc="""Special marker indicating that an assignment should be recognized as a proper type alias definition by type checkers. For example:: Predicate: TypeAlias = Callable[..., bool] It's invalid when used anywhere except as in the example above.""" ) def _set_default(type_param, default): type_param.has_default = lambda: default is not NoDefault type_param.__default__ = default def _set_module(typevarlike): # for pickling: def_mod = _caller(depth=3) if def_mod != 'typing_extensions': typevarlike.__module__ = def_mod class _DefaultMixin: """Mixin for TypeVarLike defaults.""" __slots__ = () __init__ = _set_default # Classes using this metaclass must provide a _backported_typevarlike ClassVar class _TypeVarLikeMeta(type): def __instancecheck__(cls, __instance: Any) -> bool: return isinstance(__instance, cls._backported_typevarlike) if _PEP_696_IMPLEMENTED: from typing import TypeVar else: # Add default and infer_variance parameters from PEP 696 and 695 class TypeVar(metaclass=_TypeVarLikeMeta): """Type variable.""" _backported_typevarlike = typing.TypeVar def __new__(cls, name, constraints, bound=None, covariant=False, contravariant=False, default=NoDefault, infer_variance=False): if hasattr(typing, "TypeAliasType"): # PEP 695 implemented (3.12+), can pass infer_variance to typing.TypeVar typevar = typing.TypeVar(name, constraints, bound=bound, covariant=covariant, contravariant=contravariant, infer_variance=infer_variance) else: typevar = typing.TypeVar(name, constraints, bound=bound, covariant=covariant, contravariant=contravariant) if infer_variance and (covariant or contravariant): raise ValueError("Variance cannot be specified with infer_variance.") typevar.__infer_variance__ = infer_variance _set_default(typevar, default) _set_module(typevar) def _tvar_prepare_subst(alias, args): if ( typevar.has_default() and alias.__parameters__.index(typevar) == len(args) ): args += (typevar.__default__,) return args typevar.__typing_prepare_subst__ = _tvar_prepare_subst return typevar def __init_subclass__(cls) -> None: raise TypeError(f"type '{__name__}.TypeVar' is not an acceptable base type") # Python 3.10+ has PEP 612 if hasattr(typing, 'ParamSpecArgs'): ParamSpecArgs = typing.ParamSpecArgs ParamSpecKwargs = typing.ParamSpecKwargs # 3.8-3.9 else: class _Immutable: """Mixin to indicate that object should not be copied.""" __slots__ = () def __copy__(self): return self def __deepcopy__(self, memo): return self class ParamSpecArgs(_Immutable): """The args for a ParamSpec object. Given a ParamSpec object P, P.args is an instance of ParamSpecArgs. ParamSpecArgs objects have a reference back to their ParamSpec: P.args.__origin__ is P This type is meant for runtime introspection and has no special meaning to static type checkers. """ def __init__(self, origin): self.__origin__ = origin def __repr__(self): return f"{self.__origin__.__name__}.args" def __eq__(self, other): if not isinstance(other, ParamSpecArgs): return NotImplemented return self.__origin__ == other.__origin__ class ParamSpecKwargs(_Immutable): """The kwargs for a ParamSpec object. Given a ParamSpec object P, P.kwargs is an instance of ParamSpecKwargs. ParamSpecKwargs objects have a reference back to their ParamSpec: P.kwargs.__origin__ is P This type is meant for runtime introspection and has no special meaning to static type checkers. """ def __init__(self, origin): self.__origin__ = origin def __repr__(self): return f"{self.__origin__.__name__}.kwargs" def __eq__(self, other): if not isinstance(other, ParamSpecKwargs): return NotImplemented return self.__origin__ == other.__origin__ if _PEP_696_IMPLEMENTED: from typing import ParamSpec # 3.10+ elif hasattr(typing, 'ParamSpec'): # Add default parameter - PEP 696 class ParamSpec(metaclass=_TypeVarLikeMeta): """Parameter specification.""" _backported_typevarlike = typing.ParamSpec def __new__(cls, name, , bound=None, covariant=False, contravariant=False, infer_variance=False, default=NoDefault): if hasattr(typing, "TypeAliasType"): # PEP 695 implemented, can pass infer_variance to typing.TypeVar paramspec = typing.ParamSpec(name, bound=bound, covariant=covariant, contravariant=contravariant, infer_variance=infer_variance) else: paramspec = typing.ParamSpec(name, bound=bound, covariant=covariant, contravariant=contravariant) paramspec.__infer_variance__ = infer_variance _set_default(paramspec, default) _set_module(paramspec) def _paramspec_prepare_subst(alias, args): params = alias.__parameters__ i = params.index(paramspec) if i == len(args) and paramspec.has_default(): args = [args, paramspec.__default__] if i >= len(args): raise TypeError(f"Too few arguments for {alias}") # Special case where Z[[int, str, bool]] == Z[int, str, bool] in PEP 612. if len(params) == 1 and not typing._is_param_expr(args[0]): assert i == 0 args = (args,) # Convert lists to tuples to help other libraries cache the results. elif isinstance(args[i], list): args = (args[:i], tuple(args[i]), args[i + 1:]) return args paramspec.__typing_prepare_subst__ = _paramspec_prepare_subst return paramspec def __init_subclass__(cls) -> None: raise TypeError(f"type '{__name__}.ParamSpec' is not an acceptable base type") # 3.8-3.9 else: # Inherits from list as a workaround for Callable checks in Python < 3.9.2. class ParamSpec(list, _DefaultMixin): """Parameter specification variable. Usage:: P = ParamSpec('P') Parameter specification variables exist primarily for the benefit of static type checkers. They are used to forward the parameter types of one callable to another callable, a pattern commonly found in higher order functions and decorators. They are only valid when used in ``Concatenate``, or s the first argument to ``Callable``. In Python 3.10 and higher, they are also supported in user-defined Generics at runtime. See class Generic for more information on generic types. An example for annotating a decorator:: T = TypeVar('T') P = ParamSpec('P') def add_logging(f: Callable[P, T]) -> Callable[P, T]: '''A type-safe decorator to add logging to a function.''' def inner(args: P.args, *kwargs: P.kwargs) -> T: logging.info(f'{f.__name__} was called') return f(args, *kwargs) return inner @add_logging def add_two(x: float, y: float) -> float: '''Add two numbers together.''' return x + y Parameter specification variables defined with covariant=True or contravariant=True can be used to declare covariant or contravariant generic types. These keyword arguments are valid, but their actual semantics are yet to be decided. See PEP 612 for details. Parameter specification variables can be introspected. e.g.: P.__name__ == 'T' P.__bound__ == None P.__covariant__ == False P.__contravariant__ == False Note that only parameter specification variables defined in global scope can be pickled. """ # Trick Generic __parameters__. __class__ = typing.TypeVar @property def args(self): return ParamSpecArgs(self) @property def kwargs(self): return ParamSpecKwargs(self) def __init__(self, name, , bound=None, covariant=False, contravariant=False, infer_variance=False, default=NoDefault): list.__init__(self, [self]) self.__name__ = name self.__covariant__ = bool(covariant) self.__contravariant__ = bool(contravariant) self.__infer_variance__ = bool(infer_variance) if bound: self.__bound__ = typing._type_check(bound, 'Bound must be a type.') else: self.__bound__ = None _DefaultMixin.__init__(self, default) # for pickling: def_mod = _caller() if def_mod != 'typing_extensions': self.__module__ = def_mod def __repr__(self): if self.__infer_variance__: prefix = '' elif self.__covariant__: prefix = '+' elif self.__contravariant__: prefix = '-' else: prefix = '~' return prefix + self.__name__ def __hash__(self): return object.__hash__(self) def __eq__(self, other): return self is other def __reduce__(self): return self.__name__ # Hack to get typing._type_check to pass. def __call__(self, args, kwargs): pass # 3.8-3.9 if not hasattr(typing, 'Concatenate'): # Inherits from list as a workaround for Callable checks in Python < 3.9.2. # 3.9.0-1 if not hasattr(typing, '_type_convert'): def _type_convert(arg, module=None, , allow_special_forms=False): """For converting None to type(None), and strings to ForwardRef.""" if arg is None: return type(None) if isinstance(arg, str): if sys.version_info <= (3, 9, 6): return ForwardRef(arg) if sys.version_info <= (3, 9, 7): return ForwardRef(arg, module=module) return ForwardRef(arg, module=module, is_class=allow_special_forms) return arg else: _type_convert = typing._type_convert class _ConcatenateGenericAlias(list): # Trick Generic into looking into this for __parameters__. __class__ = typing._GenericAlias # Flag in 3.8. _special = False def __init__(self, origin, args): super().__init__(args) self.__origin__ = origin self.__args__ = args def __repr__(self): _type_repr = typing._type_repr return (f'{_type_repr(self.__origin__)}' f'[{", ".join(_type_repr(arg) for arg in self.__args__)}]') def __hash__(self): return hash((self.__origin__, self.__args__)) # Hack to get typing._type_check to pass in Generic. def __call__(self, args, kwargs): pass @property def __parameters__(self): return tuple( tp for tp in self.__args__ if isinstance(tp, (typing.TypeVar, ParamSpec)) ) # 3.8; needed for typing._subst_tvars # 3.9 used by __getitem__ below def copy_with(self, params): if isinstance(params[-1], _ConcatenateGenericAlias): params = (params[:-1], params[-1].__args__) elif isinstance(params[-1], (list, tuple)): return (params[:-1], params[-1]) elif (not (params[-1] is ... or isinstance(params[-1], ParamSpec))): raise TypeError("The last parameter to Concatenate should be a " "ParamSpec variable or ellipsis.") return self.__class__(self.__origin__, params) # 3.9; accessed during GenericAlias.__getitem__ when substituting def __getitem__(self, args): if self.__origin__ in (Generic, Protocol): # Can't subscript Generic[...] or Protocol[...]. raise TypeError(f"Cannot subscript already-subscripted {self}") if not self.__parameters__: raise TypeError(f"{self} is not a generic class") if not isinstance(args, tuple): args = (args,) args = _unpack_args((_type_convert(p) for p in args)) params = self.__parameters__ for param in params: prepare = getattr(param, "__typing_prepare_subst__", None) if prepare is not None: args = prepare(self, args) # 3.8 - 3.9 & typing.ParamSpec elif isinstance(param, ParamSpec): i = params.index(param) if ( i == len(args) and getattr(param, '__default__', NoDefault) is not NoDefault ): args = [args, param.__default__] if i >= len(args): raise TypeError(f"Too few arguments for {self}") # Special case for Z[[int, str, bool]] == Z[int, str, bool] if len(params) == 1 and not _is_param_expr(args[0]): assert i == 0 args = (args,) elif ( isinstance(args[i], list) # 3.8 - 3.9 # This class inherits from list do not convert and not isinstance(args[i], _ConcatenateGenericAlias) ): args = (args[:i], tuple(args[i]), args[i + 1:]) alen = len(args) plen = len(params) if alen != plen: raise TypeError( f"Too {'many' if alen > plen else 'few'} arguments for {self};" f" actual {alen}, expected {plen}" ) subst = dict(zip(self.__parameters__, args)) # determine new args new_args = [] for arg in self.__args__: if isinstance(arg, type): new_args.append(arg) continue if isinstance(arg, TypeVar): arg = subst[arg] if ( (isinstance(arg, typing._GenericAlias) and _is_unpack(arg)) or ( hasattr(_types, "GenericAlias") and isinstance(arg, _types.GenericAlias) and getattr(arg, "__unpacked__", False) ) ): raise TypeError(f"{arg} is not valid as type argument") elif isinstance(arg, typing._GenericAlias if not hasattr(_types, "GenericAlias") else (typing._GenericAlias, _types.GenericAlias) ): subparams = arg.__parameters__ if subparams: subargs = tuple(subst[x] for x in subparams) arg = arg[subargs] new_args.append(arg) return self.copy_with(tuple(new_args)) # 3.10+ else: _ConcatenateGenericAlias = typing._ConcatenateGenericAlias # 3.10 if sys.version_info < (3, 11): class _ConcatenateGenericAlias(typing._ConcatenateGenericAlias, _root=True): # needed for checks in collections.abc.Callable to accept this class __module__ = "typing" def copy_with(self, params): if isinstance(params[-1], (list, tuple)): return (params[:-1], params[-1]) if isinstance(params[-1], typing._ConcatenateGenericAlias): params = (params[:-1], params[-1].__args__) elif not (params[-1] is ... or isinstance(params[-1], ParamSpec)): raise TypeError("The last parameter to Concatenate should be a " "ParamSpec variable or ellipsis.") return super(typing._ConcatenateGenericAlias, self).copy_with(params) def __getitem__(self, args): value = super().__getitem__(args) if isinstance(value, tuple) and any(_is_unpack(t) for t in value): return tuple(_unpack_args((n for n in value))) return value # 3.8-3.9.2 class _EllipsisDummy: ... # 3.8-3.10 def _create_concatenate_alias(origin, parameters): if parameters[-1] is ... and sys.version_info < (3, 9, 2): # Hack: Arguments must be types, replace it with one. parameters = (parameters[:-1], _EllipsisDummy) if sys.version_info >= (3, 10, 3): concatenate = _ConcatenateGenericAlias(origin, parameters, _typevar_types=(TypeVar, ParamSpec), _paramspec_tvars=True) else: concatenate = _ConcatenateGenericAlias(origin, parameters) if parameters[-1] is not _EllipsisDummy: return concatenate # Remove dummy again concatenate.__args__ = tuple(p if p is not _EllipsisDummy else ... for p in concatenate.__args__) if sys.version_info < (3, 10): # backport needs __args__ adjustment only return concatenate concatenate.__parameters__ = tuple(p for p in concatenate.__parameters__ if p is not _EllipsisDummy) return concatenate # 3.8-3.10 @typing._tp_cache def _concatenate_getitem(self, parameters): if parameters == (): raise TypeError("Cannot take a Concatenate of no types.") if not isinstance(parameters, tuple): parameters = (parameters,) if not (parameters[-1] is ... or isinstance(parameters[-1], ParamSpec)): raise TypeError("The last parameter to Concatenate should be a " "ParamSpec variable or ellipsis.") msg = "Concatenate[arg, ...]: each arg must be a type." parameters = ((typing._type_check(p, msg) for p in parameters[:-1]), parameters[-1]) return _create_concatenate_alias(self, parameters) # 3.11+; Concatenate does not accept ellipsis in 3.10 if sys.version_info >= (3, 11): Concatenate = typing.Concatenate # 3.9-3.10 elif sys.version_info[:2] >= (3, 9): @_ExtensionsSpecialForm def Concatenate(self, parameters): """Used in conjunction with ``ParamSpec`` and ``Callable`` to represent a higher order function which adds, removes or transforms parameters of a callable. For example:: Callable[Concatenate[int, P], int] See PEP 612 for detailed information. """ return _concatenate_getitem(self, parameters) # 3.8 else: class _ConcatenateForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): return _concatenate_getitem(self, parameters) Concatenate = _ConcatenateForm( 'Concatenate', doc="""Used in conjunction with ``ParamSpec`` and ``Callable`` to represent a higher order function which adds, removes or transforms parameters of a callable. For example:: Callable[Concatenate[int, P], int] See PEP 612 for detailed information. """) # 3.10+ if hasattr(typing, 'TypeGuard'): TypeGuard = typing.TypeGuard # 3.9 elif sys.version_info[:2] >= (3, 9): @_ExtensionsSpecialForm def TypeGuard(self, parameters): """Special typing form used to annotate the return type of a user-defined type guard function. ``TypeGuard`` only accepts a single type argument. At runtime, functions marked this way should return a boolean. ``TypeGuard`` aims to benefit type narrowing -- a technique used by static type checkers to determine a more precise type of an expression within a program's code flow. Usually type narrowing is done by analyzing conditional code flow and applying the narrowing to a block of code. The conditional expression here is sometimes referred to as a "type guard". Sometimes it would be convenient to use a user-defined boolean function as a type guard. Such a function should use ``TypeGuard[...]`` as its return type to alert static type checkers to this intention. Using ``-> TypeGuard`` tells the static type checker that for a given function: 1. The return value is a boolean. 2. If the return value is ``True``, the type of its argument is the type inside ``TypeGuard``. For example:: def is_str(val: Union[str, float]): # "isinstance" type guard if isinstance(val, str): # Type of ``val`` is narrowed to ``str`` ... else: # Else, type of ``val`` is narrowed to ``float``. ... Strict type narrowing is not enforced -- ``TypeB`` need not be a narrower form of ``TypeA`` (it can even be a wider form) and this may lead to type-unsafe results. The main reason is to allow for things like narrowing ``List[object]`` to ``List[str]`` even though the latter is not a subtype of the former, since ``List`` is invariant. The responsibility of writing type-safe type guards is left to the user. ``TypeGuard`` also works with type variables. For more information, see PEP 647 (User-Defined Type Guards). """ item = typing._type_check(parameters, f'{self} accepts only a single type.') return typing._GenericAlias(self, (item,)) # 3.8 else: class _TypeGuardForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type') return typing._GenericAlias(self, (item,)) TypeGuard = _TypeGuardForm( 'TypeGuard', doc="""Special typing form used to annotate the return type of a user-defined type guard function. ``TypeGuard`` only accepts a single type argument. At runtime, functions marked this way should return a boolean. ``TypeGuard`` aims to benefit type narrowing -- a technique used by static type checkers to determine a more precise type of an expression within a program's code flow. Usually type narrowing is done by analyzing conditional code flow and applying the narrowing to a block of code. The conditional expression here is sometimes referred to as a "type guard". Sometimes it would be convenient to use a user-defined boolean function as a type guard. Such a function should use ``TypeGuard[...]`` as its return type to alert static type checkers to this intention. Using ``-> TypeGuard`` tells the static type checker that for a given function: 1. The return value is a boolean. 2. If the return value is ``True``, the type of its argument is the type inside ``TypeGuard``. For example:: def is_str(val: Union[str, float]): # "isinstance" type guard if isinstance(val, str): # Type of ``val`` is narrowed to ``str`` ... else: # Else, type of ``val`` is narrowed to ``float``. ... Strict type narrowing is not enforced -- ``TypeB`` need not be a narrower form of ``TypeA`` (it can even be a wider form) and this may lead to type-unsafe results. The main reason is to allow for things like narrowing ``List[object]`` to ``List[str]`` even though the latter is not a subtype of the former, since ``List`` is invariant. The responsibility of writing type-safe type guards is left to the user. ``TypeGuard`` also works with type variables. For more information, see PEP 647 (User-Defined Type Guards). """) # 3.13+ if hasattr(typing, 'TypeIs'): TypeIs = typing.TypeIs # 3.9 elif sys.version_info[:2] >= (3, 9): @_ExtensionsSpecialForm def TypeIs(self, parameters): """Special typing form used to annotate the return type of a user-defined type narrower function. ``TypeIs`` only accepts a single type argument. At runtime, functions marked this way should return a boolean. ``TypeIs`` aims to benefit type narrowing -- a technique used by static type checkers to determine a more precise type of an expression within a program's code flow. Usually type narrowing is done by analyzing conditional code flow and applying the narrowing to a block of code. The conditional expression here is sometimes referred to as a "type guard". Sometimes it would be convenient to use a user-defined boolean function as a type guard. Such a function should use ``TypeIs[...]`` as its return type to alert static type checkers to this intention. Using ``-> TypeIs`` tells the static type checker that for a given function: 1. The return value is a boolean. 2. If the return value is ``True``, the type of its argument is the intersection of the type inside ``TypeIs`` and the argument's previously known type. For example:: def is_awaitable(val: object) -> TypeIs[Awaitable[Any]]: return hasattr(val, '__await__') def f(val: Union[int, Awaitable[int]]) -> int: if is_awaitable(val): assert_type(val, Awaitable[int]) else: assert_type(val, int) ``TypeIs`` also works with type variables. For more information, see PEP 742 (Narrowing types with TypeIs). """ item = typing._type_check(parameters, f'{self} accepts only a single type.') return typing._GenericAlias(self, (item,)) # 3.8 else: class _TypeIsForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type') return typing._GenericAlias(self, (item,)) TypeIs = _TypeIsForm( 'TypeIs', doc="""Special typing form used to annotate the return type of a user-defined type narrower function. ``TypeIs`` only accepts a single type argument. At runtime, functions marked this way should return a boolean. ``TypeIs`` aims to benefit type narrowing -- a technique used by static type checkers to determine a more precise type of an expression within a program's code flow. Usually type narrowing is done by analyzing conditional code flow and applying the narrowing to a block of code. The conditional expression here is sometimes referred to as a "type guard". Sometimes it would be convenient to use a user-defined boolean function as a type guard. Such a function should use ``TypeIs[...]`` as its return type to alert static type checkers to this intention. Using ``-> TypeIs`` tells the static type checker that for a given function: 1. The return value is a boolean. 2. If the return value is ``True``, the type of its argument is the intersection of the type inside ``TypeIs`` and the argument's previously known type. For example:: def is_awaitable(val: object) -> TypeIs[Awaitable[Any]]: return hasattr(val, '__await__') def f(val: Union[int, Awaitable[int]]) -> int: if is_awaitable(val): assert_type(val, Awaitable[int]) else: assert_type(val, int) ``TypeIs`` also works with type variables. For more information, see PEP 742 (Narrowing types with TypeIs). """) # 3.14+? if hasattr(typing, 'TypeForm'): TypeForm = typing.TypeForm # 3.9 elif sys.version_info[:2] >= (3, 9): class _TypeFormForm(_ExtensionsSpecialForm, _root=True): # TypeForm(X) is equivalent to X but indicates to the type checker # that the object is a TypeForm. def __call__(self, obj, /): return obj @_TypeFormForm def TypeForm(self, parameters): """A special form representing the value that results from the evaluation of a type expression. This value encodes the information supplied in the type expression, and it represents the type described by that type expression. When used in a type expression, TypeForm describes a set of type form objects. It accepts a single type argument, which must be a valid type expression. ``TypeForm[T]`` describes the set of all type form objects that represent the type T or types that are assignable to T. Usage: def cast[T](typ: TypeForm[T], value: Any) -> T: ... reveal_type(cast(int, "x")) # int See PEP 747 for more information. """ item = typing._type_check(parameters, f'{self} accepts only a single type.') return typing._GenericAlias(self, (item,)) # 3.8 else: class _TypeFormForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type') return typing._GenericAlias(self, (item,)) def __call__(self, obj, /): return obj TypeForm = _TypeFormForm( 'TypeForm', doc="""A special form representing the value that results from the evaluation of a type expression. This value encodes the information supplied in the type expression, and it represents the type described by that type expression. When used in a type expression, TypeForm describes a set of type form objects. It accepts a single type argument, which must be a valid type expression. ``TypeForm[T]`` describes the set of all type form objects that represent the type T or types that are assignable to T. Usage: def cast[T](typ: TypeForm[T], value: Any) -> T: ... reveal_type(cast(int, "x")) # int See PEP 747 for more information. """) # Vendored from cpython typing._SpecialFrom class _SpecialForm(typing._Final, _root=True): __slots__ = ('_name', '__doc__', '_getitem') def __init__(self, getitem): self._getitem = getitem self._name = getitem.__name__ self.__doc__ = getitem.__doc__ def __getattr__(self, item): if item in {'__name__', '__qualname__'}: return self._name raise AttributeError(item) def __mro_entries__(self, bases): raise TypeError(f"Cannot subclass {self!r}") def __repr__(self): return f'typing_extensions.{self._name}' def __reduce__(self): return self._name def __call__(self, args, kwds): raise TypeError(f"Cannot instantiate {self!r}") def __or__(self, other): return typing.Union[self, other] def __ror__(self, other): return typing.Union[other, self] def __instancecheck__(self, obj): raise TypeError(f"{self} cannot be used with isinstance()") def __subclasscheck__(self, cls): raise TypeError(f"{self} cannot be used with issubclass()") @typing._tp_cache def __getitem__(self, parameters): return self._getitem(self, parameters) if hasattr(typing, "LiteralString"): # 3.11+ LiteralString = typing.LiteralString else: @_SpecialForm def LiteralString(self, params): """Represents an arbitrary literal string. Example:: from typing_extensions import LiteralString def query(sql: LiteralString) -> ...: ... query("SELECT FROM table") # ok query(f"SELECT * FROM {input()}") # not ok See PEP 675 for details. """ raise TypeError(f"{self} is not subscriptable") if hasattr(typing, "Self"): # 3.11+ Self = typing.Self else: @_SpecialForm def Self(self, params): """Used to spell the type of "self" in classes. Example:: from typing import Self class ReturnsSelf: def parse(self, data: bytes) -> Self: ... return self """ raise TypeError(f"{self} is not subscriptable") if hasattr(typing, "Never"): # 3.11+ Never = typing.Never else: @_SpecialForm def Never(self, params): """The bottom type, a type that has no members. This can be used to define a function that should never be called, or a function that never returns:: from typing_extensions import Never def never_call_me(arg: Never) -> None: pass def int_or_str(arg: int \| str) -> None: never_call_me(arg) # type checker error match arg: case int(): print("It's an int") case str(): print("It's a str") case _: never_call_me(arg) # ok, arg is of type Never """ raise TypeError(f"{self} is not subscriptable") if hasattr(typing, 'Required'): # 3.11+ Required = typing.Required NotRequired = typing.NotRequired elif sys.version_info[:2] >= (3, 9): # 3.9-3.10 @_ExtensionsSpecialForm def Required(self, parameters): """A special typing construct to mark a key of a total=False TypedDict as required. For example: class Movie(TypedDict, total=False): title: Required[str] year: int m = Movie( title='The Matrix', # typechecker error if key is omitted year=1999, ) There is no runtime checking that a required key is actually provided when instantiating a related TypedDict. """ item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return typing._GenericAlias(self, (item,)) @_ExtensionsSpecialForm def NotRequired(self, parameters): """A special typing construct to mark a key of a TypedDict as potentially missing. For example: class Movie(TypedDict): title: str year: NotRequired[int] m = Movie( title='The Matrix', # typechecker error if key is omitted year=1999, ) """ item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return typing._GenericAlias(self, (item,)) else: # 3.8 class _RequiredForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return typing._GenericAlias(self, (item,)) Required = _RequiredForm( 'Required', doc="""A special typing construct to mark a key of a total=False TypedDict as required. For example: class Movie(TypedDict, total=False): title: Required[str] year: int m = Movie( title='The Matrix', # typechecker error if key is omitted year=1999, ) There is no runtime checking that a required key is actually provided when instantiating a related TypedDict. """) NotRequired = _RequiredForm( 'NotRequired', doc="""A special typing construct to mark a key of a TypedDict as potentially missing. For example: class Movie(TypedDict): title: str year: NotRequired[int] m = Movie( title='The Matrix', # typechecker error if key is omitted year=1999, ) """) if hasattr(typing, 'ReadOnly'): ReadOnly = typing.ReadOnly elif sys.version_info[:2] >= (3, 9): # 3.9-3.12 @_ExtensionsSpecialForm def ReadOnly(self, parameters): """A special typing construct to mark an item of a TypedDict as read-only. For example: class Movie(TypedDict): title: ReadOnly[str] year: int def mutate_movie(m: Movie) -> None: m["year"] = 1992 # allowed m["title"] = "The Matrix" # typechecker error There is no runtime checking for this property. """ item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return typing._GenericAlias(self, (item,)) else: # 3.8 class _ReadOnlyForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return typing._GenericAlias(self, (item,)) ReadOnly = _ReadOnlyForm( 'ReadOnly', doc="""A special typing construct to mark a key of a TypedDict as read-only. For example: class Movie(TypedDict): title: ReadOnly[str] year: int def mutate_movie(m: Movie) -> None: m["year"] = 1992 # allowed m["title"] = "The Matrix" # typechecker error There is no runtime checking for this propery. """) _UNPACK_DOC = """\ Type unpack operator. The type unpack operator takes the child types from some container type, such as `tuple[int, str]` or a `TypeVarTuple`, and 'pulls them out'. For example: # For some generic class `Foo`: Foo[Unpack[tuple[int, str]]] # Equivalent to Foo[int, str] Ts = TypeVarTuple('Ts') # Specifies that `Bar` is generic in an arbitrary number of types. # (Think of `Ts` as a tuple of an arbitrary number of individual # `TypeVar`s, which the `Unpack` is 'pulling out' directly into the # `Generic[]`.) class Bar(Generic[Unpack[Ts]]): ... Bar[int] # Valid Bar[int, str] # Also valid From Python 3.11, this can also be done using the `` operator: Foo[tuple[int, str]] class Bar(Generic[Ts]): ... The operator can also be used along with a `TypedDict` to annotate `kwargs` in a function signature. For instance: class Movie(TypedDict): name: str year: int # This function expects two keyword arguments - name* of type `str` and # year of type `int`. def foo(*kwargs: Unpack[Movie]): ... Note that there is only some runtime checking of this operator. Not everything the runtime allows may be accepted by static type checkers. For more information, see PEP 646 and PEP 692. """ if sys.version_info >= (3, 12): # PEP 692 changed the repr of Unpack[] Unpack = typing.Unpack def _is_unpack(obj): return get_origin(obj) is Unpack elif sys.version_info[:2] >= (3, 9): # 3.9+ class _UnpackSpecialForm(_ExtensionsSpecialForm, _root=True): def __init__(self, getitem): super().__init__(getitem) self.__doc__ = _UNPACK_DOC class _UnpackAlias(typing._GenericAlias, _root=True): if sys.version_info < (3, 11): # needed for compatibility with Generic[Unpack[Ts]] __class__ = typing.TypeVar @property def __typing_unpacked_tuple_args__(self): assert self.__origin__ is Unpack assert len(self.__args__) == 1 arg, = self.__args__ if isinstance(arg, (typing._GenericAlias, _types.GenericAlias)): if arg.__origin__ is not tuple: raise TypeError("Unpack[...] must be used with a tuple type") return arg.__args__ return None @property def __typing_is_unpacked_typevartuple__(self): assert self.__origin__ is Unpack assert len(self.__args__) == 1 return isinstance(self.__args__[0], TypeVarTuple) def __getitem__(self, args): if self.__typing_is_unpacked_typevartuple__: return args return super().__getitem__(args) @_UnpackSpecialForm def Unpack(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return _UnpackAlias(self, (item,)) def _is_unpack(obj): return isinstance(obj, _UnpackAlias) else: # 3.8 class _UnpackAlias(typing._GenericAlias, _root=True): __class__ = typing.TypeVar @property def __typing_unpacked_tuple_args__(self): assert self.__origin__ is Unpack assert len(self.__args__) == 1 arg, = self.__args__ if isinstance(arg, typing._GenericAlias): if arg.__origin__ is not tuple: raise TypeError("Unpack[...] must be used with a tuple type") return arg.__args__ return None @property def __typing_is_unpacked_typevartuple__(self): assert self.__origin__ is Unpack assert len(self.__args__) == 1 return isinstance(self.__args__[0], TypeVarTuple) def __getitem__(self, args): if self.__typing_is_unpacked_typevartuple__: return args return super().__getitem__(args) class _UnpackForm(_ExtensionsSpecialForm, _root=True): def __getitem__(self, parameters): item = typing._type_check(parameters, f'{self._name} accepts only a single type.') return _UnpackAlias(self, (item,)) Unpack = _UnpackForm('Unpack', doc=_UNPACK_DOC) def _is_unpack(obj): return isinstance(obj, _UnpackAlias) def _unpack_args(args): newargs = [] for arg in args: subargs = getattr(arg, '__typing_unpacked_tuple_args__', None) if subargs is not None and (not (subargs and subargs[-1] is ...)): newargs.extend(subargs) else: newargs.append(arg) return newargs if _PEP_696_IMPLEMENTED: from typing import TypeVarTuple elif hasattr(typing, "TypeVarTuple"): # 3.11+ # Add default parameter - PEP 696 class TypeVarTuple(metaclass=_TypeVarLikeMeta): """Type variable tuple.""" _backported_typevarlike = typing.TypeVarTuple def __new__(cls, name, , default=NoDefault): tvt = typing.TypeVarTuple(name) _set_default(tvt, default) _set_module(tvt) def _typevartuple_prepare_subst(alias, args): params = alias.__parameters__ typevartuple_index = params.index(tvt) for param in params[typevartuple_index + 1:]: if isinstance(param, TypeVarTuple): raise TypeError( f"More than one TypeVarTuple parameter in {alias}" ) alen = len(args) plen = len(params) left = typevartuple_index right = plen - typevartuple_index - 1 var_tuple_index = None fillarg = None for k, arg in enumerate(args): if not isinstance(arg, type): subargs = getattr(arg, '__typing_unpacked_tuple_args__', None) if subargs and len(subargs) == 2 and subargs[-1] is ...: if var_tuple_index is not None: raise TypeError( "More than one unpacked " "arbitrary-length tuple argument" ) var_tuple_index = k fillarg = subargs[0] if var_tuple_index is not None: left = min(left, var_tuple_index) right = min(right, alen - var_tuple_index - 1) elif left + right > alen: raise TypeError(f"Too few arguments for {alias};" f" actual {alen}, expected at least {plen - 1}") if left == alen - right and tvt.has_default(): replacement = _unpack_args(tvt.__default__) else: replacement = args[left: alen - right] return ( args[:left], ([fillarg] (typevartuple_index - left)), replacement, ([fillarg] (plen - right - left - typevartuple_index - 1)), args[alen - right:], ) tvt.__typing_prepare_subst__ = _typevartuple_prepare_subst return tvt def __init_subclass__(self, args, *kwds): raise TypeError("Cannot subclass special typing classes") else: # <=3.10 class TypeVarTuple(_DefaultMixin): """Type variable tuple. Usage:: Ts = TypeVarTuple('Ts') In the same way that a normal type variable is a stand-in for a single type such as ``int``, a type variable tuple* is a stand-in for a tuple type such as ``Tuple[int, str]``. Type variable tuples can be used in ``Generic`` declarations. Consider the following example:: class Array(Generic[Ts]): ... The ``Ts`` type variable tuple here behaves like ``tuple[T1, T2]``, where ``T1`` and ``T2`` are type variables. To use these type variables as type parameters of ``Array``, we must unpack* the type variable tuple using the star operator: ``Ts``. The signature of ``Array`` then behaves as if we had simply written ``class Array(Generic[T1, T2]): ...``. In contrast to ``Generic[T1, T2]``, however, ``Generic[Shape]`` allows us to parameterise the class with an arbitrary number of type parameters. Type variable tuples can be used anywhere a normal ``TypeVar`` can. This includes class definitions, as shown above, as well as function signatures and variable annotations:: class Array(Generic[Ts]): def __init__(self, shape: Tuple[Ts]): self._shape: Tuple[Ts] = shape def get_shape(self) -> Tuple[Ts]: return self._shape shape = (Height(480), Width(640)) x: Array[Height, Width] = Array(shape) y = abs(x) # Inferred type is Array[Height, Width] z = x + x # ... is Array[Height, Width] x.get_shape() # ... is tuple[Height, Width] """ # Trick Generic __parameters__. __class__ = typing.TypeVar def __iter__(self): yield self.__unpacked__ def __init__(self, name, , default=NoDefault): self.__name__ = name _DefaultMixin.__init__(self, default) # for pickling: def_mod = _caller() if def_mod != 'typing_extensions': self.__module__ = def_mod self.__unpacked__ = Unpack[self] def __repr__(self): return self.__name__ def __hash__(self): return object.__hash__(self) def __eq__(self, other): return self is other def __reduce__(self): return self.__name__ def __init_subclass__(self, args, *kwds): if '_root' not in kwds: raise TypeError("Cannot subclass special typing classes") if hasattr(typing, "reveal_type"): # 3.11+ reveal_type = typing.reveal_type else: # <=3.10 def reveal_type(obj: T, /) -> T: """Reveal the inferred type of a variable. When a static type checker encounters a call to ``reveal_type()``, it will emit the inferred type of the argument:: x: int = 1 reveal_type(x) Running a static type checker (e.g., ``mypy``) on this example will produce output similar to 'Revealed type is "builtins.int"'. At runtime, the function prints the runtime type of the argument and returns it unchanged. """ print(f"Runtime type is {type(obj).__name__!r}", file=sys.stderr) return obj if hasattr(typing, "_ASSERT_NEVER_REPR_MAX_LENGTH"): # 3.11+ _ASSERT_NEVER_REPR_MAX_LENGTH = typing._ASSERT_NEVER_REPR_MAX_LENGTH else: # <=3.10 _ASSERT_NEVER_REPR_MAX_LENGTH = 100 if hasattr(typing, "assert_never"): # 3.11+ assert_never = typing.assert_never else: # <=3.10 def assert_never(arg: Never, /) -> Never: """Assert to the type checker that a line of code is unreachable. Example:: def int_or_str(arg: int \| str) -> None: match arg: case int(): print("It's an int") case str(): print("It's a str") case _: assert_never(arg) If a type checker finds that a call to assert_never() is reachable, it will emit an error. At runtime, this throws an exception when called. """ value = repr(arg) if len(value) > _ASSERT_NEVER_REPR_MAX_LENGTH: value = value[:_ASSERT_NEVER_REPR_MAX_LENGTH] + '...' raise AssertionError(f"Expected code to be unreachable, but got: {value}") if sys.version_info >= (3, 12): # 3.12+ # dataclass_transform exists in 3.11 but lacks the frozen_default parameter dataclass_transform = typing.dataclass_transform else: # <=3.11 def dataclass_transform( , eq_default: bool = True, order_default: bool = False, kw_only_default: bool = False, frozen_default: bool = False, field_specifiers: typing.Tuple[ typing.Union[typing.Type[typing.Any], typing.Callable[..., typing.Any]], ... ] = (), *kwargs: typing.Any, ) -> typing.Callable[[T], T]: """Decorator that marks a function, class, or metaclass as providing dataclass-like behavior. Example: from typing_extensions import dataclass_transform _T = TypeVar("_T") # Used on a decorator function @dataclass_transform() def create_model(cls: type[_T]) -> type[_T]: ... return cls @create_model class CustomerModel: id: int name: str # Used on a base class @dataclass_transform() class ModelBase: ... class CustomerModel(ModelBase): id: int name: str # Used on a metaclass @dataclass_transform() class ModelMeta(type): ... class ModelBase(metaclass=ModelMeta): ... class CustomerModel(ModelBase): id: int name: str Each of the ``CustomerModel`` classes defined in this example will now behave similarly to a dataclass created with the ``@dataclasses.dataclass`` decorator. For example, the type checker will synthesize an ``__init__`` method. The arguments to this decorator can be used to customize this behavior: - ``eq_default`` indicates whether the ``eq`` parameter is assumed to be True or False if it is omitted by the caller. - ``order_default`` indicates whether the ``order`` parameter is assumed to be True or False if it is omitted by the caller. - ``kw_only_default`` indicates whether the ``kw_only`` parameter is assumed to be True or False if it is omitted by the caller. - ``frozen_default`` indicates whether the ``frozen`` parameter is assumed to be True or False if it is omitted by the caller. - ``field_specifiers`` specifies a static list of supported classes or functions that describe fields, similar to ``dataclasses.field()``. At runtime, this decorator records its arguments in the ``__dataclass_transform__`` attribute on the decorated object. See PEP 681 for details. """ def decorator(cls_or_fn): cls_or_fn.__dataclass_transform__ = { "eq_default": eq_default, "order_default": order_default, "kw_only_default": kw_only_default, "frozen_default": frozen_default, "field_specifiers": field_specifiers, "kwargs": kwargs, } return cls_or_fn return decorator if hasattr(typing, "override"): # 3.12+ override = typing.override else: # <=3.11 _F = typing.TypeVar("_F", bound=typing.Callable[..., typing.Any]) def override(arg: _F, /) -> _F: """Indicate that a method is intended to override a method in a base class. Usage: class Base: def method(self) -> None: pass class Child(Base): @override def method(self) -> None: super().method() When this decorator is applied to a method, the type checker will validate that it overrides a method with the same name on a base class. This helps prevent bugs that may occur when a base class is changed without an equivalent change to a child class. There is no runtime checking of these properties. The decorator sets the ``__override__`` attribute to ``True`` on the decorated object to allow runtime introspection. See PEP 698 for details. """ try: arg.__override__ = True except (AttributeError, TypeError): # Skip the attribute silently if it is not writable. # AttributeError happens if the object has __slots__ or a # read-only property, TypeError if it's a builtin class. pass return arg # Python 3.13.3+ contains a fix for the wrapped __new__ if sys.version_info >= (3, 13, 3): deprecated = warnings.deprecated else: _T = typing.TypeVar("_T") class deprecated: """Indicate that a class, function or overload is deprecated. When this decorator is applied to an object, the type checker will generate a diagnostic on usage of the deprecated object. Usage: @deprecated("Use B instead") class A: pass @deprecated("Use g instead") def f(): pass @overload @deprecated("int support is deprecated") def g(x: int) -> int: ... @overload def g(x: str) -> int: ... The warning specified by category* will be emitted at runtime on use of deprecated objects. For functions, that happens on calls; for classes, on instantiation and on creation of subclasses. If the category is ``None``, no warning is emitted at runtime. The stacklevel determines where the warning is emitted. If it is ``1`` (the default), the warning is emitted at the direct caller of the deprecated object; if it is higher, it is emitted further up the stack. Static type checker behavior is not affected by the category and stacklevel arguments. The deprecation message passed to the decorator is saved in the ``__deprecated__`` attribute on the decorated object. If applied to an overload, the decorator must be after the ``@overload`` decorator for the attribute to exist on the overload as returned by ``get_overloads()``. See PEP 702 for details. """ def __init__( self, message: str, /, , category: typing.Optional[typing.Type[Warning]] = DeprecationWarning, stacklevel: int = 1, ) -> None: if not isinstance(message, str): raise TypeError( "Expected an object of type str for 'message', not " f"{type(message).__name__!r}" ) self.message = message self.category = category self.stacklevel = stacklevel def __call__(self, arg: _T, /) -> _T: # Make sure the inner functions created below don't # retain a reference to self. msg = self.message category = self.category stacklevel = self.stacklevel if category is None: arg.__deprecated__ = msg return arg elif isinstance(arg, type): import functools from types import MethodType original_new = arg.__new__ @functools.wraps(original_new) def __new__(cls, /, args, *kwargs): if cls is arg: warnings.warn(msg, category=category, stacklevel=stacklevel + 1) if original_new is not object.__new__: return original_new(cls, args, *kwargs) # Mirrors a similar check in object.__new__. elif cls.__init__ is object.__init__ and (args or kwargs): raise TypeError(f"{cls.__name__}() takes no arguments") else: return original_new(cls) arg.__new__ = staticmethod(__new__) original_init_subclass = arg.__init_subclass__ # We need slightly different behavior if __init_subclass__ # is a bound method (likely if it was implemented in Python) if isinstance(original_init_subclass, MethodType): original_init_subclass = original_init_subclass.__func__ @functools.wraps(original_init_subclass) def __init_subclass__(args, *kwargs): warnings.warn(msg, category=category, stacklevel=stacklevel + 1) return original_init_subclass(args, *kwargs) arg.__init_subclass__ = classmethod(__init_subclass__) # Or otherwise, which likely means it's a builtin such as # object's implementation of __init_subclass__. else: @functools.wraps(original_init_subclass) def __init_subclass__(args, *kwargs): warnings.warn(msg, category=category, stacklevel=stacklevel + 1) return original_init_subclass(args, *kwargs) arg.__init_subclass__ = __init_subclass__ arg.__deprecated__ = __new__.__deprecated__ = msg __init_subclass__.__deprecated__ = msg return arg elif callable(arg): import asyncio.coroutines import functools import inspect @functools.wraps(arg) def wrapper(args, *kwargs): warnings.warn(msg, category=category, stacklevel=stacklevel + 1) return arg(args, *kwargs) if asyncio.coroutines.iscoroutinefunction(arg): if sys.version_info >= (3, 12): wrapper = inspect.markcoroutinefunction(wrapper) else: wrapper._is_coroutine = asyncio.coroutines._is_coroutine arg.__deprecated__ = wrapper.__deprecated__ = msg return wrapper else: raise TypeError( "@deprecated decorator with non-None category must be applied to " f"a class or callable, not {arg!r}" ) if sys.version_info < (3, 10): def _is_param_expr(arg): return arg is ... or isinstance( arg, (tuple, list, ParamSpec, _ConcatenateGenericAlias) ) else: def _is_param_expr(arg): return arg is ... or isinstance( arg, ( tuple, list, ParamSpec, _ConcatenateGenericAlias, typing._ConcatenateGenericAlias, ), ) # We have to do some monkey patching to deal with the dual nature of # Unpack/TypeVarTuple: # - We want Unpack to be a kind of TypeVar so it gets accepted in # Generic[Unpack[Ts]] # - We want it to not* be treated as a TypeVar for the purposes of # counting generic parameters, so that when we subscript a generic, # the runtime doesn't try to substitute the Unpack with the subscripted type. if not hasattr(typing, "TypeVarTuple"): def _check_generic(cls, parameters, elen=_marker): """Check correct count for parameters of a generic cls (internal helper). This gives a nice error message in case of count mismatch. """ # If substituting a single ParamSpec with multiple arguments # we do not check the count if (inspect.isclass(cls) and issubclass(cls, typing.Generic) and len(cls.__parameters__) == 1 and isinstance(cls.__parameters__[0], ParamSpec) and parameters and not _is_param_expr(parameters[0]) ): # Generic modifies parameters variable, but here we cannot do this return if not elen: raise TypeError(f"{cls} is not a generic class") if elen is _marker: if not hasattr(cls, "__parameters__") or not cls.__parameters__: raise TypeError(f"{cls} is not a generic class") elen = len(cls.__parameters__) alen = len(parameters) if alen != elen: expect_val = elen if hasattr(cls, "__parameters__"): parameters = [p for p in cls.__parameters__ if not _is_unpack(p)] num_tv_tuples = sum(isinstance(p, TypeVarTuple) for p in parameters) if (num_tv_tuples > 0) and (alen >= elen - num_tv_tuples): return # deal with TypeVarLike defaults # required TypeVarLikes cannot appear after a defaulted one. if alen < elen: # since we validate TypeVarLike default in _collect_type_vars # or _collect_parameters we can safely check parameters[alen] if ( getattr(parameters[alen], '__default__', NoDefault) is not NoDefault ): return num_default_tv = sum(getattr(p, '__default__', NoDefault) is not NoDefault for p in parameters) elen -= num_default_tv expect_val = f"at least {elen}" things = "arguments" if sys.version_info >= (3, 10) else "parameters" raise TypeError(f"Too {'many' if alen > elen else 'few'} {things}" f" for {cls}; actual {alen}, expected {expect_val}") else: # Python 3.11+ def _check_generic(cls, parameters, elen): """Check correct count for parameters of a generic cls (internal helper). This gives a nice error message in case of count mismatch. """ if not elen: raise TypeError(f"{cls} is not a generic class") alen = len(parameters) if alen != elen: expect_val = elen if hasattr(cls, "__parameters__"): parameters = [p for p in cls.__parameters__ if not _is_unpack(p)] # deal with TypeVarLike defaults # required TypeVarLikes cannot appear after a defaulted one. if alen < elen: # since we validate TypeVarLike default in _collect_type_vars # or _collect_parameters we can safely check parameters[alen] if ( getattr(parameters[alen], '__default__', NoDefault) is not NoDefault ): return num_default_tv = sum(getattr(p, '__default__', NoDefault) is not NoDefault for p in parameters) elen -= num_default_tv expect_val = f"at least {elen}" raise TypeError(f"Too {'many' if alen > elen else 'few'} arguments" f" for {cls}; actual {alen}, expected {expect_val}") if not _PEP_696_IMPLEMENTED: typing._check_generic = _check_generic def _has_generic_or_protocol_as_origin() -> bool: try: frame = sys._getframe(2) # - Catch AttributeError: not all Python implementations have sys._getframe() # - Catch ValueError: maybe we're called from an unexpected module # and the call stack isn't deep enough except (AttributeError, ValueError): return False # err on the side of leniency else: # If we somehow get invoked from outside typing.py, # also err on the side of leniency if frame.f_globals.get("__name__") != "typing": return False origin = frame.f_locals.get("origin") # Cannot use "in" because origin may be an object with a buggy __eq__ that # throws an error. return origin is typing.Generic or origin is Protocol or origin is typing.Protocol _TYPEVARTUPLE_TYPES = {TypeVarTuple, getattr(typing, "TypeVarTuple", None)} def _is_unpacked_typevartuple(x) -> bool: if get_origin(x) is not Unpack: return False args = get_args(x) return ( bool(args) and len(args) == 1 and type(args[0]) in _TYPEVARTUPLE_TYPES ) # Python 3.11+ _collect_type_vars was renamed to _collect_parameters if hasattr(typing, '_collect_type_vars'): def _collect_type_vars(types, typevar_types=None): """Collect all type variable contained in types in order of first appearance (lexicographic order). For example:: _collect_type_vars((T, List[S, T])) == (T, S) """ if typevar_types is None: typevar_types = typing.TypeVar tvars = [] # A required TypeVarLike cannot appear after a TypeVarLike with a default # if it was a direct call to `Generic[]` or `Protocol[]` enforce_default_ordering = _has_generic_or_protocol_as_origin() default_encountered = False # Also, a TypeVarLike with a default cannot appear after a TypeVarTuple type_var_tuple_encountered = False for t in types: if _is_unpacked_typevartuple(t): type_var_tuple_encountered = True elif ( isinstance(t, typevar_types) and not isinstance(t, _UnpackAlias) and t not in tvars ): if enforce_default_ordering: has_default = getattr(t, '__default__', NoDefault) is not NoDefault if has_default: if type_var_tuple_encountered: raise TypeError('Type parameter with a default' ' follows TypeVarTuple') default_encountered = True elif default_encountered: raise TypeError(f'Type parameter {t!r} without a default' ' follows type parameter with a default') tvars.append(t) if _should_collect_from_parameters(t): tvars.extend([t for t in t.__parameters__ if t not in tvars]) elif isinstance(t, tuple): # Collect nested type_vars # tuple wrapped by _prepare_paramspec_params(cls, params) for x in t: for collected in _collect_type_vars([x]): if collected not in tvars: tvars.append(collected) return tuple(tvars) typing._collect_type_vars = _collect_type_vars else: def _collect_parameters(args): """Collect all type variables and parameter specifications in args in order of first appearance (lexicographic order). For example:: assert _collect_parameters((T, Callable[P, T])) == (T, P) """ parameters = [] # A required TypeVarLike cannot appear after a TypeVarLike with default # if it was a direct call to `Generic[]` or `Protocol[]` enforce_default_ordering = _has_generic_or_protocol_as_origin() default_encountered = False # Also, a TypeVarLike with a default cannot appear after a TypeVarTuple type_var_tuple_encountered = False for t in args: if isinstance(t, type): # We don't want __parameters__ descriptor of a bare Python class. pass elif isinstance(t, tuple): # `t` might be a tuple, when `ParamSpec` is substituted with # `[T, int]`, or `[int, Ts]`, etc. for x in t: for collected in _collect_parameters([x]): if collected not in parameters: parameters.append(collected) elif hasattr(t, '__typing_subst__'): if t not in parameters: if enforce_default_ordering: has_default = ( getattr(t, '__default__', NoDefault) is not NoDefault ) if type_var_tuple_encountered and has_default: raise TypeError('Type parameter with a default' ' follows TypeVarTuple') if has_default: default_encountered = True elif default_encountered: raise TypeError(f'Type parameter {t!r} without a default' ' follows type parameter with a default') parameters.append(t) else: if _is_unpacked_typevartuple(t): type_var_tuple_encountered = True for x in getattr(t, '__parameters__', ()): if x not in parameters: parameters.append(x) return tuple(parameters) if not _PEP_696_IMPLEMENTED: typing._collect_parameters = _collect_parameters # Backport typing.NamedTuple as it exists in Python 3.13. # In 3.11, the ability to define generic `NamedTuple`s was supported. # This was explicitly disallowed in 3.9-3.10, and only half-worked in <=3.8. # On 3.12, we added __orig_bases__ to call-based NamedTuples # On 3.13, we deprecated kwargs-based NamedTuples if sys.version_info >= (3, 13): NamedTuple = typing.NamedTuple else: def _make_nmtuple(name, types, module, defaults=()): fields = [n for n, t in types] annotations = {n: typing._type_check(t, f"field {n} annotation must be a type") for n, t in types} nm_tpl = collections.namedtuple(name, fields, defaults=defaults, module=module) nm_tpl.__annotations__ = nm_tpl.__new__.__annotations__ = annotations # The `_field_types` attribute was removed in 3.9; # in earlier versions, it is the same as the `__annotations__` attribute if sys.version_info < (3, 9): nm_tpl._field_types = annotations return nm_tpl _prohibited_namedtuple_fields = typing._prohibited _special_namedtuple_fields = frozenset({'__module__', '__name__', '__annotations__'}) class _NamedTupleMeta(type): def __new__(cls, typename, bases, ns): assert _NamedTuple in bases for base in bases: if base is not _NamedTuple and base is not typing.Generic: raise TypeError( 'can only inherit from a NamedTuple type and Generic') bases = tuple(tuple if base is _NamedTuple else base for base in bases) if "__annotations__" in ns: types = ns["__annotations__"] elif "__annotate__" in ns: # TODO: Use inspect.VALUE here, and make the annotations lazily evaluated types = ns["__annotate__"](1) else: types = {} default_names = [] for field_name in types: if field_name in ns: default_names.append(field_name) elif default_names: raise TypeError(f"Non-default namedtuple field {field_name} " f"cannot follow default field" f"{'s' if len(default_names) > 1 else ''} " f"{', '.join(default_names)}") nm_tpl = _make_nmtuple( typename, types.items(), defaults=[ns[n] for n in default_names], module=ns['__module__'] ) nm_tpl.__bases__ = bases if typing.Generic in bases: if hasattr(typing, '_generic_class_getitem'): # 3.12+ nm_tpl.__class_getitem__ = classmethod(typing._generic_class_getitem) else: class_getitem = typing.Generic.__class_getitem__.__func__ nm_tpl.__class_getitem__ = classmethod(class_getitem) # update from user namespace without overriding special namedtuple attributes for key, val in ns.items(): if key in _prohibited_namedtuple_fields: raise AttributeError("Cannot overwrite NamedTuple attribute " + key) elif key not in _special_namedtuple_fields: if key not in nm_tpl._fields: setattr(nm_tpl, key, ns[key]) try: set_name = type(val).__set_name__ except AttributeError: pass else: try: set_name(val, nm_tpl, key) except BaseException as e: msg = ( f"Error calling __set_name__ on {type(val).__name__!r} " f"instance {key!r} in {typename!r}" ) # BaseException.add_note() existed on py311, # but the __set_name__ machinery didn't start # using add_note() until py312. # Making sure exceptions are raised in the same way # as in "normal" classes seems most important here. if sys.version_info >= (3, 12): e.add_note(msg) raise else: raise RuntimeError(msg) from e if typing.Generic in bases: nm_tpl.__init_subclass__() return nm_tpl _NamedTuple = type.__new__(_NamedTupleMeta, 'NamedTuple', (), {}) def _namedtuple_mro_entries(bases): assert NamedTuple in bases return (_NamedTuple,) @_ensure_subclassable(_namedtuple_mro_entries) def NamedTuple(typename, fields=_marker, /, kwargs): """Typed version of namedtuple. Usage:: class Employee(NamedTuple): name: str id: int This is equivalent to:: Employee = collections.namedtuple('Employee', ['name', 'id']) The resulting class has an extra __annotations__ attribute, giving a dict that maps field names to types. (The field names are also in the _fields attribute, which is part of the namedtuple API.) An alternative equivalent functional syntax is also accepted:: Employee = NamedTuple('Employee', [('name', str), ('id', int)]) """ if fields is _marker: if kwargs: deprecated_thing = "Creating NamedTuple classes using keyword arguments" deprecation_msg = ( "{name} is deprecated and will be disallowed in Python {remove}. " "Use the class-based or functional syntax instead." ) else: deprecated_thing = "Failing to pass a value for the 'fields' parameter" example = f"`{typename} = NamedTuple({typename!r}, [])`" deprecation_msg = ( "{name} is deprecated and will be disallowed in Python {remove}. " "To create a NamedTuple class with 0 fields " "using the functional syntax, " "pass an empty list, e.g. " ) + example + "." elif fields is None: if kwargs: raise TypeError( "Cannot pass `None` as the 'fields' parameter " "and also specify fields using keyword arguments" ) else: deprecated_thing = "Passing `None` as the 'fields' parameter" example = f"`{typename} = NamedTuple({typename!r}, [])`" deprecation_msg = ( "{name} is deprecated and will be disallowed in Python {remove}. " "To create a NamedTuple class with 0 fields " "using the functional syntax, " "pass an empty list, e.g. " ) + example + "." elif kwargs: raise TypeError("Either list of fields or keywords" " can be provided to NamedTuple, not both") if fields is _marker or fields is None: warnings.warn( deprecation_msg.format(name=deprecated_thing, remove="3.15"), DeprecationWarning, stacklevel=2, ) fields = kwargs.items() nt = _make_nmtuple(typename, fields, module=_caller()) nt.__orig_bases__ = (NamedTuple,) return nt if hasattr(collections.abc, "Buffer"): Buffer = collections.abc.Buffer else: class Buffer(abc.ABC): # noqa: B024 """Base class for classes that implement the buffer protocol. The buffer protocol allows Python objects to expose a low-level memory buffer interface. Before Python 3.12, it is not possible to implement the buffer protocol in pure Python code, or even to check whether a class implements the buffer protocol. In Python 3.12 and higher, the ``__buffer__`` method allows access to the buffer protocol from Python code, and the ``collections.abc.Buffer`` ABC allows checking whether a class implements the buffer protocol. To indicate support for the buffer protocol in earlier versions, inherit from this ABC, either in a stub file or at runtime, or use ABC registration. This ABC provides no methods, because there is no Python-accessible methods shared by pre-3.12 buffer classes. It is useful primarily for static checks. """ # As a courtesy, register the most common stdlib buffer classes. Buffer.register(memoryview) Buffer.register(bytearray) Buffer.register(bytes) # Backport of types.get_original_bases, available on 3.12+ in CPython if hasattr(_types, "get_original_bases"): get_original_bases = _types.get_original_bases else: def get_original_bases(cls, /): """Return the class's "original" bases prior to modification by `__mro_entries__`. Examples:: from typing import TypeVar, Generic from typing_extensions import NamedTuple, TypedDict T = TypeVar("T") class Foo(Generic[T]): ... class Bar(Foo[int], float): ... class Baz(list[str]): ... Eggs = NamedTuple("Eggs", [("a", int), ("b", str)]) Spam = TypedDict("Spam", {"a": int, "b": str}) assert get_original_bases(Bar) == (Foo[int], float) assert get_original_bases(Baz) == (list[str],) assert get_original_bases(Eggs) == (NamedTuple,) assert get_original_bases(Spam) == (TypedDict,) assert get_original_bases(int) == (object,) """ try: return cls.__dict__.get("__orig_bases__", cls.__bases__) except AttributeError: raise TypeError( f'Expected an instance of type, not {type(cls).__name__!r}' ) from None # NewType is a class on Python 3.10+, making it pickleable # The error message for subclassing instances of NewType was improved on 3.11+ if sys.version_info >= (3, 11): NewType = typing.NewType else: class NewType: """NewType creates simple unique types with almost zero runtime overhead. NewType(name, tp) is considered a subtype of tp by static type checkers. At runtime, NewType(name, tp) returns a dummy callable that simply returns its argument. Usage:: UserId = NewType('UserId', int) def name_by_id(user_id: UserId) -> str: ... UserId('user') # Fails type check name_by_id(42) # Fails type check name_by_id(UserId(42)) # OK num = UserId(5) + 1 # type: int """ def __call__(self, obj, /): return obj def __init__(self, name, tp): self.__qualname__ = name if '.' in name: name = name.rpartition('.')[-1] self.__name__ = name self.__supertype__ = tp def_mod = _caller() if def_mod != 'typing_extensions': self.__module__ = def_mod def __mro_entries__(self, bases): # We defined __mro_entries__ to get a better error message # if a user attempts to subclass a NewType instance. bpo-46170 supercls_name = self.__name__ class Dummy: def __init_subclass__(cls): subcls_name = cls.__name__ raise TypeError( f"Cannot subclass an instance of NewType. " f"Perhaps you were looking for: " f"`{subcls_name} = NewType({subcls_name!r}, {supercls_name})`" ) return (Dummy,) def __repr__(self): return f'{self.__module__}.{self.__qualname__}' def __reduce__(self): return self.__qualname__ if sys.version_info >= (3, 10): # PEP 604 methods # It doesn't make sense to have these methods on Python <3.10 def __or__(self, other): return typing.Union[self, other] def __ror__(self, other): return typing.Union[other, self] if sys.version_info >= (3, 14): TypeAliasType = typing.TypeAliasType # 3.8-3.13 else: if sys.version_info >= (3, 12): # 3.12-3.14 def _is_unionable(obj): """Corresponds to is_unionable() in unionobject.c in CPython.""" return obj is None or isinstance(obj, ( type, _types.GenericAlias, _types.UnionType, typing.TypeAliasType, TypeAliasType, )) else: # 3.8-3.11 def _is_unionable(obj): """Corresponds to is_unionable() in unionobject.c in CPython.""" return obj is None or isinstance(obj, ( type, _types.GenericAlias, _types.UnionType, TypeAliasType, )) if sys.version_info < (3, 10): # Copied and pasted from https://github.com/python/cpython/blob/986a4e1b6fcae7fe7a1d0a26aea446107dd58dd2/Objects/genericaliasobject.c#L568-L582, # so that we emulate the behaviour of `types.GenericAlias` # on the latest versions of CPython _ATTRIBUTE_DELEGATION_EXCLUSIONS = frozenset({ "__class__", "__bases__", "__origin__", "__args__", "__unpacked__", "__parameters__", "__typing_unpacked_tuple_args__", "__mro_entries__", "__reduce_ex__", "__reduce__", "__copy__", "__deepcopy__", }) class _TypeAliasGenericAlias(typing._GenericAlias, _root=True): def __getattr__(self, attr): if attr in _ATTRIBUTE_DELEGATION_EXCLUSIONS: return object.__getattr__(self, attr) return getattr(self.__origin__, attr) if sys.version_info < (3, 9): def __getitem__(self, item): result = super().__getitem__(item) result.__class__ = type(self) return result class TypeAliasType: """Create named, parameterized type aliases. This provides a backport of the new `type` statement in Python 3.12: type ListOrSet[T] = list[T] \| set[T] is equivalent to: T = TypeVar("T") ListOrSet = TypeAliasType("ListOrSet", list[T] \| set[T], type_params=(T,)) The name ListOrSet can then be used as an alias for the type it refers to. The type_params argument should contain all the type parameters used in the value of the type alias. If the alias is not generic, this argument is omitted. Static type checkers should only support type aliases declared using TypeAliasType that follow these rules: - The first argument (the name) must be a string literal. - The TypeAliasType instance must be immediately assigned to a variable of the same name. (For example, 'X = TypeAliasType("Y", int)' is invalid, as is 'X, Y = TypeAliasType("X", int), TypeAliasType("Y", int)'). """ def __init__(self, name: str, value, , type_params=()): if not isinstance(name, str): raise TypeError("TypeAliasType name must be a string") if not isinstance(type_params, tuple): raise TypeError("type_params must be a tuple") self.__value__ = value self.__type_params__ = type_params default_value_encountered = False parameters = [] for type_param in type_params: if ( not isinstance(type_param, (TypeVar, TypeVarTuple, ParamSpec)) # 3.8-3.11 # Unpack Backport passes isinstance(type_param, TypeVar) or _is_unpack(type_param) ): raise TypeError(f"Expected a type param, got {type_param!r}") has_default = ( getattr(type_param, '__default__', NoDefault) is not NoDefault ) if default_value_encountered and not has_default: raise TypeError(f"non-default type parameter '{type_param!r}'" " follows default type parameter") if has_default: default_value_encountered = True if isinstance(type_param, TypeVarTuple): parameters.extend(type_param) else: parameters.append(type_param) self.__parameters__ = tuple(parameters) def_mod = _caller() if def_mod != 'typing_extensions': self.__module__ = def_mod # Setting this attribute closes the TypeAliasType from further modification self.__name__ = name def __setattr__(self, name: str, value: object, /) -> None: if hasattr(self, "__name__"): self._raise_attribute_error(name) super().__setattr__(name, value) def __delattr__(self, name: str, /) -> Never: self._raise_attribute_error(name) def _raise_attribute_error(self, name: str) -> Never: # Match the Python 3.12 error messages exactly if name == "__name__": raise AttributeError("readonly attribute") elif name in {"__value__", "__type_params__", "__parameters__", "__module__"}: raise AttributeError( f"attribute '{name}' of 'typing.TypeAliasType' objects " "is not writable" ) else: raise AttributeError( f"'typing.TypeAliasType' object has no attribute '{name}'" ) def __repr__(self) -> str: return self.__name__ if sys.version_info < (3, 11): def _check_single_param(self, param, recursion=0): # Allow [], [int], [int, str], [int, ...], [int, T] if param is ...: return ... if param is None: return None # Note in <= 3.9 _ConcatenateGenericAlias inherits from list if isinstance(param, list) and recursion == 0: return [self._check_single_param(arg, recursion+1) for arg in param] return typing._type_check( param, f'Subscripting {self.__name__} requires a type.' ) def _check_parameters(self, parameters): if sys.version_info < (3, 11): return tuple( self._check_single_param(item) for item in parameters ) return tuple(typing._type_check( item, f'Subscripting {self.__name__} requires a type.' ) for item in parameters ) def __getitem__(self, parameters): if not self.__type_params__: raise TypeError("Only generic type aliases are subscriptable") if not isinstance(parameters, tuple): parameters = (parameters,) # Using 3.9 here will create problems with Concatenate if sys.version_info >= (3, 10): return _types.GenericAlias(self, parameters) type_vars = _collect_type_vars(parameters) parameters = self._check_parameters(parameters) alias = _TypeAliasGenericAlias(self, parameters) # alias.__parameters__ is not complete if Concatenate is present # as it is converted to a list from which no parameters are extracted. if alias.__parameters__ != type_vars: alias.__parameters__ = type_vars return alias def __reduce__(self): return self.__name__ def __init_subclass__(cls, args, kwargs): raise TypeError( "type 'typing_extensions.TypeAliasType' is not an acceptable base type" ) # The presence of this method convinces typing._type_check # that TypeAliasTypes are types. def __call__(self): raise TypeError("Type alias is not callable") if sys.version_info >= (3, 10): def __or__(self, right): # For forward compatibility with 3.12, reject Unions # that are not accepted by the built-in Union. if not _is_unionable(right): return NotImplemented return typing.Union[self, right] def __ror__(self, left): if not _is_unionable(left): return NotImplemented return typing.Union[left, self] if hasattr(typing, "is_protocol"): is_protocol = typing.is_protocol get_protocol_members = typing.get_protocol_members else: def is_protocol(tp: type, /) -> bool: """Return True if the given type is a Protocol. Example:: >>> from typing_extensions import Protocol, is_protocol >>> class P(Protocol): ... def a(self) -> str: ... ... b: int >>> is_protocol(P) True >>> is_protocol(int) False """ return ( isinstance(tp, type) and getattr(tp, '_is_protocol', False) and tp is not Protocol and tp is not typing.Protocol ) def get_protocol_members(tp: type, /) -> typing.FrozenSet[str]: """Return the set of members defined in a Protocol. Example:: >>> from typing_extensions import Protocol, get_protocol_members >>> class P(Protocol): ... def a(self) -> str: ... ... b: int >>> get_protocol_members(P) frozenset({'a', 'b'}) Raise a TypeError for arguments that are not Protocols. """ if not is_protocol(tp): raise TypeError(f'{tp!r} is not a Protocol') if hasattr(tp, '__protocol_attrs__'): return frozenset(tp.__protocol_attrs__) return frozenset(_get_protocol_attrs(tp)) if hasattr(typing, "Doc"): Doc = typing.Doc else: class Doc: """Define the documentation of a type annotation using ``Annotated``, to be used in class attributes, function and method parameters, return values, and variables. The value should be a positional-only string literal to allow static tools like editors and documentation generators to use it. This complements docstrings. The string value passed is available in the attribute ``documentation``. Example:: >>> from typing_extensions import Annotated, Doc >>> def hi(to: Annotated[str, Doc("Who to say hi to")]) -> None: ... """ def __init__(self, documentation: str, /) -> None: self.documentation = documentation def __repr__(self) -> str: return f"Doc({self.documentation!r})" def __hash__(self) -> int: return hash(self.documentation) def __eq__(self, other: object) -> bool: if not isinstance(other, Doc): return NotImplemented return self.documentation == other.documentation _CapsuleType = getattr(_types, "CapsuleType", None) if _CapsuleType is None: try: import _socket except ImportError: pass else: _CAPI = getattr(_socket, "CAPI", None) if _CAPI is not None: _CapsuleType = type(_CAPI) if _CapsuleType is not None: CapsuleType = _CapsuleType __all__.append("CapsuleType") # Using this convoluted approach so that this keeps working # whether we end up using PEP 649 as written, PEP 749, or # some other variation: in any case, inspect.get_annotations # will continue to exist and will gain a `format` parameter. _PEP_649_OR_749_IMPLEMENTED = ( hasattr(inspect, 'get_annotations') and inspect.get_annotations.__kwdefaults__ is not None and "format" in inspect.get_annotations.__kwdefaults__ ) class Format(enum.IntEnum): VALUE = 1 FORWARDREF = 2 STRING = 3 if _PEP_649_OR_749_IMPLEMENTED: get_annotations = inspect.get_annotations else: def get_annotations(obj, , globals=None, locals=None, eval_str=False, format=Format.VALUE): """Compute the annotations dict for an object. obj may be a callable, class, or module. Passing in an object of any other type raises TypeError. Returns a dict. get_annotations() returns a new dict every time it's called; calling it twice on the same object will return two different but equivalent dicts. This is a backport of `inspect.get_annotations`, which has been in the standard library since Python 3.10. See the standard library documentation for more: https://docs.python.org/3/library/inspect.html#inspect.get_annotations This backport adds the format argument introduced by PEP 649. The three formats supported are: * VALUE: the annotations are returned as-is. This is the default and it is compatible with the behavior on previous Python versions. * FORWARDREF: return annotations as-is if possible, but replace any undefined names with ForwardRef objects. The implementation proposed by PEP 649 relies on language changes that cannot be backported; the typing-extensions implementation simply returns the same result as VALUE. * STRING: return annotations as strings, in a format close to the original source. Again, this behavior cannot be replicated directly in a backport. As an approximation, typing-extensions retrieves the annotations under VALUE semantics and then stringifies them. The purpose of this backport is to allow users who would like to use FORWARDREF or STRING semantics once PEP 649 is implemented, but who also want to support earlier Python versions, to simply write: typing_extensions.get_annotations(obj, format=Format.FORWARDREF) """ format = Format(format) if eval_str and format is not Format.VALUE: raise ValueError("eval_str=True is only supported with format=Format.VALUE") if isinstance(obj, type): # class obj_dict = getattr(obj, '__dict__', None) if obj_dict and hasattr(obj_dict, 'get'): ann = obj_dict.get('__annotations__', None) if isinstance(ann, _types.GetSetDescriptorType): ann = None else: ann = None obj_globals = None module_name = getattr(obj, '__module__', None) if module_name: module = sys.modules.get(module_name, None) if module: obj_globals = getattr(module, '__dict__', None) obj_locals = dict(vars(obj)) unwrap = obj elif isinstance(obj, _types.ModuleType): # module ann = getattr(obj, '__annotations__', None) obj_globals = obj.__dict__ obj_locals = None unwrap = None elif callable(obj): # this includes types.Function, types.BuiltinFunctionType, # types.BuiltinMethodType, functools.partial, functools.singledispatch, # "class funclike" from Lib/test/test_inspect... on and on it goes. ann = getattr(obj, '__annotations__', None) obj_globals = getattr(obj, '__globals__', None) obj_locals = None unwrap = obj elif hasattr(obj, '__annotations__'): ann = obj.__annotations__ obj_globals = obj_locals = unwrap = None else: raise TypeError(f"{obj!r} is not a module, class, or callable.") if ann is None: return {} if not isinstance(ann, dict): raise ValueError(f"{obj!r}.__annotations__ is neither a dict nor None") if not ann: return {} if not eval_str: if format is Format.STRING: return { key: value if isinstance(value, str) else typing._type_repr(value) for key, value in ann.items() } return dict(ann) if unwrap is not None: while True: if hasattr(unwrap, '__wrapped__'): unwrap = unwrap.__wrapped__ continue if isinstance(unwrap, functools.partial): unwrap = unwrap.func continue break if hasattr(unwrap, "__globals__"): obj_globals = unwrap.__globals__ if globals is None: globals = obj_globals if locals is None: locals = obj_locals or {} # "Inject" type parameters into the local namespace # (unless they are shadowed by assignments in the local namespace), # as a way of emulating annotation scopes when calling `eval()` if type_params := getattr(obj, "__type_params__", ()): locals = {param.__name__: param for param in type_params} \| locals return_value = {key: value if not isinstance(value, str) else eval(value, globals, locals) for key, value in ann.items() } return return_value if hasattr(typing, "evaluate_forward_ref"): evaluate_forward_ref = typing.evaluate_forward_ref else: # Implements annotationlib.ForwardRef.evaluate def _eval_with_owner( forward_ref, , owner=None, globals=None, locals=None, type_params=None ): if forward_ref.__forward_evaluated__: return forward_ref.__forward_value__ if getattr(forward_ref, "__cell__", None) is not None: try: value = forward_ref.__cell__.cell_contents except ValueError: pass else: forward_ref.__forward_evaluated__ = True forward_ref.__forward_value__ = value return value if owner is None: owner = getattr(forward_ref, "__owner__", None) if ( globals is None and getattr(forward_ref, "__forward_module__", None) is not None ): globals = getattr( sys.modules.get(forward_ref.__forward_module__, None), "__dict__", None ) if globals is None: globals = getattr(forward_ref, "__globals__", None) if globals is None: if isinstance(owner, type): module_name = getattr(owner, "__module__", None) if module_name: module = sys.modules.get(module_name, None) if module: globals = getattr(module, "__dict__", None) elif isinstance(owner, _types.ModuleType): globals = getattr(owner, "__dict__", None) elif callable(owner): globals = getattr(owner, "__globals__", None) # If we pass None to eval() below, the globals of this module are used. if globals is None: globals = {} if locals is None: locals = {} if isinstance(owner, type): locals.update(vars(owner)) if type_params is None and owner is not None: # "Inject" type parameters into the local namespace # (unless they are shadowed by assignments in* the local namespace), # as a way of emulating annotation scopes when calling `eval()` type_params = getattr(owner, "__type_params__", None) # type parameters require some special handling, # as they exist in their own scope # but `eval()` does not have a dedicated parameter for that scope. # For classes, names in type parameter scopes should override # names in the global scope (which here are called `localns`!), # but should in turn be overridden by names in the class scope # (which here are called `globalns`!) if type_params is not None: globals = dict(globals) locals = dict(locals) for param in type_params: param_name = param.__name__ if ( _FORWARD_REF_HAS_CLASS and not forward_ref.__forward_is_class__ ) or param_name not in globals: globals[param_name] = param locals.pop(param_name, None) arg = forward_ref.__forward_arg__ if arg.isidentifier() and not keyword.iskeyword(arg): if arg in locals: value = locals[arg] elif arg in globals: value = globals[arg] elif hasattr(builtins, arg): return getattr(builtins, arg) else: raise NameError(arg) else: code = forward_ref.__forward_code__ value = eval(code, globals, locals) forward_ref.__forward_evaluated__ = True forward_ref.__forward_value__ = value return value def _lax_type_check( value, msg, is_argument=True, , module=None, allow_special_forms=False ): """ A lax Python 3.11+ like version of typing._type_check """ if hasattr(typing, "_type_convert"): if ( sys.version_info >= (3, 10, 3) or (3, 9, 10) < sys.version_info[:3] < (3, 10) ): # allow_special_forms introduced later cpython/#30926 (bpo-46539) type_ = typing._type_convert( value, module=module, allow_special_forms=allow_special_forms, ) # module was added with bpo-41249 before is_class (bpo-46539) elif "__forward_module__" in typing.ForwardRef.__slots__: type_ = typing._type_convert(value, module=module) else: type_ = typing._type_convert(value) else: if value is None: return type(None) if isinstance(value, str): return ForwardRef(value) type_ = value invalid_generic_forms = (Generic, Protocol) if not allow_special_forms: invalid_generic_forms += (ClassVar,) if is_argument: invalid_generic_forms += (Final,) if ( isinstance(type_, typing._GenericAlias) and get_origin(type_) in invalid_generic_forms ): raise TypeError(f"{type_} is not valid as type argument") from None if type_ in (Any, LiteralString, NoReturn, Never, Self, TypeAlias): return type_ if allow_special_forms and type_ in (ClassVar, Final): return type_ if ( isinstance(type_, (_SpecialForm, typing._SpecialForm)) or type_ in (Generic, Protocol) ): raise TypeError(f"Plain {type_} is not valid as type argument") from None if type(type_) is tuple: # lax version with tuple instead of callable raise TypeError(f"{msg} Got {type_!r:.100}.") return type_ def evaluate_forward_ref( forward_ref, , owner=None, globals=None, locals=None, type_params=None, format=Format.VALUE, _recursive_guard=frozenset(), ): """Evaluate a forward reference as a type hint. This is similar to calling the ForwardRef.evaluate() method, but unlike that method, evaluate_forward_ref() also: * Recursively evaluates forward references nested within the type hint. * Rejects certain objects that are not valid type hints. * Replaces type hints that evaluate to None with types.NoneType. * Supports the FORWARDREF and STRING formats. forward_ref must be an instance of ForwardRef. owner, if given, should be the object that holds the annotations that the forward reference derived from, such as a module, class object, or function. It is used to infer the namespaces to use for looking up names. globals and locals can also be explicitly given to provide the global and local namespaces. type_params is a tuple of type parameters that are in scope when evaluating the forward reference. This parameter must be provided (though it may be an empty tuple) if owner is not given and the forward reference does not already have an owner set. format specifies the format of the annotation and is a member of the annotationlib.Format enum. """ if format == Format.STRING: return forward_ref.__forward_arg__ if forward_ref.__forward_arg__ in _recursive_guard: return forward_ref # Evaluate the forward reference try: value = _eval_with_owner( forward_ref, owner=owner, globals=globals, locals=locals, type_params=type_params, ) except NameError: if format == Format.FORWARDREF: return forward_ref else: raise msg = "Forward references must evaluate to types." if not _FORWARD_REF_HAS_CLASS: allow_special_forms = not forward_ref.__forward_is_argument__ else: allow_special_forms = forward_ref.__forward_is_class__ type_ = _lax_type_check( value, msg, is_argument=forward_ref.__forward_is_argument__, allow_special_forms=allow_special_forms, ) # Recursively evaluate the type if isinstance(type_, ForwardRef): if getattr(type_, "__forward_module__", True) is not None: globals = None return evaluate_forward_ref( type_, globals=globals, locals=locals, type_params=type_params, owner=owner, _recursive_guard=_recursive_guard, format=format ) if sys.version_info < (3, 12, 5) and type_params: # Make use of type_params locals = dict(locals) if locals else {} for tvar in type_params: if tvar.__name__ not in locals: # lets not overwrite something present locals[tvar.__name__] = tvar if sys.version_info < (3, 9): return typing._eval_type( type_, globals, locals, ) if sys.version_info < (3, 12, 5): return typing._eval_type( type_, globals, locals, recursive_guard=_recursive_guard \| {forward_ref.__forward_arg__}, ) if sys.version_info < (3, 14): return typing._eval_type( type_, globals, locals, type_params, recursive_guard=_recursive_guard \| {forward_ref.__forward_arg__}, ) return typing._eval_type( type_, globals, locals, type_params, recursive_guard=_recursive_guard \| {forward_ref.__forward_arg__}, format=format, owner=owner, ) # Aliases for items that have always been in typing. # Explicitly assign these (rather than using `from typing import *` at the top), # so that we get a CI error if one of these is deleted from typing.py # in a future version of Python AbstractSet = typing.AbstractSet AnyStr = typing.AnyStr BinaryIO = typing.BinaryIO Callable = typing.Callable Collection = typing.Collection Container = typing.Container Dict = typing.Dict ForwardRef = typing.ForwardRef FrozenSet = typing.FrozenSet Generic = typing.Generic Hashable = typing.Hashable IO = typing.IO ItemsView = typing.ItemsView Iterable = typing.Iterable Iterator = typing.Iterator KeysView = typing.KeysView List = typing.List Mapping = typing.Mapping MappingView = typing.MappingView Match = typing.Match MutableMapping = typing.MutableMapping MutableSequence = typing.MutableSequence MutableSet = typing.MutableSet Optional = typing.Optional Pattern = typing.Pattern Reversible = typing.Reversible Sequence = typing.Sequence Set = typing.Set Sized = typing.Sized TextIO = typing.TextIO Tuple = typing.Tuple Union = typing.Union ValuesView = typing.ValuesView cast = typing.cast no_type_check = typing.no_type_check no_type_check_decorator = typing.no_type_check_decorator	168.61 KB	2025-05-25 12:26
📁 typish	-	2025-05-25 12:26
📁 typish-1.9.3.dist-info	-	2025-05-25 12:26
📁 urllib3	-	2025-05-25 12:26
📁 urllib3-1.26.20.dist-info	-	2025-05-25 12:26
📁 validators	-	2025-05-25 12:26
📁 validators-0.35.0.dist-info	-	2025-05-25 12:26
📁 zipp	-	2025-05-25 12:26
📁 zipp-3.21.0.dist-info	-	2025-05-25 12:26

Edit File:

Terminal