Class CLISessionOrchestrator

awscli/telemetry.py:219–286

class CLISessionOrchestrator:
    def __init__(self, generator, writer, reader, sweeper):
        self._generator = generator
        self._writer = writer
        self._reader = reader
        self._sweeper = sweeper

        self._sweep_cache()

    @cached_property
    def cache_key(self):
        return self._generator.generate_cache_key(self._host_id, self._tty)

    @cached_property
    def _session_id(self):
        return self._generator.generate_session_id(
            self._host_id, self._tty, self._timestamp
        )

    @cached_property
    def session_id(self):
        if (cached_data := self._reader.read(self.cache_key)) is not None:
            # Cache hit, but session id is expired. Generate new id and update.
            if (
                cached_data.timestamp + _SESSION_LENGTH_SECONDS
                < self._timestamp
            ):
                cached_data.session_id = self._session_id
            # Always update the timestamp to last used.
            cached_data.timestamp = self._timestamp
            self._writer.write(cached_data)
            return cached_data.session_id
        # Cache miss, generate and write new record.
        session_id = self._session_id
        session_data = CLISessionData(
            self.cache_key, session_id, self._timestamp
        )
        self._writer.write(session_data)
        return session_id

    @cached_property
    def _tty(self):
        # os.ttyname is only available on Unix platforms.
        if is_windows:
            return
        try:
            return os.ttyname(sys.stdin.fileno())
        except (OSError, io.UnsupportedOperation):
            # Standard input was redirected to a pseudofile.
            # This can happen when running tests on IDEs or
            # running scripts with redirected input, etc.
            return

    @cached_property
    def _host_id(self):
        return self._reader.read_host_id()

    @cached_property