Skip to content

API

csda-client is a library for interacting with CSDA API services.

PRODUCTION_URL module-attribute 

PRODUCTION_URL = 'https://csdap.earthdata.nasa.gov'

The CSDA service url.

CsdaClient

A client for logging into and interacting with CSDA API services.

A CsdaClient can be used as a context manager:

>>> with CsdaClient.open(NetrcAuth()) as client:
...     client.verify()
...
>>> # The underlying http connection has been closed.
Source code in src/csda_client/client.py 
 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
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
class CsdaClient:
    """
    A client for logging into and interacting with CSDA API services.

    A `CsdaClient` can be used as a context manager:

    ```python
    >>> with CsdaClient.open(NetrcAuth()) as client:
    ...     client.verify()
    ...
    >>> # The underlying http connection has been closed.
    ```
    """

    @classmethod
    def open(cls, auth: Auth, url: str = PRODUCTION_URL) -> CsdaClient:
        """Opens and logs in a CSDA client.

        Args:
            auth: A [`httpx.Auth`](https://www.python-httpx.org/advanced/authentication/) that will be used for [Earthdata login](https://urs.earthdata.nasa.gov).
                We recommend either `BasicAuth` or `NetrcAuth`.
            url: The CSDA instance to use for queries.

        Returns:
            A logged-in client.
        """
        client = CsdaClient(url)
        client.login(auth)
        return client

    def __init__(self, url: str = PRODUCTION_URL) -> None:
        """Creates a new, un-logged-in CSDA client.

        Once you've created a client, use [login][csda_client.CsdaClient.login] to get an auth token.

        Args:
            url: The CSDA instance to use for queries.

        Returns:
            An un-logged-in client.
        """
        self.client = Client()
        self.url = url

    def __enter__(self) -> CsdaClient:
        return self

    def __exit__(
        self,
        exc_type: type[Exception] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        self.client.close()

    def verify(self) -> str:
        """Verifies the currently logged in user, returning the response.

        Returns:
            The string response from the authentication verification.
        """
        response = self.request(path="/api/v1/auth/verify", method="GET")
        return response.json()

    def profile(self, username: str) -> Profile:
        """Returns a user's [profile][csda_client.models.Profile].

        Args:
            username: A Earthdata Login username

        Returns:
            That user's CSDA profile
        """
        response = self.request(
            path=f"/signup/api/users/{username}/",
            method="GET",
        )
        return Profile.model_validate(response.json())

    def download_item(self, item: Item, asset_key: str, path: Path) -> None:
        """Downloads a single asset from a STAC item to a local file.

        Args:
            item: A [pystac.Item][]
            asset_key: The item's asset key that you would like to download
            path: The file to which the asset will be saved
        """
        if item.collection_id:
            self.download(item.collection_id, item.id, asset_key, path)
        else:
            raise ValueError("Cannot download an item without a collection id")

    def download(
        self, collection_id: str, item_id: str, asset_key: str, path: Path
    ) -> None:
        """Downloads an asset.

        Args:
            collection_id: The STAC collection id
            item_id: The STAC item id
            asset_key: The asset key
            path: The file to which the asset will be saved
        """
        request_path = f"/api/v2/download/{collection_id}/{item_id}/{asset_key}"
        with self.stream(method="GET", path=request_path) as response:
            with open(path, "wb") as f:
                for chunk in response.iter_bytes(1024 * 8):
                    if chunk:
                        f.write(chunk)

    def vendors(self) -> Iterator[Vendor]:
        """Iterates over all vendors.

        Returns:
            An iterator over all vendors in the CSDA system that you have permissions to see
        """
        response = self.request(method="GET", path="/signup/vendors/api/vendors/")
        for value in response.json():
            yield Vendor.model_validate(value)

    def products(self, vendor_id: int) -> Iterator[Product]:
        """Iterates over all products for a given vendor id.

        Args:
            vendor_id: The id of the vendor

        Returns:
            An iterator over all products that belong to the provided vendor.
        """
        response = self.request(
            method="GET", path=f"/signup/vendors/api/products/?vendor={vendor_id}"
        )
        for value in response.json():
            yield Product.model_validate(value)

    def create_tasking_proposal(
        self, tasking_proposal: CreateTaskingProposal, submit: bool
    ) -> TaskingProposal:
        """Creates a new tasking proposal.

        Args:
            tasking_proposal: The tasking proposal to create
            submit: Whether to submit the tasking proposal, or only save it as a draft

        Returns:
            The created tasking proposal
        """
        path = "/signup/tasking/api/proposals"
        if submit:
            path += "?submit=true"
        response = self.request(
            method="POST",
            path=path,
            json=tasking_proposal.model_dump(mode="json"),
        )
        return TaskingProposal.model_validate(response.json())

    def get_tasking_order_parameters(self, product_id: str) -> OrderParameters:
        """Returns the tasking order parameters for a given STAPI product.

        Args:
            product_id: The STAPI product id

        Returns:
            That product's order parameters
        """
        path = f"/api/v1/stapi/products/{product_id}/order-parameters"
        response = self.request(method="GET", path=path)
        return OrderParameters.model_validate(response.json())

    def create_tasking_request(
        self, product_id: str, order_payload: OrderPayload
    ) -> Order:
        """Creates a new tasking request.

        Args:
            product_id: The STAPI product id
            order_payload: The parameters that will be used to create the order

        Returns:
            The created order
        """
        path = f"/api/v1/stapi/products/{product_id}/orders"
        response = self.request(
            method="POST", path=path, json=order_payload.model_dump(mode="json")
        )
        return Order.model_validate(response.json())

    def _get_url(self, path: str) -> str:
        return urljoin(self.url, path)

    def login(self, auth: Auth) -> None:
        """Log in this client with authentication.

        The retrieved token is saved in the session headers.

        Args:
            auth: The `httpx.Auth` to use for logging in. This is usually either `BasicAuth` or `NetrcAuth`.
        """
        response = self._request_auth(
            "",
            method="GET",
            params={"redirect_uri": self.url},
            follow_redirects=False,
            raise_for_status=False,
        )
        if response.status_code not in (302, 307):
            raise AuthError(
                f"Expected API to respond with a redirect, got {response.status_code}"
            )

        edl_url = response.headers.get("Location", "")
        redirect_path = urlparse(edl_url).path
        if not redirect_path.startswith("/oauth/authorize"):
            raise AuthError(
                f"Expected redirect to /oauth/authorize, got {redirect_path}"
            )

        logger.debug("Authenticating with Earthdata Login...")
        response = self.client.request(url=edl_url, method="GET", auth=auth)
        if response.status_code not in (302, 307):
            raise AuthError(
                "Expected Earthdata Login to respond with a redirect, "
                f"got {response.status_code}: {response.text}"
            )

        querystring = parse_qs(urlparse(response.headers["Location"]).query)
        if (
            querystring.get("error")
            and response.status_code == 302
            and "resolution_url" in response.text
        ):
            start = response.text.find("resolution_url") + len("resolution_url") + 1
            end = response.text.find('"', start)
            raise AuthError(
                "\n".join(
                    [
                        "Authorization required for this application,",
                        "please authorize by visiting the resolution url",
                        response.text[start:end],
                    ]
                )
            )

        if querystring.get("error"):
            raise AuthError(querystring["error_msg"])

        logger.debug("Exchanging authorization code for access token...")
        code = querystring["code"]
        response = self._request_auth("token", method="POST", data={"code": code})
        token = response.json()["access_token"]

        self.client.headers["Authorization"] = f"Bearer {token}"

    def request(
        self,
        method: Method,
        path: str,
        *,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        follow_redirects: bool = False,
        auth: Auth | None = None,
        json: dict[str, Any] | None = None,
        raise_for_status: bool = True,
    ) -> Response:
        """Sends a request and returns its response.

        Args:
            method: The HTTP method to use for the request.
            path: The path to make the request, relative to the url provided on client instantiation
            params: Any URL parameters to add to the request
            data: Any data to include in the request body
            follow_redirects: Whether to follow redirects
            auth: A custom auth to use for the request
            json: Any JSON data to include in the request body
            raise_for_status: Whether to raise an exception if the request is not successful

        Returns:
            The response

        Raises:
            HTTPStatusError: Raise if `raise_for_status` is true and the request is not successful
        """
        response = self.client.request(
            method=method,
            url=self._get_url(path),
            params=params,
            data=data,
            follow_redirects=follow_redirects,
            auth=auth,
            json=json,
        )
        if raise_for_status:
            try:
                response.raise_for_status()
            except HTTPStatusError as e:
                logger.error(f"HTTP status error ({e}): {response.text}")
                raise e
        return response

    @contextmanager
    def stream(self, method: Method, path: str) -> Iterator[Response]:
        """Streams a response.

        This method raises an error on an unsuccessful request.

        Args:
            method: The method to use for the streaming request
            path: The path to stream

        Returns:
            A streaming response
        """
        with self.client.stream(
            method=method, url=self._get_url(path), follow_redirects=True
        ) as response:
            response.raise_for_status()
            yield response

    def _request_auth(
        self,
        path: str,
        method: Method,
        *,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        follow_redirects: bool = False,
        raise_for_status: bool = True,
    ) -> Response:
        """Sends a request to an auth endpoint."""
        return self.request(
            path=f"/api/v1/auth/{path}",
            method=method,
            params=params,
            data=data,
            follow_redirects=follow_redirects,
            raise_for_status=raise_for_status,
        )

open classmethod 

open(auth: Auth, url: str = PRODUCTION_URL) -> CsdaClient

Opens and logs in a CSDA client.

Parameters:

Name Type Description Default
auth Auth

A httpx.Auth that will be used for Earthdata login. We recommend either BasicAuth or NetrcAuth.

 required
url str

The CSDA instance to use for queries.

 PRODUCTION_URL

Returns:

Type Description
CsdaClient

A logged-in client.
Source code in src/csda_client/client.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@classmethod
def open(cls, auth: Auth, url: str = PRODUCTION_URL) -> CsdaClient:
    """Opens and logs in a CSDA client.

    Args:
        auth: A [`httpx.Auth`](https://www.python-httpx.org/advanced/authentication/) that will be used for [Earthdata login](https://urs.earthdata.nasa.gov).
            We recommend either `BasicAuth` or `NetrcAuth`.
        url: The CSDA instance to use for queries.

    Returns:
        A logged-in client.
    """
    client = CsdaClient(url)
    client.login(auth)
    return client

__init__ 

__init__(url: str = PRODUCTION_URL) -> None

Creates a new, un-logged-in CSDA client.

Once you've created a client, use login to get an auth token.

Parameters:

Name Type Description Default
url str

The CSDA instance to use for queries.

 PRODUCTION_URL

Returns:

Type Description
None

An un-logged-in client.
Source code in src/csda_client/client.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(self, url: str = PRODUCTION_URL) -> None:
    """Creates a new, un-logged-in CSDA client.

    Once you've created a client, use [login][csda_client.CsdaClient.login] to get an auth token.

    Args:
        url: The CSDA instance to use for queries.

    Returns:
        An un-logged-in client.
    """
    self.client = Client()
    self.url = url

verify 

verify() -> str

Verifies the currently logged in user, returning the response.

Returns:

Type Description
str

The string response from the authentication verification.
Source code in src/csda_client/client.py 
92
93
94
95
96
97
98
99
def verify(self) -> str:
    """Verifies the currently logged in user, returning the response.

    Returns:
        The string response from the authentication verification.
    """
    response = self.request(path="/api/v1/auth/verify", method="GET")
    return response.json()

profile 

profile(username: str) -> Profile

Returns a user's profile.

Parameters:

Name Type Description Default
username str

A Earthdata Login username

 required

Returns:

Type Description
Profile

That user's CSDA profile
Source code in src/csda_client/client.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def profile(self, username: str) -> Profile:
    """Returns a user's [profile][csda_client.models.Profile].

    Args:
        username: A Earthdata Login username

    Returns:
        That user's CSDA profile
    """
    response = self.request(
        path=f"/signup/api/users/{username}/",
        method="GET",
    )
    return Profile.model_validate(response.json())

download_item 

download_item(
    item: Item, asset_key: str, path: Path
) -> None

Downloads a single asset from a STAC item to a local file.

Parameters:

Name Type Description Default
item Item

A pystac.Item

 required
asset_key str

The item's asset key that you would like to download

 required
path Path

The file to which the asset will be saved

 required
Source code in src/csda_client/client.py 
116
117
118
119
120
121
122
123
124
125
126
127
def download_item(self, item: Item, asset_key: str, path: Path) -> None:
    """Downloads a single asset from a STAC item to a local file.

    Args:
        item: A [pystac.Item][]
        asset_key: The item's asset key that you would like to download
        path: The file to which the asset will be saved
    """
    if item.collection_id:
        self.download(item.collection_id, item.id, asset_key, path)
    else:
        raise ValueError("Cannot download an item without a collection id")

download 

download(
    collection_id: str,
    item_id: str,
    asset_key: str,
    path: Path,
) -> None

Downloads an asset.

Parameters:

Name Type Description Default
collection_id str

The STAC collection id

 required
item_id str

The STAC item id

 required
asset_key str

The asset key

 required
path Path

The file to which the asset will be saved

 required
Source code in src/csda_client/client.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def download(
    self, collection_id: str, item_id: str, asset_key: str, path: Path
) -> None:
    """Downloads an asset.

    Args:
        collection_id: The STAC collection id
        item_id: The STAC item id
        asset_key: The asset key
        path: The file to which the asset will be saved
    """
    request_path = f"/api/v2/download/{collection_id}/{item_id}/{asset_key}"
    with self.stream(method="GET", path=request_path) as response:
        with open(path, "wb") as f:
            for chunk in response.iter_bytes(1024 * 8):
                if chunk:
                    f.write(chunk)

vendors 

vendors() -> Iterator[Vendor]

Iterates over all vendors.

Returns:

Type Description
Iterator[Vendor]

An iterator over all vendors in the CSDA system that you have permissions to see
Source code in src/csda_client/client.py 
147
148
149
150
151
152
153
154
155
def vendors(self) -> Iterator[Vendor]:
    """Iterates over all vendors.

    Returns:
        An iterator over all vendors in the CSDA system that you have permissions to see
    """
    response = self.request(method="GET", path="/signup/vendors/api/vendors/")
    for value in response.json():
        yield Vendor.model_validate(value)

products 

products(vendor_id: int) -> Iterator[Product]

Iterates over all products for a given vendor id.

Parameters:

Name Type Description Default
vendor_id int

The id of the vendor

 required

Returns:

Type Description
Iterator[Product]

An iterator over all products that belong to the provided vendor.
Source code in src/csda_client/client.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def products(self, vendor_id: int) -> Iterator[Product]:
    """Iterates over all products for a given vendor id.

    Args:
        vendor_id: The id of the vendor

    Returns:
        An iterator over all products that belong to the provided vendor.
    """
    response = self.request(
        method="GET", path=f"/signup/vendors/api/products/?vendor={vendor_id}"
    )
    for value in response.json():
        yield Product.model_validate(value)

create_tasking_proposal 

create_tasking_proposal(
    tasking_proposal: CreateTaskingProposal, submit: bool
) -> TaskingProposal

Creates a new tasking proposal.

Parameters:

Name Type Description Default
tasking_proposal CreateTaskingProposal

The tasking proposal to create

 required
submit bool

Whether to submit the tasking proposal, or only save it as a draft

 required

Returns:

Type Description
TaskingProposal

The created tasking proposal
Source code in src/csda_client/client.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def create_tasking_proposal(
    self, tasking_proposal: CreateTaskingProposal, submit: bool
) -> TaskingProposal:
    """Creates a new tasking proposal.

    Args:
        tasking_proposal: The tasking proposal to create
        submit: Whether to submit the tasking proposal, or only save it as a draft

    Returns:
        The created tasking proposal
    """
    path = "/signup/tasking/api/proposals"
    if submit:
        path += "?submit=true"
    response = self.request(
        method="POST",
        path=path,
        json=tasking_proposal.model_dump(mode="json"),
    )
    return TaskingProposal.model_validate(response.json())

get_tasking_order_parameters 

get_tasking_order_parameters(
    product_id: str,
) -> OrderParameters

Returns the tasking order parameters for a given STAPI product.

Parameters:

Name Type Description Default
product_id str

The STAPI product id

 required

Returns:

Type Description
OrderParameters

That product's order parameters
Source code in src/csda_client/client.py 
194
195
196
197
198
199
200
201
202
203
204
205
def get_tasking_order_parameters(self, product_id: str) -> OrderParameters:
    """Returns the tasking order parameters for a given STAPI product.

    Args:
        product_id: The STAPI product id

    Returns:
        That product's order parameters
    """
    path = f"/api/v1/stapi/products/{product_id}/order-parameters"
    response = self.request(method="GET", path=path)
    return OrderParameters.model_validate(response.json())

create_tasking_request 

create_tasking_request(
    product_id: str, order_payload: OrderPayload
) -> Order

Creates a new tasking request.

Parameters:

Name Type Description Default
product_id str

The STAPI product id

 required
order_payload OrderPayload

The parameters that will be used to create the order

 required

Returns:

Type Description
Order

The created order
Source code in src/csda_client/client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def create_tasking_request(
    self, product_id: str, order_payload: OrderPayload
) -> Order:
    """Creates a new tasking request.

    Args:
        product_id: The STAPI product id
        order_payload: The parameters that will be used to create the order

    Returns:
        The created order
    """
    path = f"/api/v1/stapi/products/{product_id}/orders"
    response = self.request(
        method="POST", path=path, json=order_payload.model_dump(mode="json")
    )
    return Order.model_validate(response.json())

login 

login(auth: Auth) -> None

Log in this client with authentication.

The retrieved token is saved in the session headers.

Parameters:

Name Type Description Default
auth Auth

The httpx.Auth to use for logging in. This is usually either BasicAuth or NetrcAuth.

 required
Source code in src/csda_client/client.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def login(self, auth: Auth) -> None:
    """Log in this client with authentication.

    The retrieved token is saved in the session headers.

    Args:
        auth: The `httpx.Auth` to use for logging in. This is usually either `BasicAuth` or `NetrcAuth`.
    """
    response = self._request_auth(
        "",
        method="GET",
        params={"redirect_uri": self.url},
        follow_redirects=False,
        raise_for_status=False,
    )
    if response.status_code not in (302, 307):
        raise AuthError(
            f"Expected API to respond with a redirect, got {response.status_code}"
        )

    edl_url = response.headers.get("Location", "")
    redirect_path = urlparse(edl_url).path
    if not redirect_path.startswith("/oauth/authorize"):
        raise AuthError(
            f"Expected redirect to /oauth/authorize, got {redirect_path}"
        )

    logger.debug("Authenticating with Earthdata Login...")
    response = self.client.request(url=edl_url, method="GET", auth=auth)
    if response.status_code not in (302, 307):
        raise AuthError(
            "Expected Earthdata Login to respond with a redirect, "
            f"got {response.status_code}: {response.text}"
        )

    querystring = parse_qs(urlparse(response.headers["Location"]).query)
    if (
        querystring.get("error")
        and response.status_code == 302
        and "resolution_url" in response.text
    ):
        start = response.text.find("resolution_url") + len("resolution_url") + 1
        end = response.text.find('"', start)
        raise AuthError(
            "\n".join(
                [
                    "Authorization required for this application,",
                    "please authorize by visiting the resolution url",
                    response.text[start:end],
                ]
            )
        )

    if querystring.get("error"):
        raise AuthError(querystring["error_msg"])

    logger.debug("Exchanging authorization code for access token...")
    code = querystring["code"]
    response = self._request_auth("token", method="POST", data={"code": code})
    token = response.json()["access_token"]

    self.client.headers["Authorization"] = f"Bearer {token}"

request 

request(
    method: Method,
    path: str,
    *,
    params: dict[str, Any] | None = None,
    data: dict[str, Any] | None = None,
    follow_redirects: bool = False,
    auth: Auth | None = None,
    json: dict[str, Any] | None = None,
    raise_for_status: bool = True,
) -> Response

Sends a request and returns its response.

Parameters:

Name Type Description Default
method Method

The HTTP method to use for the request.

 required
path str

The path to make the request, relative to the url provided on client instantiation

 required
params dict[str, Any] | None

Any URL parameters to add to the request

 None
data dict[str, Any] | None

Any data to include in the request body

 None
follow_redirects bool

Whether to follow redirects

 False
auth Auth | None

A custom auth to use for the request

 None
json dict[str, Any] | None

Any JSON data to include in the request body

 None
raise_for_status bool

Whether to raise an exception if the request is not successful

 True

Returns:

Type Description
Response

The response

Raises:

Type Description
HTTPStatusError

Raise if raise_for_status is true and the request is not successful
Source code in src/csda_client/client.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
def request(
    self,
    method: Method,
    path: str,
    *,
    params: dict[str, Any] | None = None,
    data: dict[str, Any] | None = None,
    follow_redirects: bool = False,
    auth: Auth | None = None,
    json: dict[str, Any] | None = None,
    raise_for_status: bool = True,
) -> Response:
    """Sends a request and returns its response.

    Args:
        method: The HTTP method to use for the request.
        path: The path to make the request, relative to the url provided on client instantiation
        params: Any URL parameters to add to the request
        data: Any data to include in the request body
        follow_redirects: Whether to follow redirects
        auth: A custom auth to use for the request
        json: Any JSON data to include in the request body
        raise_for_status: Whether to raise an exception if the request is not successful

    Returns:
        The response

    Raises:
        HTTPStatusError: Raise if `raise_for_status` is true and the request is not successful
    """
    response = self.client.request(
        method=method,
        url=self._get_url(path),
        params=params,
        data=data,
        follow_redirects=follow_redirects,
        auth=auth,
        json=json,
    )
    if raise_for_status:
        try:
            response.raise_for_status()
        except HTTPStatusError as e:
            logger.error(f"HTTP status error ({e}): {response.text}")
            raise e
    return response

stream 

stream(method: Method, path: str) -> Iterator[Response]

Streams a response.

This method raises an error on an unsuccessful request.

Parameters:

Name Type Description Default
method Method

The method to use for the streaming request

 required
path str

The path to stream

 required

Returns:

Type Description
Iterator[Response]

A streaming response
Source code in src/csda_client/client.py 
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
@contextmanager
def stream(self, method: Method, path: str) -> Iterator[Response]:
    """Streams a response.

    This method raises an error on an unsuccessful request.

    Args:
        method: The method to use for the streaming request
        path: The path to stream

    Returns:
        A streaming response
    """
    with self.client.stream(
        method=method, url=self._get_url(path), follow_redirects=True
    ) as response:
        response.raise_for_status()
        yield response