Skip to content

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 
 12
 13
 14
 15
 16
 17
 18
 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
 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
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

Functions

get_assets(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[Asset], dict]

Asset objects in the workspace or alternatively json info of the assets.
Source code in up42/storage.py 
 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
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(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[Order], dict]

Order objects in the workspace or alternatively json info of the orders.
Source code in up42/storage.py 
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
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