documentation
parent
a3104474df
commit
e4a7b89923
|
|
@ -16,7 +16,18 @@ import urllib.parse
|
|||
import time
|
||||
|
||||
|
||||
def get_slo(ENV, DTAPIToken, DTENV) -> pd.DataFrame:
|
||||
def get_slo(ENV: str, DTAPIToken: str, DTENV: str) -> pd.DataFrame:
|
||||
"""
|
||||
Returns SLO data from dynatrace
|
||||
|
||||
Args:
|
||||
ENV (str): Environment (euprod, naprod, cnprod)
|
||||
DTAPIToken (str): Token for respective environment
|
||||
DTENV (str): Full URL for the respective environment
|
||||
|
||||
Returns:
|
||||
pd.DataFrame: Dataframe containing data from dynatrace
|
||||
"""
|
||||
# DTENV = base url
|
||||
# DTAPIToken = sec token
|
||||
dtclient = dynatraceAPI.Dynatrace(DTENV, DTAPIToken)
|
||||
|
|
@ -45,9 +56,8 @@ def build_params(params: typing.Dict) -> str:
|
|||
return query_string
|
||||
|
||||
|
||||
# TODO: remove env parameter
|
||||
def get_data_from_dynatrace(
|
||||
throttling: float | int,
|
||||
throttling_rate: float | int,
|
||||
token: str,
|
||||
env_url: str,
|
||||
params: typing.Dict | str,
|
||||
|
|
@ -67,9 +77,7 @@ def get_data_from_dynatrace(
|
|||
typing.Dict: Returns the response as
|
||||
"""
|
||||
|
||||
# TODO: add nextpage key feature
|
||||
|
||||
time.sleep(throttling)
|
||||
time.sleep(throttling_rate)
|
||||
|
||||
if type(params) is dict:
|
||||
params_string = f"?{build_params(params)}"
|
||||
|
|
@ -179,6 +187,12 @@ def build_dataframe_for_report(report_items: typing.Dict) -> pd.DataFrame:
|
|||
|
||||
|
||||
def write_xlsx(df: pd.DataFrame) -> None:
|
||||
"""
|
||||
Takes in a pandas dataframe and generates writes it into a XLSX file
|
||||
|
||||
Args:
|
||||
df (pd.DataFrame): Dataframe containing the necessary data for the report
|
||||
"""
|
||||
filename = f"CoCo-APM-Report_{datetime.date.today()}.xlsx"
|
||||
writer = pd.ExcelWriter(filename, engine="xlsxwriter")
|
||||
df.to_excel(writer, sheet_name="hosts", index=False)
|
||||
|
|
@ -282,6 +296,16 @@ def develop_load_json():
|
|||
|
||||
|
||||
def check_if_service_already_exists(services: list, entity_id: str) -> bool:
|
||||
"""
|
||||
Requests point to the same service. This leads to double entries but we only need the data once.
|
||||
|
||||
Args:
|
||||
services (list): List with services
|
||||
entity_id (str): Entity Id for lookup
|
||||
|
||||
Returns:
|
||||
bool: Returns True if the service is already present else False.
|
||||
"""
|
||||
result = False
|
||||
for service in services:
|
||||
if service["entityId"] == entity_id:
|
||||
|
|
@ -290,6 +314,12 @@ def check_if_service_already_exists(services: list, entity_id: str) -> bool:
|
|||
|
||||
|
||||
def build_dataframe_data(data: typing.Dict) -> None:
|
||||
"""
|
||||
This function builds the data for the dataframe, which will be used to generate the report. Contains all data but process_groups.
|
||||
|
||||
Args:
|
||||
data (typing.Dict): Takes in the dictionary containing all the raw data from dynatrace.
|
||||
"""
|
||||
df_data = []
|
||||
|
||||
for hub in data:
|
||||
|
|
@ -343,6 +373,9 @@ def build_dataframe_data(data: typing.Dict) -> None:
|
|||
|
||||
@timer
|
||||
def main() -> None:
|
||||
"""
|
||||
Entrypoint.
|
||||
"""
|
||||
throttling_rate: int | float = 0.25 # only tested with 0.5
|
||||
reportItem = {}
|
||||
with open("./environment.yaml") as file:
|
||||
|
|
|
|||
Loading…
Reference in New Issue