Storage¶
The Storage class enables access to the UP42 storage. You can list your assets and orders within an UP42 workspace.
Use the storage:
storage = up42.initialize_storage()
Source code in up42/storage.py
class Storage:
"""
The Storage class enables access to the UP42 storage. You can list
your assets and orders within an UP42 workspace.
Use the storage:
```python
storage = up42.initialize_storage()
```
"""
def __init__(self, auth: Auth):
self.auth = auth
self.workspace_id = auth.workspace_id
def __repr__(self):
env = ", env: dev" if self.auth.env == "dev" else ""
return f"Storage(workspace_id: {self.workspace_id}{env})"
def _query_paginated(
self, url: str, limit: Optional[int] = None, size: int = 50
) -> List[dict]:
"""
Helper to fetch list of items in paginated endpoint, e.g. assets, orders.
Args:
url (str): The base url for paginated endpoint.
limit: Return n first elements sorted by date of creation, optional.
size: Default number of results per pagination page. Tradeoff of number
of results per page and API response time to query one page. Default 50.
Returns:
List[dict]: List of all paginated items.
"""
url = url + f"&size={size}"
first_page_response = self.auth._request(request_type="GET", url=url)
num_pages = first_page_response["data"]["totalPages"]
num_elements = first_page_response["data"]["totalElements"]
results_list = first_page_response["data"]["content"]
if limit is None:
# Also covers single page (without limit)
num_pages_to_query = num_pages
elif limit <= size:
return results_list[:limit]
else:
# Also covers single page (with limit)
num_pages_to_query = math.ceil(min(limit, num_elements) / size)
for page in range(1, num_pages_to_query):
response_json = self.auth._request(
request_type="GET", url=url + f"&page={page}"
)
results_list += response_json["data"]["content"]
return results_list[:limit]
def get_assets(
self,
return_json: bool = False,
limit: Optional[int] = None,
sortby: str = "createdAt",
descending: bool = True,
) -> Union[List[Asset], dict]:
"""
Gets all assets in the workspace as Asset objects or json.
Args:
return_json: If set to True, returns json object.
limit: Optional, only return n first assets by sorting criteria and order.
Optimal to select if your workspace contains many assets.
sortby: The sorting criteria, one of "createdAt", "updatedAt", "source", "type",
"name", "size".
descending: The sorting order, True for descending (default), False for ascending.
Returns:
Asset objects in the workspace or alternatively json info of the assets.
"""
allowed_sorting_criteria = [
"createdAt",
"updatedAt",
"source",
"type",
"name",
"size",
]
if sortby not in allowed_sorting_criteria:
raise ValueError(
f"sortby parameter must be one of {allowed_sorting_criteria}!"
)
sort = f"{sortby},{'desc' if descending else 'asc'}"
url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/assets?format=paginated&sort={sort}"
assets_json = self._query_paginated(url=url, limit=limit)
logger.info(f"Got {len(assets_json)} assets for workspace {self.workspace_id}.")
if return_json:
return assets_json # type: ignore
else:
assets = [
Asset(self.auth, asset_id=asset_json["id"], asset_info=asset_json)
for asset_json in assets_json
]
return assets
def get_orders(
self,
return_json: bool = False,
limit: Optional[int] = None,
sortby: str = "createdAt",
descending: bool = True,
) -> Union[List[Order], dict]:
"""
Gets all orders in the workspace as Order objects or json.
Args:
return_json: If set to True, returns json object.
limit: Optional, only return n first assets by sorting criteria and order.
Optimal to select if your workspace contains many assets.
sortby: The sorting criteria, one of "createdAt", "updatedAt", "status", "dataProvider", "type".
descending: The sorting order, True for descending (default), False for ascending.
Returns:
Order objects in the workspace or alternatively json info of the orders.
"""
allowed_sorting_criteria = [
"createdAt",
"updatedAt",
"type",
"status",
"dataProvider",
]
if sortby not in allowed_sorting_criteria:
raise ValueError(
f"sortby parameter must be one of {allowed_sorting_criteria}!"
)
sort = f"{sortby},{'desc' if descending else 'asc'}"
url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/orders?format=paginated&sort={sort}"
orders_json = self._query_paginated(url=url, limit=limit)
logger.info(f"Got {len(orders_json)} orders for workspace {self.workspace_id}.")
if return_json:
return orders_json # type: ignore
else:
orders = [
Order(self.auth, order_id=order_json["id"], order_info=order_json)
for order_json in orders_json
]
return orders
Methods¶
get_assets(self, return_json=False, limit=None, sortby='createdAt', descending=True)
¶
Gets all assets in the workspace as Asset objects or json.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
return_json |
bool |
If set to True, returns json object. |
False |
limit |
Optional[int] |
Optional, only return n first assets by sorting criteria and order. Optimal to select if your workspace contains many assets. |
None |
sortby |
str |
The sorting criteria, one of "createdAt", "updatedAt", "source", "type", "name", "size". |
'createdAt' |
descending |
bool |
The sorting order, True for descending (default), False for ascending. |
True |
Returns:
Type | Description |
---|---|
Union[List[up42.asset.Asset], dict] |
Asset objects in the workspace or alternatively json info of the assets. |
Source code in up42/storage.py
def get_assets(
self,
return_json: bool = False,
limit: Optional[int] = None,
sortby: str = "createdAt",
descending: bool = True,
) -> Union[List[Asset], dict]:
"""
Gets all assets in the workspace as Asset objects or json.
Args:
return_json: If set to True, returns json object.
limit: Optional, only return n first assets by sorting criteria and order.
Optimal to select if your workspace contains many assets.
sortby: The sorting criteria, one of "createdAt", "updatedAt", "source", "type",
"name", "size".
descending: The sorting order, True for descending (default), False for ascending.
Returns:
Asset objects in the workspace or alternatively json info of the assets.
"""
allowed_sorting_criteria = [
"createdAt",
"updatedAt",
"source",
"type",
"name",
"size",
]
if sortby not in allowed_sorting_criteria:
raise ValueError(
f"sortby parameter must be one of {allowed_sorting_criteria}!"
)
sort = f"{sortby},{'desc' if descending else 'asc'}"
url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/assets?format=paginated&sort={sort}"
assets_json = self._query_paginated(url=url, limit=limit)
logger.info(f"Got {len(assets_json)} assets for workspace {self.workspace_id}.")
if return_json:
return assets_json # type: ignore
else:
assets = [
Asset(self.auth, asset_id=asset_json["id"], asset_info=asset_json)
for asset_json in assets_json
]
return assets
get_orders(self, return_json=False, limit=None, sortby='createdAt', descending=True)
¶
Gets all orders in the workspace as Order objects or json.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
return_json |
bool |
If set to True, returns json object. |
False |
limit |
Optional[int] |
Optional, only return n first assets by sorting criteria and order. Optimal to select if your workspace contains many assets. |
None |
sortby |
str |
The sorting criteria, one of "createdAt", "updatedAt", "status", "dataProvider", "type". |
'createdAt' |
descending |
bool |
The sorting order, True for descending (default), False for ascending. |
True |
Returns:
Type | Description |
---|---|
Union[List[up42.order.Order], dict] |
Order objects in the workspace or alternatively json info of the orders. |
Source code in up42/storage.py
def get_orders(
self,
return_json: bool = False,
limit: Optional[int] = None,
sortby: str = "createdAt",
descending: bool = True,
) -> Union[List[Order], dict]:
"""
Gets all orders in the workspace as Order objects or json.
Args:
return_json: If set to True, returns json object.
limit: Optional, only return n first assets by sorting criteria and order.
Optimal to select if your workspace contains many assets.
sortby: The sorting criteria, one of "createdAt", "updatedAt", "status", "dataProvider", "type".
descending: The sorting order, True for descending (default), False for ascending.
Returns:
Order objects in the workspace or alternatively json info of the orders.
"""
allowed_sorting_criteria = [
"createdAt",
"updatedAt",
"type",
"status",
"dataProvider",
]
if sortby not in allowed_sorting_criteria:
raise ValueError(
f"sortby parameter must be one of {allowed_sorting_criteria}!"
)
sort = f"{sortby},{'desc' if descending else 'asc'}"
url = f"{self.auth._endpoint()}/workspaces/{self.workspace_id}/orders?format=paginated&sort={sort}"
orders_json = self._query_paginated(url=url, limit=limit)
logger.info(f"Got {len(orders_json)} orders for workspace {self.workspace_id}.")
if return_json:
return orders_json # type: ignore
else:
orders = [
Order(self.auth, order_id=order_json["id"], order_info=order_json)
for order_json in orders_json
]
return orders