Skip to content

up42

up42 is the base library module imported to Python. It provides the elementary functionality that is not bound to a specific class of the UP42 structure. From up42 you can also initialize other classes, e.g. for using the catalog, storage etc.

To import the UP42 library:

import up42

Authenticate with UP42 via

up42.authenticate(
    project_id="your-project-ID",
    project_api_key="your-project-API-key"
)

To initialize any lower level functionality use e.g.

catalog = up42.initialize_catalog()
project = up42.initialize_project()

authenticate(cfg_file=None, project_id=None, project_api_key=None, **kwargs)

Authenticate with UP42, either via project_id & project_api_key, or a config json file containing both. Also see the documentation https://sdk.up42.com/authentication/

Parameters:

Name Type Description Default
cfg_file Union[str, pathlib.Path]

A json file containing project_id & project_api_key

None
project_id Optional[str]

The UP42 project id.

None
project_api_key Optional[str]

The UP42 project api key.

None

Examples:

up42.authenticate(
    project_id="your-project-ID",
    project_api_key="your-project-API-key"
)
Source code in up42/main.py
def authenticate(
    cfg_file: Union[str, Path] = None,
    project_id: Optional[str] = None,
    project_api_key: Optional[str] = None,
    **kwargs,
):
    """
    Authenticate with UP42, either via project_id & project_api_key, or a config json file containing both.
    Also see the documentation https://sdk.up42.com/authentication/

    Args:
        cfg_file: A json file containing project_id & project_api_key
        project_id: The UP42 project id.
        project_api_key: The UP42 project api key.

    Examples:
        ```python
        up42.authenticate(
            project_id="your-project-ID",
            project_api_key="your-project-API-key"
        )
        ```
    """
    global _auth
    _auth = Auth(
        cfg_file=cfg_file,
        project_id=project_id,
        project_api_key=project_api_key,
        **kwargs,
    )

initialize_asset(asset_id)

Returns an Asset object (has to exist on UP42).

Parameters:

Name Type Description Default
asset_id str

The UP42 asset_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_asset(asset_id: str) -> "Asset":
    """
    Returns an Asset object (has to exist on UP42).
    Args:
        asset_id: The UP42 asset_id
    """
    asset = Asset(auth=main._auth, asset_id=asset_id)
    logger.info(f"Initialized {asset}")
    return asset

initialize_catalog()

Returns a Catalog object for using the catalog search.

Source code in up42/initialization.py
@_check_auth
def initialize_catalog() -> "Catalog":
    """
    Returns a Catalog object for using the catalog search.
    """
    return Catalog(auth=main._auth)

initialize_job(job_id)

Returns a Job object (has to exist on UP42).

Parameters:

Name Type Description Default
job_id str

The UP42 job_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_job(job_id: str) -> "Job":
    """
    Returns a Job object (has to exist on UP42).
    Args:
        job_id: The UP42 job_id
    """
    job = Job(auth=main._auth, job_id=job_id, project_id=str(main._auth.project_id))
    logger.info(f"Initialized {job}")
    return job

initialize_jobcollection(job_ids)

Returns a JobCollection object (the referenced jobs have to exist on UP42).

Parameters:

Name Type Description Default
job_ids List[str]

List of UP42 job_ids

required
Source code in up42/initialization.py
@_check_auth
def initialize_jobcollection(job_ids: List[str]) -> "JobCollection":
    """
    Returns a JobCollection object (the referenced jobs have to exist on UP42).
    Args:
        job_ids: List of UP42 job_ids
    """
    jobs = [
        Job(auth=main._auth, job_id=job_id, project_id=str(main._auth.project_id))
        for job_id in job_ids
    ]
    jobcollection = JobCollection(
        auth=main._auth, project_id=str(main._auth.project_id), jobs=jobs
    )
    logger.info(f"Initialized {jobcollection}")
    return jobcollection

initialize_jobtask(jobtask_id, job_id)

Returns a JobTask object (has to exist on UP42).

Parameters:

Name Type Description Default
jobtask_id str

The UP42 jobtask_id

required
job_id str

The UP42 job_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_jobtask(jobtask_id: str, job_id: str) -> "JobTask":
    """
    Returns a JobTask object (has to exist on UP42).
    Args:
        jobtask_id: The UP42 jobtask_id
        job_id: The UP42 job_id
    """
    jobtask = JobTask(
        auth=main._auth,
        jobtask_id=jobtask_id,
        job_id=job_id,
        project_id=str(main._auth.project_id),
    )
    logger.info(f"Initialized {jobtask}")
    return jobtask

initialize_order(order_id)

Returns an Order object (has to exist on UP42).

Parameters:

Name Type Description Default
order_id str

The UP42 order_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_order(order_id: str) -> "Order":
    """
    Returns an Order object (has to exist on UP42).
    Args:
        order_id: The UP42 order_id
    """
    order = Order(auth=main._auth, order_id=order_id)
    logger.info(f"Initialized {order}")
    return order

initialize_project()

Returns the correct Project object (has to exist on UP42).

Source code in up42/initialization.py
@_check_auth
def initialize_project() -> "Project":
    """
    Returns the correct Project object (has to exist on UP42).
    """
    project = Project(auth=main._auth, project_id=str(main._auth.project_id))
    logger.info(f"Initialized {project}")
    return project

initialize_storage()

Returns a Storage object to list orders and assets.

Source code in up42/initialization.py
@_check_auth
def initialize_storage() -> "Storage":
    """
    Returns a Storage object to list orders and assets.
    """
    return Storage(auth=main._auth)

initialize_tasking()

Returns a Tasking object for creating satellite tasking orders.

Source code in up42/initialization.py
@_check_auth
def initialize_tasking() -> "Tasking":
    """
    Returns a Tasking object for creating satellite tasking orders.
    """
    return Tasking(auth=main._auth)

initialize_webhook(webhook_id)

Returns a Webhook object (has to exist on UP42).

Parameters:

Name Type Description Default
webhook_id str

The UP42 webhook_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_webhook(webhook_id: str) -> Webhook:
    """
    Returns a Webhook object (has to exist on UP42).
    Args:
        webhook_id: The UP42 webhook_id
    """
    webhook = Webhook(auth=main._auth, webhook_id=webhook_id)
    logger.info(f"Initialized {webhook}")
    return webhook

initialize_workflow(workflow_id)

Returns a Workflow object (has to exist on UP42).

Parameters:

Name Type Description Default
workflow_id str

The UP42 workflow_id

required
Source code in up42/initialization.py
@_check_auth
def initialize_workflow(workflow_id: str) -> "Workflow":
    """
    Returns a Workflow object (has to exist on UP42).
    Args:
        workflow_id: The UP42 workflow_id
    """
    workflow = Workflow(
        auth=main._auth, workflow_id=workflow_id, project_id=str(main._auth.project_id)
    )
    logger.info(f"Initialized {workflow}")
    return workflow

create_webhook(name, url, events, active=False, secret=None)

Registers a new webhook in the system.

Parameters:

Name Type Description Default
name str

Webhook name

required
url str

Unique URL where the webhook will send the message (HTTPS required)

required
events List[str]

List of event types (order status / job task status)

required
active bool

Webhook status.

False
secret Optional[str]

String that acts as signature to the https request sent to the url.

None

Returns:

Type Description

A dict with details of the registered webhook.

Source code in up42/main.py
@_check_auth
def create_webhook(
    name: str,
    url: str,
    events: List[str],
    active: bool = False,
    secret: Optional[str] = None,
):
    """
    Registers a new webhook in the system.

    Args:
        name: Webhook name
        url: Unique URL where the webhook will send the message (HTTPS required)
        events: List of event types (order status / job task status)
        active: Webhook status.
        secret: String that acts as signature to the https request sent to the url.
    Returns:
        A dict with details of the registered webhook.
    """
    webhook = Webhooks(auth=_auth).create_webhook(
        name=name, url=url, events=events, active=active, secret=secret
    )
    return webhook

get_block_coverage(block_id)

Gets the spatial coverage of a data/processing block as url or GeoJson Feature Collection.

Parameters:

Name Type Description Default
block_id str

The block id.

required

Returns:

Type Description
dict

A dict of the spatial coverage for the specific block.

Source code in up42/main.py
@_check_auth
def get_block_coverage(block_id: str) -> dict:
    # pylint: disable=unused-argument
    """
    Gets the spatial coverage of a data/processing block as
    url or GeoJson Feature Collection.

    Args:
        block_id: The block id.

    Returns:
        A dict of the spatial coverage for the specific block.
    """
    url = f"{_auth._endpoint()}/blocks/{block_id}/coverage"
    response_json = _auth._request(request_type="GET", url=url)
    details_json = response_json["data"]
    response_coverage = requests.get(details_json["url"]).json()
    return response_coverage

get_block_details(block_id, as_dataframe=False)

Gets the detailed information about a specific public block from the server, includes all manifest.json and marketplace.json contents. Can not access custom blocks.

Parameters:

Name Type Description Default
block_id str

The block id.

required
as_dataframe bool

Returns a dataframe instead of json (default).

False

Returns:

Type Description
dict

A dict of the block details metadata for the specific block.

Source code in up42/main.py
@_check_auth
def get_block_details(block_id: str, as_dataframe: bool = False) -> dict:
    """
    Gets the detailed information about a specific public block from
    the server, includes all manifest.json and marketplace.json contents.
    Can not access custom blocks.

    Args:
        block_id: The block id.
        as_dataframe: Returns a dataframe instead of json (default).

    Returns:
        A dict of the block details metadata for the specific block.
    """
    url = f"{_auth._endpoint()}/blocks/{block_id}"  # public blocks
    response_json = _auth._request(request_type="GET", url=url)
    details_json = response_json["data"]

    if as_dataframe:
        return pd.DataFrame.from_dict(details_json, orient="index").transpose()
    else:
        return details_json

get_blocks(block_type=None, basic=True, as_dataframe=False)

Gets a list of all public blocks on the marketplace. Can not access custom blocks.

Parameters:

Name Type Description Default
block_type Optional[str]

Optionally filters to "data" or "processing" blocks, default None.

None
basic bool

Optionally returns simple version {block_id : block_name}

True
as_dataframe bool

Returns a dataframe instead of json (default).

False

Returns:

Type Description
Union[List[Dict], dict]

A list of the public blocks and their metadata. Optional a simpler version dict.

Source code in up42/main.py
@_check_auth
def get_blocks(
    block_type: Optional[str] = None,
    basic: bool = True,
    as_dataframe: bool = False,
) -> Union[List[Dict], dict]:
    """
    Gets a list of all public blocks on the marketplace. Can not access custom blocks.

    Args:
        block_type: Optionally filters to "data" or "processing" blocks, default None.
        basic: Optionally returns simple version {block_id : block_name}
        as_dataframe: Returns a dataframe instead of json (default).

    Returns:
        A list of the public blocks and their metadata. Optional a simpler version
        dict.
    """
    try:
        block_type = block_type.lower()  # type: ignore
    except AttributeError:
        pass
    url = f"{_auth._endpoint()}/blocks"
    response_json = _auth._request(request_type="GET", url=url)
    public_blocks_json = response_json["data"]

    if block_type == "data":
        logger.info("Getting only data blocks.")
        blocks_json = [block for block in public_blocks_json if block["type"] == "DATA"]
    elif block_type == "processing":
        logger.info("Getting only processing blocks.")
        blocks_json = [
            block for block in public_blocks_json if block["type"] == "PROCESSING"
        ]
    else:
        blocks_json = public_blocks_json

    if basic:
        logger.info(
            "Getting blocks name and id, use basic=False for all block details."
        )
        blocks_basic = {block["name"]: block["id"] for block in blocks_json}
        if as_dataframe:
            return pd.DataFrame.from_dict(blocks_basic, orient="index")
        else:
            return blocks_basic

    else:
        if as_dataframe:
            return pd.DataFrame(blocks_json)
        else:
            return blocks_json

get_credits_balance()

Display the overall credits available in your account.

Returns:

Type Description
dict

A dict with the balance of credits available in your account.

Source code in up42/main.py
@_check_auth
def get_credits_balance() -> dict:
    """
    Display the overall credits available in your account.

    Returns:
        A dict with the balance of credits available in your account.
    """
    endpoint_url = f"{_auth._endpoint()}/accounts/me/credits/balance"
    response_json = _auth._request(request_type="GET", url=endpoint_url)
    details_json = response_json["data"]
    return details_json

get_credits_history(start_date=None, end_date=None)

Display the overall credits history consumed in your account. The consumption history will be returned for all workspace_ids on your account.

Parameters:

Name Type Description Default
start_date Union[str, datetime.datetime]

The start date for the credit consumption search e.g. 2021-12-01. Default start_date None uses 2000-01-01.

None
end_date Union[str, datetime.datetime]

The end date for the credit consumption search e.g. 2021-12-31. Default end_date None uses current date.

None

Returns:

Type Description
A dict with the information of the credit consumption records for all the users linked by the account_id. (see https

//docs.up42.com/developers/api#operation/getHistory for output description)

Source code in up42/main.py
@_check_auth
def get_credits_history(
    start_date: Optional[Union[str, datetime]] = None,
    end_date: Optional[Union[str, datetime]] = None,
) -> Dict[str, Union[str, int, Dict]]:
    """
    Display the overall credits history consumed in your account.
    The consumption history will be returned for all workspace_ids on your account.

    Args:
        start_date: The start date for the credit consumption search e.g.
            2021-12-01. Default start_date None uses 2000-01-01.
        end_date: The end date for the credit consumption search e.g.
            2021-12-31. Default end_date None uses current date.

    Returns:
        A dict with the information of the credit consumption records for
        all the users linked by the account_id.
        (see https://docs.up42.com/developers/api#operation/getHistory for
        output description)
    """
    if start_date is None:
        start_date = "2000-01-01"
    if end_date is None:
        tomorrow_date = date.today() + timedelta(days=1)
        tomorrow_datetime = datetime(
            year=tomorrow_date.year,
            month=tomorrow_date.month,
            day=tomorrow_date.day,
        )
        end_date = tomorrow_datetime.strftime("%Y-%m-%d")

    [start_formatted_date, end_formatted_date] = format_time_period(
        start_date=start_date, end_date=end_date
    ).split("/")
    search_parameters = dict(
        {
            "from": start_formatted_date,
            "to": end_formatted_date,
            "size": 2000,  # 2000 is the maximum page size for this call
            "page": 0,
        }
    )
    endpoint_url = f"{_auth._endpoint()}/accounts/me/credits/history"
    response_json: dict = _auth._request(
        request_type="GET", url=endpoint_url, querystring=search_parameters
    )
    isLastPage = response_json["data"]["last"]
    credit_history = response_json["data"]["content"].copy()
    result = dict(response_json["data"])
    del result["content"]
    while not isLastPage:
        search_parameters["page"] += 1
        response_json = _auth._request(
            request_type="GET", url=endpoint_url, querystring=search_parameters
        )
        isLastPage = response_json["data"]["last"]
        credit_history.extend(response_json["data"]["content"].copy())
    result["content"] = credit_history
    return result

get_webhook_events()

Gets all available webhook events.

Returns:

Type Description
dict

A dict of the available webhook events.

Source code in up42/main.py
@_check_auth
def get_webhook_events() -> dict:
    """
    Gets all available webhook events.

    Returns:
        A dict of the available webhook events.
    """
    webhook_events = Webhooks(auth=_auth).get_webhook_events()
    return webhook_events

get_webhooks(return_json=False)

Gets all registered webhooks for this workspace.

Parameters:

Name Type Description Default
return_json bool

If true returns the webhooks information as json instead of webhook class objects.

False

Returns:

Type Description
List[up42.webhooks.Webhook]

A list of the registered webhooks for this workspace.

Source code in up42/main.py
@_check_auth
def get_webhooks(return_json: bool = False) -> List[Webhook]:
    """
    Gets all registered webhooks for this workspace.

    Args:
        return_json: If true returns the webhooks information as json instead of webhook class objects.
    Returns:
        A list of the registered webhooks for this workspace.
    """
    webhooks = Webhooks(auth=_auth).get_webhooks(return_json=return_json)
    return webhooks

validate_manifest(path_or_json)

Validates a block manifest json.

The block manifest is required to build a custom block on UP42 and contains the metadata about the block as well as block input and output capabilities. Also see the manifest chapter in the UP42 documentation.

Parameters:

Name Type Description Default
path_or_json Union[str, pathlib.Path, dict]

The input manifest, either a filepath or json string, see example.

required

Returns:

Type Description
dict

A dictionary with the validation results and potential validation errors.

Source code in up42/main.py
@_check_auth
def validate_manifest(path_or_json: Union[str, Path, dict]) -> dict:
    """
    Validates a block manifest json.

    The block manifest is required to build a custom block on UP42 and contains
    the metadata about the block as well as block input and output capabilities.
    Also see the
    [manifest chapter in the UP42 documentation](https://docs.up42.com/reference/block-manifest.html).

    Args:
        path_or_json: The input manifest, either a filepath or json string, see example.

    Returns:
        A dictionary with the validation results and potential validation errors.
    """
    if isinstance(path_or_json, (str, Path)):
        with open(path_or_json) as src:
            manifest_json = json.load(src)
    else:
        manifest_json = path_or_json
    url = f"{_auth._endpoint()}/validate-schema/block"
    response_json = _auth._request(request_type="POST", url=url, data=manifest_json)
    logger.info("The manifest is valid.")
    return response_json["data"]