Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def available_products(self):
"""Get the list of product identifiers you have access to.
:return: List of product ids
:rtype: DotList
Example::
>>> from descarteslabs.client.services import Metadata
>>> products = Metadata().available_products()
>>> products # doctest: +SKIP
['landsat:LC08:PRE:TOAR']
"""
r = self.session.get("/products")
return DotList(r.json())
if country:
params["country"] = country
if region:
params["region"] = region
if placetype:
params["placetype"] = placetype
if limit:
params["n"] = limit
r = self.session.get("/search", params=params, timeout=self.TIMEOUT)
return DotList(r.json())
>>> from descarteslabs.client.services import Places
>>> results = Places().find('morocco')
>>> _ = results[0].pop('bbox')
>>> results
[
{
'id': 85632693,
'name': 'Morocco',
'path': 'continent:africa_country:morocco',
'placetype': 'country',
'slug': 'africa_morocco'
}
]
"""
r = self.session.get("/find/%s" % path, params=kwargs)
return DotList(r.json())
:rtype: DotList
:raises ~descarteslabs.client.exceptions.NotFoundError: Raised if the
task group cannot be found.
"""
rerun = []
task_ids = list(itertools.islice(task_id_iterable, self.RERUN_BATCH_SIZE))
while task_ids:
r = self.session.post(
"/groups/{group_id}/tasks/rerun".format(group_id=group_id),
json={"task_ids": task_ids, "retry_count": retry_count},
)
r.raise_for_status()
rerun += r.json()["tasks"]
task_ids = list(itertools.islice(task_id_iterable, self.RERUN_BATCH_SIZE))
return DotList(rerun)
end_datetime=end_datetime,
cloud_fraction=cloud_fraction,
cloud_fraction_0=cloud_fraction_0,
fill_fraction=fill_fraction,
storage_state=storage_state,
q=q,
fields=fields,
dltile=dltile,
sort_field=sort_field,
sort_order=sort_order,
randomize=randomize,
batch_size=1000 if limit is None else min(limit, 1000),
**kwargs
)
limited_features = itertools.islice(features_iter, limit)
return DotDict(type="FeatureCollection", features=DotList(limited_features))
:return: List of dicts containing at most `limit` products. Empty if no matching
products are found.
:rtype: DotList(DotDict)
"""
params = ["limit", "offset", "bands", "owner", "text"]
args = locals()
kwargs = dict(
kwargs,
**{param: args[param] for param in params if args[param] is not None}
)
check_deprecated_kwargs(kwargs, {"band": "bands"})
r = self.session.post("/products/search", json=kwargs)
return DotList(r.json())
end_datetime=end_datetime,
cloud_fraction=cloud_fraction,
cloud_fraction_0=cloud_fraction_0,
fill_fraction=fill_fraction,
storage_state=storage_state,
q=q,
limit=limit,
fields=[],
dltile=dltile,
sort_field=sort_field,
sort_order=sort_order,
randomize=randomize,
**kwargs
)
return DotList(feature["id"] for feature in result["features"])
:param int limit: Number of results to return.
:param int offset: Index to start at when returning results.
:return: List of dicts containing at most `limit` bands.
:rtype: DotList(DotDict)
"""
params = ["bands", "require_bands", "limit", "offset"]
args = locals()
kwargs = dict(
kwargs,
**{param: args[param] for param in params if args[param] is not None}
)
r = self.session.post("/bands/derived/search", json=kwargs)
return DotList(r.json())
def categories(self):
"""Get a list of categories
"""
r = self.session.get("/categories", timeout=self.TIMEOUT)
return DotList(r.json())