from __future__ import annotations
import time
import typing as t
from globus_sdk.response import GlobusHTTPResponse
[docs]
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")
)
)
[docs]
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