Skip to content

EOSC Client

EOSCClient

A client for interacting with the EOSC Data Transfer API. Handles authentication, request sending, and error parsing.

Source code in eosc_data_transfer_client/client.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
class EOSCClient:
    """
    A client for interacting with the EOSC Data Transfer API.
    Handles authentication, request sending, and error parsing.
    """
    def __init__(self, base_url: str, token: str = None):
        """
        Initializes the EOSCClient.

        Args:
            base_url (str): The base URL of the EOSC API.
            token (str, optional): Bearer token for authorization.
        """
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        if token:
            self.session.headers.update({"Authorization": f"Bearer {token}"})
        self.session.headers.update({"Content-Type": "application/json"})

    def request(self, method, endpoint, **kwargs: Any) -> Union[dict, str]:
        """
        Send a request to the API and handle errors.

        Args:
            method (str): HTTP method (e.g., 'GET', 'POST').
            endpoint (str): API endpoint path.
            **kwargs: Additional request options.

        Returns:
            Union[dict, str]: Parsed response from the API.

        Raises:
            EOSCClientError: For 4xx errors.
            EOSCServerError: For 5xx errors.
            EOSCRequestError: For network issues.
        """
        url = f"{self.base_url}{endpoint}"
        try:
            response = self.session.request(method, url, **kwargs)
            if 400 <= response.status_code < 500:
                try:
                    message = response.json()
                except ValueError:
                    message = response.text
                raise EOSCClientError(response.status_code, message, response=response)

            elif 500 <= response.status_code < 600:
                try:
                    message = response.json()
                except ValueError:
                    message = response.text
                raise EOSCServerError(response.status_code, message, response=response)

            if response.content:
                try:
                    return response.json()
                except ValueError:
                    return response.text
            else:
                return {}

        except requests.exceptions.RequestException as e:
            raise EOSCRequestError(str(e)) from e

__init__(base_url, token=None)

Initializes the EOSCClient.

Parameters:

Name Type Description Default
base_url str

The base URL of the EOSC API.

 required
token str

Bearer token for authorization.

 None
Source code in eosc_data_transfer_client/client.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, base_url: str, token: str = None):
    """
    Initializes the EOSCClient.

    Args:
        base_url (str): The base URL of the EOSC API.
        token (str, optional): Bearer token for authorization.
    """
    self.base_url = base_url.rstrip('/')
    self.session = requests.Session()
    if token:
        self.session.headers.update({"Authorization": f"Bearer {token}"})
    self.session.headers.update({"Content-Type": "application/json"})

request(method, endpoint, **kwargs)

Send a request to the API and handle errors.

Parameters:

Name Type Description Default
method str

HTTP method (e.g., 'GET', 'POST').

 required
endpoint str

API endpoint path.

 required
**kwargs Any

Additional request options.

 {}

Returns:

Type Description
Union[dict, str]

Union[dict, str]: Parsed response from the API.

Raises:

Type Description
EOSCClientError

For 4xx errors.
EOSCServerError

For 5xx errors.
EOSCRequestError

For network issues.
Source code in eosc_data_transfer_client/client.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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
def request(self, method, endpoint, **kwargs: Any) -> Union[dict, str]:
    """
    Send a request to the API and handle errors.

    Args:
        method (str): HTTP method (e.g., 'GET', 'POST').
        endpoint (str): API endpoint path.
        **kwargs: Additional request options.

    Returns:
        Union[dict, str]: Parsed response from the API.

    Raises:
        EOSCClientError: For 4xx errors.
        EOSCServerError: For 5xx errors.
        EOSCRequestError: For network issues.
    """
    url = f"{self.base_url}{endpoint}"
    try:
        response = self.session.request(method, url, **kwargs)
        if 400 <= response.status_code < 500:
            try:
                message = response.json()
            except ValueError:
                message = response.text
            raise EOSCClientError(response.status_code, message, response=response)

        elif 500 <= response.status_code < 600:
            try:
                message = response.json()
            except ValueError:
                message = response.text
            raise EOSCServerError(response.status_code, message, response=response)

        if response.content:
            try:
                return response.json()
            except ValueError:
                return response.text
        else:
            return {}

    except requests.exceptions.RequestException as e:
        raise EOSCRequestError(str(e)) from e