def _create_odps_account()

in tzrec/datasets/odps_dataset.py [0:0]

• 37 lines of code
• 8 McCabe index (conditional complexity)

def _create_odps_account() -> Tuple[BaseAccount, str]:
    account = None
    sts_token = None
    if "ODPS_CONFIG_FILE_PATH" in os.environ:
        account_id, account_key, odps_endpoint = _parse_odps_config_file(
            os.environ["ODPS_CONFIG_FILE_PATH"]
        )
        account = AliyunAccount(account_id, account_key)
    elif (
        "ALIBABA_CLOUD_CREDENTIALS_URI" in os.environ
        or "ALIBABA_CLOUD_SECURITY_TOKEN" in os.environ
        or "ALIBABA_CLOUD_CREDENTIALS_FILE" in os.environ
        or "ALIBABA_CLOUD_ECS_METADATA" in os.environ
    ):
        credentials_client = CredClient()
        # prevent too much request to credential server after forked
        credential = credentials_client.get_credential()
        account_id = credential.access_key_id
        account_key = credential.access_key_secret
        sts_token = credential.security_token
        account = CredentialProviderAccount(credentials_client)
        try:
            odps_endpoint = os.environ["ODPS_ENDPOINT"]
        except KeyError as err:
            raise RuntimeError(
                "ODPS_ENDPOINT does not exist in environment variables."
            ) from err
    else:
        account_id, account_key, odps_endpoint = _parse_odps_config_file(
            os.path.join(os.getenv("HOME", "/home/admin"), ".odps_config.ini")
        )
        account = AliyunAccount(account_id, account_key)

    # prevent graph-learn parse odps config hang
    os.environ["ACCESS_ID"] = account_id
    os.environ["ACCESS_KEY"] = account_key
    os.environ["END_POINT"] = odps_endpoint
    if sts_token:
        os.environ["STS_TOKEN"] = sts_token

    return account, odps_endpoint