Source code for globus_sdk.services.transfer.response.activation

from __future__ import annotations

import time
import typing as t

from globus_sdk.response import GlobusHTTPResponse




class ActivationRequirementsResponse(GlobusHTTPResponse):
    """
    Response class for Activation Requirements responses.

    All Activation Requirements documents refer to a specific Endpoint, from
    whence they were acquired. References to "the Endpoint" implicitly refer to
    that originating Endpoint, and not to some other Endpoint.

    **External Documentation**

    See
    `Activation Requirements Document\
    <https://docs.globus.org/api/transfer/endpoint_activation/#activation_requirements_document>`_
    in the API documentation for details.
    """

    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
        super().__init__(*args, **kwargs)

        # at initialization time, capture expires_in and convert to an absolute
        # timestamp -- otherwise, the time between receiving the response and
        # querying its status will start to matter
        if self["expires_in"] == -1:
            # expires_in=-1 is the "infinite lifetime" case
            self.expires_at: int | None = None
        else:
            self.expires_at = int(time.time() + self["expires_in"])

    @property
    def supports_auto_activation(self) -> bool:
        r"""
        Check if the document lists Auto-Activation as an available type of
        activation.
        Typically good to use when you need to catch endpoints that require web
        activation before proceeding.

        >>> endpoint_id = "..."
        >>> tc = TransferClient(...)
        >>> reqs_doc = tc.endpoint_get_activation_requirements(endpoint_id)
        >>> if not reqs_doc.supports_auto_activation:
        >>>     # use `from __future__ import print_function` in py2
        >>>     print(("This endpoint requires web activation. "
        >>>            "Please login and activate the endpoint here:\n"
        >>>            "https://app.globus.org/file-manager?origin_id={}")
        >>>           .format(endpoint_id), file=sys.stderr)
        >>>     # py3 calls it `input()` in py2, use `raw_input()`
        >>>     input("Please Hit Enter When You Are Done")
        """
        return t.cast(bool, self["auto_activation_supported"])

    @property
    def supports_web_activation(self) -> bool:
        """
        Check if the document lists known types of activation that can be done
        through the web. If this returns ``False``, it means that the endpoint
        is of a highly unusual type, and you should directly inspect the
        response's ``data`` attribute to see what is required. Sending users to
        the web page for activation is also a fairly safe action to take.
        Note that ``ActivationRequirementsResponse.supports_auto_activation``
        directly implies
        ``ActivationRequirementsResponse.supports_web_activation``, so these
        are *not* exclusive.

        For example,

        >>> tc = TransferClient(...)
        >>> reqs_doc = tc.endpoint_get_activation_requirements(...)
        >>> if not reqs_doc.supports_web_activation:
        >>>     # use `from __future__ import print_function` in py2
        >>>     print("Highly unusual endpoint. " +
        >>>           "Cannot webactivate. Raw doc: " +
        >>>           str(reqs_doc), file=sys.stderr)
        >>>     print("Sending user to web anyway, just in case.",
        >>>           file=sys.stderr)
        >>> ...
        """
        return (
            self.supports_auto_activation
            or self["oauth_server"] is not None
            or any(
                x for x in self["DATA"] if x["type"] in ("myproxy", "delegate_myproxy")
            )
        )



    def active_until(self, time_seconds: int, relative_time: bool = True) -> bool:
        """
        Check if the Endpoint will be active until some time in the future,
        given as an integer number of seconds.
        When ``relative_time=False``, the ``time_seconds`` is interpreted as a
        POSIX timestamp.

        This supports queries using both relative and absolute timestamps to
        better support a wide range of use cases. For example, if I have a task
        that I know will typically take N seconds, and I want an M second
        safety margin:

        >>> num_secs_allowed = N + M
        >>> tc = TransferClient(...)
        >>> reqs_doc = tc.endpoint_get_activation_requirements(...)
        >>> if not reqs_doc.active_until(num_secs_allowed):
        >>>     raise Exception("Endpoint won't be active long enough")
        >>> ...

        or, alternatively, if I know that the endpoint must be active until
        October 18th, 2016 for my tasks to complete:

        >>> oct18_2016 = 1476803436
        >>> tc = TransferClient(...)
        >>> reqs_doc = tc.endpoint_get_activation_requirements(...)
        >>> if not reqs_doc.active_until(oct18_2016, relative_time=False):
        >>>     raise Exception("Endpoint won't be active long enough")
        >>> ...

        :param time_seconds: Number of seconds into the future.
        :param relative_time: Defaults to True. When False, ``time_seconds`` is treated
            as a POSIX timestamp (i.e. seconds since epoch as an integer) instead of
            its ordinary behavior.


        :return: True if the Endpoint will be active until the deadline, False otherwise
        """
        # inactive endpoint
        if not self["activated"]:
            return False
        # infinite activation period
        if self.expires_at is None:
            return True

        if relative_time:
            return (time.time() + time_seconds) < self.expires_at
        else:
            return time_seconds < self.expires_at


    @property
    def always_activated(self) -> bool:
        """
        Returns True if the endpoint activation never expires
        (e.g. shared endpoints, globus connect personal endpoints).
        """
        return t.cast(int, self["expires_in"]) == -1