Skip to content

Http

HTTP connections with various methods.

ApiKeyConnectionMethod

Bases: BareConnectionMethod

API key connection method.

Source code in teledetection/sdk/http.py 
46
47
48
49
50
51
52
53
class ApiKeyConnectionMethod(BareConnectionMethod):
    """API key connection method."""

    api_key: ApiKey

    def get_headers(self):
        """Return the headers."""
        return self.api_key.to_dict()

get_headers()

Return the headers.

Source code in teledetection/sdk/http.py 
51
52
53
def get_headers(self):
    """Return the headers."""
    return self.api_key.to_dict()

BareConnectionMethod

Bases: BaseModel

Bare connection method, no extra headers.

Source code in teledetection/sdk/http.py 
16
17
18
19
20
21
22
23
24
class BareConnectionMethod(BaseModel):
    """Bare connection method, no extra headers."""

    model_config = ConfigDict(arbitrary_types_allowed=True)
    endpoint: str = ENV.tld_signing_endpoint

    def get_headers(self) -> Dict[str, str]:
        """Get the headers."""
        return {}

get_headers()

Get the headers.

Source code in teledetection/sdk/http.py 
22
23
24
def get_headers(self) -> Dict[str, str]:
    """Get the headers."""
    return {}

HTTPSession

HTTP session class.

Source code in teledetection/sdk/http.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class HTTPSession:
    """HTTP session class."""

    def __init__(self):
        """Initialize the HTTP session."""
        self.session = create_session()
        self.timeout = TIMEOUT
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
        }
        self._method = None

    def get_method(self):
        """Get method."""
        log.debug("Get method")
        if not self._method:
            # Lazy instantiation
            self.prepare_connection_method()
        return self._method

    def prepare_connection_method(self):
        """Set the connection method."""
        # Custom server without authentication method
        if ENV.tld_disable_auth:
            self._method = BareConnectionMethod(endpoint=ENV.tld_signing_endpoint)

        # API key method
        elif api_key := ApiKey.grab():
            self._method = ApiKeyConnectionMethod(
                endpoint=ENV.tld_signing_endpoint, api_key=api_key
            )

        # OAuth2 method
        else:
            self._method = OAuth2ConnectionMethod(endpoint=ENV.tld_signing_endpoint)

    def post(self, route: str, params: Dict):
        """Perform a POST request."""
        method = self.get_method()
        url = f"{method.endpoint}{route}"
        headers = {**self.headers, **method.get_headers()}
        log.debug("POST to %s", url)
        response = self.session.post(url, json=params, headers=headers, timeout=TIMEOUT)
        try:
            response.raise_for_status()
        except Exception as e:
            log.error(literal_eval(response.text))
            raise e

        return response

__init__()

Initialize the HTTP session.

Source code in teledetection/sdk/http.py 
59
60
61
62
63
64
65
66
67
def __init__(self):
    """Initialize the HTTP session."""
    self.session = create_session()
    self.timeout = TIMEOUT
    self.headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    self._method = None

get_method()

Get method.

Source code in teledetection/sdk/http.py 
69
70
71
72
73
74
75
def get_method(self):
    """Get method."""
    log.debug("Get method")
    if not self._method:
        # Lazy instantiation
        self.prepare_connection_method()
    return self._method

post(route, params)

Perform a POST request.

Source code in teledetection/sdk/http.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def post(self, route: str, params: Dict):
    """Perform a POST request."""
    method = self.get_method()
    url = f"{method.endpoint}{route}"
    headers = {**self.headers, **method.get_headers()}
    log.debug("POST to %s", url)
    response = self.session.post(url, json=params, headers=headers, timeout=TIMEOUT)
    try:
        response.raise_for_status()
    except Exception as e:
        log.error(literal_eval(response.text))
        raise e

    return response

prepare_connection_method()

Set the connection method.

Source code in teledetection/sdk/http.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def prepare_connection_method(self):
    """Set the connection method."""
    # Custom server without authentication method
    if ENV.tld_disable_auth:
        self._method = BareConnectionMethod(endpoint=ENV.tld_signing_endpoint)

    # API key method
    elif api_key := ApiKey.grab():
        self._method = ApiKeyConnectionMethod(
            endpoint=ENV.tld_signing_endpoint, api_key=api_key
        )

    # OAuth2 method
    else:
        self._method = OAuth2ConnectionMethod(endpoint=ENV.tld_signing_endpoint)

OAuth2ConnectionMethod

Bases: BareConnectionMethod

OAuth2 connection method.

Source code in teledetection/sdk/http.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class OAuth2ConnectionMethod(BareConnectionMethod):
    """OAuth2 connection method."""

    oauth2_session: OAuth2Session = OAuth2Session()

    def get_headers(self):
        """Return the headers."""
        return {"authorization": f"bearer {self.oauth2_session.get_access_token()}"}

    def get_userinfo(self):
        """Override parent method from BareConnectionMethod."""
        openapi_url = retrieve_token_endpoint().replace("/token", "/userinfo")
        return (
            create_session()
            .get(openapi_url, timeout=TIMEOUT, headers=self.get_headers())
            .json()
        )

get_headers()

Return the headers.

Source code in teledetection/sdk/http.py 
32
33
34
def get_headers(self):
    """Return the headers."""
    return {"authorization": f"bearer {self.oauth2_session.get_access_token()}"}

get_userinfo()

Override parent method from BareConnectionMethod.

Source code in teledetection/sdk/http.py 
36
37
38
39
40
41
42
43
def get_userinfo(self):
    """Override parent method from BareConnectionMethod."""
    openapi_url = retrieve_token_endpoint().replace("/token", "/userinfo")
    return (
        create_session()
        .get(openapi_url, timeout=TIMEOUT, headers=self.get_headers())
        .json()
    )

get_headers()

Return the headers needed to authenticate on the system.

Source code in teledetection/sdk/http.py 
112
113
114
def get_headers() -> dict[str, Any]:
    """Return the headers needed to authenticate on the system."""
    return session.get_method().get_headers()

get_userinfo()

Return userinfo.

Source code in teledetection/sdk/http.py 
117
118
119
def get_userinfo() -> dict[str, str]:
    """Return userinfo."""
    return OAuth2ConnectionMethod().get_userinfo()

get_username()

Return username.

Source code in teledetection/sdk/http.py 
122
123
124
125
126
def get_username() -> str:
    """Return username."""
    user_info = get_userinfo()
    assert user_info, "Could not fetch user info"
    return user_info["preferred_username"]