Historical Data Export
Imports¶
Python modules
import logging
import time
import warnings
from typing import UnionThird-party modules
from sgqlc.operation import Operation
from sgqlc.types import Arg, Variable, non_nullnewrelic_sb_sdk module
from newrelic_sb_sdk.client import NewRelicClient
from newrelic_sb_sdk.graphql import nerdgraph
from newrelic_sb_sdk.graphql.enums import HistoricalDataExportStatus
from newrelic_sb_sdk.graphql.objects import (
Account,
HistoricalDataExportCustomerExportResponse,
)
from newrelic_sb_sdk.graphql.scalars import ID, Int, Nrql, String
from newrelic_sb_sdk.utils.download import download_files
from newrelic_sb_sdk.utils.exceptions import NewRelicError
from newrelic_sb_sdk.utils.response import raise_response_errorsLogger¶
logger = logging.getLogger("newrelic_sb_sdk")Constants¶
Quotas and Limits¶
New Relic imposes strict configuration quotas for running massive Historical Data Exports. These values mimic the system limits to prevent accidental API denial configurations.
# Limits imposed by New Relic, check out documentation for more details.
# https://docs.newrelic.com/docs/apis/nerdgraph/examples/nerdgraph-historical-data-export/
MAX_ALLOWED_DATA_POINTS: int = 200000000
MAX_ALLOWED_CONCURRENT_EXPORTS: int = 2Functions¶
create_historical_data_export¶
Initializes a server-side request to generate an asynchronous export file for given metrics.
def create_historical_data_export(
*,
client: NewRelicClient,
account: Account,
nrql_query: Nrql,
) -> HistoricalDataExportCustomerExportResponse:
"""Schedule a historical data export task through NerdGraph.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
nrql_query: The data query describing what to export.
Returns:
The state definition of the export task.
"""
operation = Operation(
nerdgraph.mutation_type,
variables={
"accountId": Arg(non_null(Int)),
"nrql": Arg(Nrql),
},
)
create_export = operation.historical_data_export_create_export(
account_id=Variable("accountId"),
nrql=Variable("nrql"),
)
create_export.id()
create_export.nrql()
create_export.created_at()
create_export.event_count()
create_export.event_types()
create_export.percent_complete()
create_export.results()
create_export.status()
create_export.message()
create_export.user.email()
create_export.user.id()
create_export.user.name()
create_export.account.id()
create_export.account.name()
response = client.execute(
operation.__to_graphql__(),
variables={
"accountId": account.id,
"nrql": nrql_query,
},
)
raise_response_errors(response=response, account=account)
data = operation + response.json()
return data.historical_data_export_create_exportget_all_historical_data_exports¶
Lists all running and recently completed architectural exports across the account.
def get_all_historical_data_exports(
*,
client: NewRelicClient,
account: Account,
) -> list[HistoricalDataExportCustomerExportResponse]:
"""Retrieve recent or running historical data exports.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
Returns:
A collection of historical data export task states.
"""
operation = Operation(
nerdgraph.query_type,
variables={
"accountId": Arg(non_null(Int)),
},
)
exports = operation.actor.account(
id=Variable("accountId"),
).historical_data_export.exports()
exports.id()
exports.nrql()
exports.created_at()
exports.event_count()
exports.event_types()
exports.percent_complete()
exports.results()
exports.status()
exports.message()
exports.available_until()
exports.begin_time()
exports.end_time()
exports.user.id()
exports.user.name()
exports.user.email()
exports.account.id()
exports.account.name()
response = client.execute(
operation.__to_graphql__(),
variables={
"accountId": account.id,
},
)
raise_response_errors(response=response, account=account)
return (operation + response.json()).actor.account.historical_data_export.exportsget_historical_data_export¶
Queries the progression or exact state parameters of a specific instantiated export configuration.
def get_historical_data_export(
*,
client: NewRelicClient,
account: Account,
export_id: ID,
) -> HistoricalDataExportCustomerExportResponse:
"""Fetch the state of a specific data export.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
export_id: Internal ID generated during task creation.
Returns:
The target export state response.
"""
operation = Operation(
nerdgraph.query_type,
variables={
"accountId": Arg(non_null(Int)),
"exportId": Arg(non_null(ID)),
},
)
export = operation.actor.account(
id=Variable("accountId"),
).historical_data_export.export(
id=Variable("exportId"),
)
export.id()
export.nrql()
export.created_at()
export.event_count()
export.event_types()
export.percent_complete()
export.results()
export.status()
export.message()
export.available_until()
export.begin_time()
export.end_time()
export.user.id()
export.user.name()
export.user.email()
export.account.id()
export.account.name()
response = client.execute(
operation.__to_graphql__(),
variables={
"accountId": account.id,
"exportId": export_id,
},
)
raise_response_errors(response=response, account=account)
return (operation + response.json()).actor.account.historical_data_export.exportcancel_historical_data_export¶
Sends a direct signal to the New Relic backend instructing it to safely terminate a currently running background export.
def cancel_historical_data_export(
*,
client: NewRelicClient,
account: Account,
export_id: str,
) -> HistoricalDataExportCustomerExportResponse:
"""Terminate a progressing data export operation.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
export_id: Internal ID generating during task creation.
Returns:
The modified state response of the cancelled task.
"""
operation = Operation(
nerdgraph.mutation_type,
variables={
"accountId": Arg(non_null(Int)),
"exportId": Arg(non_null(String)),
},
)
cancel_export = operation.historical_data_export_cancel_export(
account_id=Variable("accountId"),
id=Variable("exportId"),
)
cancel_export.id()
cancel_export.nrql()
cancel_export.created_at()
cancel_export.event_count()
cancel_export.event_types()
cancel_export.percent_complete()
cancel_export.results()
cancel_export.status()
cancel_export.message()
cancel_export.available_until()
cancel_export.begin_time()
cancel_export.end_time()
cancel_export.user.id()
cancel_export.user.name()
cancel_export.user.email()
cancel_export.account.id()
cancel_export.account.name()
response = client.execute(
operation.__to_graphql__(),
variables={
"accountId": account.id,
"exportId": export_id,
},
)
raise_response_errors(response=response, account=account)
return (operation + response.json()).historical_data_export_cancel_exportcan_execute_historical_data_export¶
Helper function checking limit configurations to make sure the server will gracefully map a new active running state.
def can_execute_historical_data_export(
*, client: NewRelicClient, account: Account
) -> bool:
"""Determine whether another data export process can start.
Checks the environment to verify that the New Relic account has capacity to
begin processing a new data export.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
Returns:
A boolean flag granting permission to initiate a background export task.
"""
historical_data_exports = get_all_historical_data_exports(
client=client, account=account
)
still_running_statuses = [
HistoricalDataExportStatus("WAITING"),
HistoricalDataExportStatus("IN_PROGRESS"),
]
can_execute = (
len(
[
export
for export in historical_data_exports
if export.status in still_running_statuses
]
)
< MAX_ALLOWED_CONCURRENT_EXPORTS
)
return can_execute_perform_historical_data_export¶
The core internal synchronous logic handling polling and waiting intervals to return a unified structured completion condition once the API flags task successful.
def _perform_historical_data_export(
*,
client: NewRelicClient,
account: Account,
nrql_query: Nrql,
max_retry: int | None = None,
max_retries: int = 5,
retry_delay: int = 30,
) -> HistoricalDataExportCustomerExportResponse:
"""Manage export execution and poll for completion.
Polls the task endpoint to ensure the remote system generates and finalizes
the payload.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
nrql_query: The data query describing what to export.
max_retry: Deprecated iteration limiter. Defaults to None.
max_retries: The backoff iteration parameter. Defaults to 5.
retry_delay: The polling delay. Defaults to 30.
Returns:
Unified execution metadata.
"""
# pylint: disable=too-complex,inconsistent-return-statements
if max_retry is not None:
warnings.warn(
"max_retry is deprecated, use max_retries instead",
DeprecationWarning,
stacklevel=2,
)
max_retries = max_retry
logger.debug(
"%d - %s - Getting historical data",
account.id,
account.name,
)
logger.debug(
"%d - %s - Extracting data with query: %s",
account.id,
account.name,
nrql_query,
)
still_running_statuses = [
HistoricalDataExportStatus("WAITING"),
HistoricalDataExportStatus("IN_PROGRESS"),
]
for retry in range(max_retries + 1):
try:
if not can_execute_historical_data_export(client=client, account=account):
logger.debug(
"%d - %s - Too many concurrent historical data exports. "
"Retrying in %d seconds.",
account.id,
account.name,
(retry + 1) * retry_delay,
)
time.sleep((retry + 1) * retry_delay)
continue
logger.debug(
"%d - %s - Creating historical data export with query: %s",
account.id,
account.name,
nrql_query,
)
historical_data_export = create_historical_data_export(
client=client, account=account, nrql_query=nrql_query
)
logger.debug(
"%d - %s - Historical data export created with ID: %s",
account.id,
account.name,
historical_data_export.id,
)
while historical_data_export.status != HistoricalDataExportStatus(
"COMPLETE_SUCCESS"
):
logger.debug(
"%d - %s - Historical data export with ID: %s "
"is still running. Current status: %s "
"(%.2f%%)",
account.id,
account.name,
historical_data_export.id,
historical_data_export.status,
historical_data_export.percent_complete,
)
time.sleep(retry_delay)
historical_data_export = get_historical_data_export(
client=client, account=account, export_id=historical_data_export.id
)
if historical_data_export.status not in still_running_statuses:
break
if historical_data_export.status != HistoricalDataExportStatus(
"COMPLETE_SUCCESS"
):
logger.error(
"%d - %s - Failed historical data export with ID: %s "
"- Status: %s - Message: '%s'",
account.id,
account.name,
historical_data_export.id,
historical_data_export.status,
historical_data_export.message,
)
raise NewRelicError(
f"{account.id} - {account.name} - Failed historical data "
f"export with ID: {historical_data_export.id} - Status: "
f"{historical_data_export.status} - Message: "
f"'{historical_data_export.message}'"
)
return historical_data_export
except Exception as e: # pylint: disable=broad-except
if historical_data_export:
cancel_historical_data_export(
client=client, account=account, export_id=historical_data_export.id
)
if retry == max_retries - 1:
raise NewRelicError(
f"{account.id} - {account.name} - Failed to extract data: {e}"
) from e
time.sleep(retry_delay)perform_historical_data_export¶
A complete high-level orchestration binding function that natively downloads the resultant files and automatically cleans the API records upon completion mapping logic mapping .
In all seriousness, it polls the generation of the nrql_query, runs the threaded
downloader, and saves the payloads seamlessly.
def perform_historical_data_export(
*,
client: NewRelicClient,
account: Account,
nrql_query: Nrql,
base_file_name: str,
):
"""Generate, wait for, and download an export payload.
Args:
client: The New Relic API client instance for authentication.
account: The target New Relic account object.
nrql_query: The database statement describing what to export.
base_file_name: Expected payload prefix strings matched to files.
"""
export = _perform_historical_data_export(
client=client, account=account, nrql_query=nrql_query
)
if not export.results:
logger.error(
"Historical export with ID %s completed successfully but no "
"results were found",
export.id,
)
else:
download_files(
urls=export.results,
base_file_name=base_file_name,
file_extension="json.gz",
)
cancel_historical_data_export(client=client, account=account, export_id=export.id)