buildtimetrend.get_latest_buildtime() - Code Metrics - Inspection of "add more config specs" - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 889440...e965d2 )

by Dieter

created 2016-05-15 19:56 UTC

buildtimetrend.get_latest_buildtime() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	30
rs	3

# vim: set expandtab sw=4 ts=4:
"""
Interface to Keen IO.

Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>

This file is part of buildtimetrend/python-lib
<https://github.com/buildtimetrend/python-lib/>

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Affero General Public License as published by
the Free Software Foundation, either version 3 of the License, or
any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
"""

from __future__ import division
from builtins import str
import os
import copy
import keen
import math
from datetime import datetime
import requests
from buildtimetrend import logger
from keen import scoped_keys
from buildtimetrend.settings import Settings
from buildtimetrend.tools import check_dict
from buildtimetrend.tools import is_list
from buildtimetrend.tools import is_string
from buildtimetrend.buildjob import BuildJob


TIME_INTERVALS = {
    'week': {'name': 'week', 'timeframe': 'this_7_days', 'max_age': 600},
    'month': {'name': 'month', 'timeframe': 'this_30_days', 'max_age': 600},
    'year': {'name': 'year', 'timeframe': 'this_52_weeks', 'max_age': 1800}
}
KEEN_PROJECT_INFO_NAME = "buildtime_trend"


def has_project_id():
    """Check if Keen.io project ID is set."""
    if "KEEN_PROJECT_ID" in os.environ or keen.project_id is not None:
        return True

    logger.warning("Keen.io Project ID is not set")
    return False


def has_master_key():
    """Check if Keen.io Master API key is set."""
    if "KEEN_MASTER_KEY" in os.environ or keen.master_key is not None:
        return True

    logger.warning("Keen.io Master API Key is not set")
    return False


def has_write_key():
    """Check if Keen.io Write Key is set."""
    if "KEEN_WRITE_KEY" in os.environ or keen.write_key is not None:
        return True

    logger.warning("Keen.io Write Key is not set")
    return False


def has_read_key():
    """Check if Keen.io Read key is set."""
    if "KEEN_READ_KEY" in os.environ or keen.read_key is not None:
        return True

    logger.warning("Keen.io Read Key is not set")
    return False


def is_writable():
    """Check if login keys for Keen IO API are set, to allow writing."""
    if has_project_id() and has_write_key():
        return True

    logger.warning("Keen.io Write Key is not set")
    return False


def is_readable():
    """Check if login keys for Keen IO API are set, to allow reading."""
    if has_project_id() and has_read_key():
        return True

    logger.warning("Keen.io Read Key is not set")
    return False


def generate_read_key(repo):
    """
    Create scoped key for reading only the build-stages related data.

    Param repo : github repository slug (fe. buildtimetrend/python-lib)
    """
    if not has_master_key():
        logger.warning("Keen.io Read Key was not created,"
                       " keen.master_key is not defined.")
        return None

    master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")

    privileges = {
        "allowed_operations": ["read"]
    }

    if repo is not None:
        privileges["filters"] = [get_repo_filter(repo)]

    logger.info("Keen.io Read Key is created for %s", repo)
    return scoped_keys.encrypt(master_key, privileges)


def generate_write_key():
    """Create scoped key for write access to Keen.io database."""
    if not has_master_key():
        logger.warning("Keen.io Write Key was not created,"
                       " keen.master_key is not defined.")
        return None

    master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")

    privileges = {
        "allowed_operations": ["write"]
    }

    logger.info("Keen.io Write Key is created")
    return scoped_keys.encrypt(master_key, privileges)


def send_build_data(buildjob, detail=None):
    """
    Send build data generated by client to keen.io.

    Parameters:
    - buildjob : BuildJob instance
    - detail : Data storage detail level :
               'minimal', 'basic', 'full', 'extended'
    """
    if not isinstance(buildjob, BuildJob):
        raise TypeError("param buildjob should be a BuildJob instance")

    data_detail = Settings().get_value_or_setting("data_detail", detail)

    if is_writable():
        logger.info(
            "Sending client build job data to Keen.io (data detail: %s)",
            data_detail
        )
        # store build job data
        add_event("build_jobs", {"job": buildjob.to_dict()})

        # store build stages
        if data_detail in ("full", "extended"):
            add_events("build_stages", buildjob.stages_to_list())


def send_build_data_service(buildjob, detail=None):
    """
    Send build data generated by service to keen.io.

    Parameters:
    - buildjob : BuildJob instance
    - detail : Data storage detail level :
               'minimal', 'basic', 'full', 'extended'
    """
    if not isinstance(buildjob, BuildJob):
        raise TypeError("param buildjob should be a BuildJob instance")

    data_detail = Settings().get_value_or_setting("data_detail", detail)

    if is_writable():
        logger.info(
            "Sending service build job data to Keen.io (data detail: %s)",
            data_detail
        )
        add_event("build_jobs", {"job": buildjob.to_dict()})
        if data_detail in ("full", "extended"):
            add_events("build_substages", buildjob.stages_to_list())


def add_event(event_collection, payload):
    """
    Wrapper for keen.add_event(), adds project info.

    Param event_collection : collection event data is submitted to
    Param payload : data that is submitted
    """
    # add project info to this event
    payload = add_project_info_dict(payload)

    # submit list of events to Keen.io
    keen.add_event(event_collection, payload)
    logger.info(
        "Sent single event to '%s' collection (Keen.io)",
        event_collection
    )


def add_events(event_collection, payload):
    """
    Wrapper for keen.add_events(), adds project info to each event.

    Param event_collection : collection event data is submitted to
    Param payload : array of events that is submitted
    """
    # add project info to each event
    payload = add_project_info_list(payload)

    # submit list of events to Keen.io
    keen.add_events({event_collection: payload})
    logger.info(
        "Sent multiple events to '%s' collection (Keen.io)",
        event_collection
    )


def add_project_info_dict(payload):
    """
    Add project info to a dictonary.

    Param payload: dictonary payload
    """
    # check if payload is a dictionary, throws an exception if it isn't
    check_dict(payload, "payload")

    payload_as_dict = copy.deepcopy(payload)

    payload_as_dict[KEEN_PROJECT_INFO_NAME] = Settings().get_project_info()

    if "job" in payload:
        # override project_name, set to build_job repo
        if "repo" in payload["job"]:
            payload_as_dict[KEEN_PROJECT_INFO_NAME]["project_name"] = \
                payload["job"]["repo"]

        # override timestamp, set to finished_at timestamp
        if "finished_at" in payload["job"]:
            payload_as_dict["keen"] = {
                "timestamp": payload["job"]["finished_at"]["isotimestamp"]
            }

    return payload_as_dict


def add_project_info_list(payload):
    """
    Add project info to a list of dictionaries.

    Param payload: list of dictionaries
    """
    # check if payload is a list, throws an exception if it isn't
    is_list(payload, "payload")

    payload_as_list = []

    # loop over dicts in payload and add project info to each one
    for event_dict in payload:
        payload_as_list.append(add_project_info_dict(event_dict))

    return payload_as_list


def get_dashboard_keen_config(repo):
    """
    Generate the Keen.io settings for the configuration of the dashboard.

    The dashboard is Javascript powered HTML file that contains the
    graphs generated by Keen.io.
    """
    # initialise config settings
    keen_config = {}

    if not has_project_id() or not has_master_key():
        logger.warning("Keen.io related config settings could not be created,"
                       " keen.project_id and/or keen.master_key"
                       " are not defined.")
        return keen_config

    # set keen project ID
    if keen.project_id is None:
        keen_project_id = os.environ["KEEN_PROJECT_ID"]
    else:
        keen_project_id = keen.project_id
    keen_config['projectId'] = str(keen_project_id)

    # generate read key
    read_key = generate_read_key(repo)
    if read_key is not None:
        # convert bytes to string
        if isinstance(read_key, bytes):
            read_key = read_key.decode('utf-8')

        keen_config['readKey'] = str(read_key)

    # return keen config settings
    return keen_config


def get_avg_buildtime(repo=None, interval=None):

    """
    Query Keen.io database and retrieve average build time.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.average(
            "build_jobs",
            target_property="job.duration",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_avg_buildtime() : " + str(msg))
        return -1


def get_total_build_jobs(repo=None, interval=None):

    """
    Query Keen.io database and retrieve total number of build jobs.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.job",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_total_build_jobs() : " + str(msg))
        return -1


def get_passed_build_jobs(repo=None, interval=None):

    """
    Query Keen.io database and retrieve total number of build jobs that passed.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.job",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[
                get_repo_filter(repo),
                {
                    "property_name": "job.result",
                    "operator": "eq",
                    "property_value": "passed"
                }
            ]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_passed_build_jobs() : " + str(msg))
        return -1


def get_pct_passed_build_jobs(repo=None, interval=None):
    """
    Calculate percentage of passed build jobs.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    total_jobs = get_total_build_jobs(repo, interval)
    passed_jobs = get_passed_build_jobs(repo, interval)

    logger.debug("passed/total build jobs : %d/%d", passed_jobs, total_jobs)

    # calculate percentage if at least one job was executed
    # passed is a valid number (not -1)
    if total_jobs > 0 and passed_jobs >= 0:
        return int(float(passed_jobs) / float(total_jobs) * 100.0)

    return -1


def get_result_color(value=0, ok_thershold=90, warning_thershold=70):
    """
    Get color code that corresponds to the result.

    Parameters:
    - value : value to check
    - ok_thershold : OK threshold value
    - warning_thershold : warning thershold value
    """
    if not(isinstance(value, (int, float)) and
           isinstance(ok_thershold, (int, float)) and
           isinstance(warning_thershold, (int, float))):
        return "lightgrey"

    # check thresholds
    if value >= ok_thershold:
        return "green"
    elif value >= warning_thershold:
        return "yellow"
    else:
        return "red"


def get_total_builds(repo=None, interval=None):

    """
    Query Keen.io database and retrieve total number of builds.

    Parameters :
    - repo : repo name (fe. buildtimetrend/service)
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if repo is None or not is_readable():
        return -1

    interval_data = check_time_interval(interval)

    try:
        return keen.count_unique(
            "build_jobs",
            target_property="job.build",
            timeframe=interval_data['timeframe'],
            max_age=interval_data['max_age'],
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_total_builds() : " + str(msg))
        return -1


def get_latest_buildtime(repo=None):
    """
    Query Keen.io database and retrieve buildtime duration of last build.

    Parameters :
    - repo : repo name (fe. buildtimetrend/python-lib)
    """
    if repo is None or not is_readable():
        return -1

    try:
        result = keen.extraction(
            "build_jobs",
            property_names="job.duration",
            latest=1,
            filters=[get_repo_filter(repo)]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_latest_buildtime() : " + str(msg))
        return -1

    if is_list(result) and len(result) > 0 and \
            check_dict(result[0], None, ['job']) and \
            check_dict(result[0]['job'], None, ['duration']):
        return result[0]['job']['duration']

    return -1


def get_days_since_fail(repo=None):
    """
    Query Keen.io database and retrieve time since last failed buildjob.

    Parameters :
    - repo : repo name (fe. buildtimetrend/python-lib)
    """
    if repo is None or not is_readable():
        return -1

    try:
        failed_timestamp = keen.maximum(
            "build_jobs",
            target_property="job.finished_at.timestamp_seconds",
            filters=[
                get_repo_filter(repo),
                {
                    "operator": "ne",
                    "property_name": "job.result",
                    "property_value": "passed"
                }
            ]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return -1
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_days_since_fail() : " + str(msg))
        return -1

    if isinstance(failed_timestamp, (int, float)) and failed_timestamp > 0:
        dt_failed = datetime.fromtimestamp(failed_timestamp)
        dt_now = datetime.now()
        return math.floor((dt_now - dt_failed).total_seconds() / (3600 * 24))

    return -1


def has_build_id(repo=None, build_id=None):
    """
    Check if build_id exists in Keen.io database.

    Parameters :
    - repo : repo name (fe. buildtimetrend/python-lib)
    - build_id : ID of the build
    """
    if repo is None or build_id is None:
        logger.error("Repo or build_id is not set")
        raise ValueError("Repo or build_id is not set")
    if not is_readable():
        raise SystemError("Keen.io Project ID or API Read Key is not set")

    try:
        count = keen.count(
            "build_jobs",
            filters=[get_repo_filter(repo), {
                "property_name": "job.build",
                "operator": "eq",
                "property_value": str(build_id)}]
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        raise SystemError("Connection to Keen.io API failed")
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.has_build_id : " + str(msg))
        raise SystemError(msg)

    return count > 0


def get_all_projects():
    """Query Keen.io database and retrieve a list of all projects."""
    if not is_readable():
        return []

    try:
        result = keen.select_unique(
            "build_jobs",
            "buildtime_trend.project_name",
            max_age=3600 * 24  # cache for 24 hours
        )
    except requests.ConnectionError:
        logger.error("Connection to Keen.io API failed")
        return []
    except keen.exceptions.KeenApiError as msg:
        logger.error("Error in keenio.get_all_projects() : " + str(msg))
        return []

    if is_list(result):
        return result

    return []


def get_repo_filter(repo=None):
    """
    Return filter for analysis request.

    Parameters
    - repo : repo slug name, fe. buildtimetrend/python-lib
    """
    if repo is None:
        return None

    return {
        "property_name": "{}.project_name".format(KEEN_PROJECT_INFO_NAME),
        "operator": "eq",
        "property_value": str(repo)
    }


def check_time_interval(interval=None):
    """
    Check time interval and returns corresponding parameters.

    Parameters :
    - interval : timeframe, possible values : 'week', 'month', 'year',
                 anything else defaults to 'week'
    """
    if is_string(interval):
        # convert to lowercase
        interval = interval.lower()

        if interval in TIME_INTERVALS:
            return TIME_INTERVALS[interval]

    return TIME_INTERVALS['week']


1		# vim: set expandtab sw=4 ts=4:
2		"""
3		Interface to Keen IO.
4
5		Copyright (C) 2014-2016 Dieter Adriaenssens <[email protected]>
6
7		This file is part of buildtimetrend/python-lib
8		<https://github.com/buildtimetrend/python-lib/>
9
10		This program is free software: you can redistribute it and/or modify
11		it under the terms of the GNU Affero General Public License as published by
12		the Free Software Foundation, either version 3 of the License, or
13		any later version.
14
15		This program is distributed in the hope that it will be useful,
16		but WITHOUT ANY WARRANTY; without even the implied warranty of
17		MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18		GNU Affero General Public License for more details.
19
20		You should have received a copy of the GNU Affero General Public License
21		along with this program. If not, see <http://www.gnu.org/licenses/>.
22		"""
23
24		from __future__ import division
25		from builtins import str
26		import os
27		import copy
28		import keen
29		import math
30		from datetime import datetime
31		import requests
32		from buildtimetrend import logger
33		from keen import scoped_keys
34		from buildtimetrend.settings import Settings
35		from buildtimetrend.tools import check_dict
36		from buildtimetrend.tools import is_list
37		from buildtimetrend.tools import is_string
38		from buildtimetrend.buildjob import BuildJob
39
40
41		TIME_INTERVALS = {
42		'week': {'name': 'week', 'timeframe': 'this_7_days', 'max_age': 600},
43		'month': {'name': 'month', 'timeframe': 'this_30_days', 'max_age': 600},
44		'year': {'name': 'year', 'timeframe': 'this_52_weeks', 'max_age': 1800}
45		}
46		KEEN_PROJECT_INFO_NAME = "buildtime_trend"
47
48
49		def has_project_id():
50		"""Check if Keen.io project ID is set."""
51		if "KEEN_PROJECT_ID" in os.environ or keen.project_id is not None:
52		return True
53
54		logger.warning("Keen.io Project ID is not set")
55		return False
56
57
58		def has_master_key():
59		"""Check if Keen.io Master API key is set."""
60		if "KEEN_MASTER_KEY" in os.environ or keen.master_key is not None:
61		return True
62
63		logger.warning("Keen.io Master API Key is not set")
64		return False
65
66
67		def has_write_key():
68		"""Check if Keen.io Write Key is set."""
69		if "KEEN_WRITE_KEY" in os.environ or keen.write_key is not None:
70		return True
71
72		logger.warning("Keen.io Write Key is not set")
73		return False
74
75
76		def has_read_key():
77		"""Check if Keen.io Read key is set."""
78		if "KEEN_READ_KEY" in os.environ or keen.read_key is not None:
79		return True
80
81		logger.warning("Keen.io Read Key is not set")
82		return False
83
84
85		def is_writable():
86		"""Check if login keys for Keen IO API are set, to allow writing."""
87		if has_project_id() and has_write_key():
88		return True
89
90		logger.warning("Keen.io Write Key is not set")
91		return False
92
93
94		def is_readable():
95		"""Check if login keys for Keen IO API are set, to allow reading."""
96		if has_project_id() and has_read_key():
97		return True
98
99		logger.warning("Keen.io Read Key is not set")
100		return False
101
102
103		def generate_read_key(repo):
104		"""
105		Create scoped key for reading only the build-stages related data.
106
107		Param repo : github repository slug (fe. buildtimetrend/python-lib)
108		"""
109		if not has_master_key():
110		logger.warning("Keen.io Read Key was not created,"
111		" keen.master_key is not defined.")
112		return None
113
114		master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")
115
116		privileges = {
117		"allowed_operations": ["read"]
118		}
119
120		if repo is not None:
121		privileges["filters"] = [get_repo_filter(repo)]
122
123		logger.info("Keen.io Read Key is created for %s", repo)
124		return scoped_keys.encrypt(master_key, privileges)
125
126
127		def generate_write_key():
128		"""Create scoped key for write access to Keen.io database."""
129		if not has_master_key():
130		logger.warning("Keen.io Write Key was not created,"
131		" keen.master_key is not defined.")
132		return None
133
134		master_key = keen.master_key or os.environ.get("KEEN_MASTER_KEY")
135
136		privileges = {
137		"allowed_operations": ["write"]
138		}
139
140		logger.info("Keen.io Write Key is created")
141		return scoped_keys.encrypt(master_key, privileges)
142
143
144		def send_build_data(buildjob, detail=None):
145		"""
146		Send build data generated by client to keen.io.
147
148		Parameters:
149		- buildjob : BuildJob instance
150		- detail : Data storage detail level :
151		'minimal', 'basic', 'full', 'extended'
152		"""
153		if not isinstance(buildjob, BuildJob):
154		raise TypeError("param buildjob should be a BuildJob instance")
155
156		data_detail = Settings().get_value_or_setting("data_detail", detail)
157
158		if is_writable():
159		logger.info(
160		"Sending client build job data to Keen.io (data detail: %s)",
161		data_detail
162		)
163		# store build job data
164		add_event("build_jobs", {"job": buildjob.to_dict()})
165
166		# store build stages
167		if data_detail in ("full", "extended"):
168		add_events("build_stages", buildjob.stages_to_list())
169
170
171		def send_build_data_service(buildjob, detail=None):
172		"""
173		Send build data generated by service to keen.io.
174
175		Parameters:
176		- buildjob : BuildJob instance
177		- detail : Data storage detail level :
178		'minimal', 'basic', 'full', 'extended'
179		"""
180		if not isinstance(buildjob, BuildJob):
181		raise TypeError("param buildjob should be a BuildJob instance")
182
183		data_detail = Settings().get_value_or_setting("data_detail", detail)
184
185		if is_writable():
186		logger.info(
187		"Sending service build job data to Keen.io (data detail: %s)",
188		data_detail
189		)
190		add_event("build_jobs", {"job": buildjob.to_dict()})
191		if data_detail in ("full", "extended"):
192		add_events("build_substages", buildjob.stages_to_list())
193
194
195		def add_event(event_collection, payload):
196		"""
197		Wrapper for keen.add_event(), adds project info.
198
199		Param event_collection : collection event data is submitted to
200		Param payload : data that is submitted
201		"""
202		# add project info to this event
203		payload = add_project_info_dict(payload)
204
205		# submit list of events to Keen.io
206		keen.add_event(event_collection, payload)
207		logger.info(
208		"Sent single event to '%s' collection (Keen.io)",
209		event_collection
210		)
211
212
213		def add_events(event_collection, payload):
214		"""
215		Wrapper for keen.add_events(), adds project info to each event.
216
217		Param event_collection : collection event data is submitted to
218		Param payload : array of events that is submitted
219		"""
220		# add project info to each event
221		payload = add_project_info_list(payload)
222
223		# submit list of events to Keen.io
224		keen.add_events({event_collection: payload})
225		logger.info(
226		"Sent multiple events to '%s' collection (Keen.io)",
227		event_collection
228		)
229
230
231		def add_project_info_dict(payload):
232		"""
233		Add project info to a dictonary.
234
235		Param payload: dictonary payload
236		"""
237		# check if payload is a dictionary, throws an exception if it isn't
238		check_dict(payload, "payload")
239
240		payload_as_dict = copy.deepcopy(payload)
241
242		payload_as_dict[KEEN_PROJECT_INFO_NAME] = Settings().get_project_info()
243
244		if "job" in payload:
245		# override project_name, set to build_job repo
246		if "repo" in payload["job"]:
247		payload_as_dict[KEEN_PROJECT_INFO_NAME]["project_name"] = \
248		payload["job"]["repo"]
249
250		# override timestamp, set to finished_at timestamp
251		if "finished_at" in payload["job"]:
252		payload_as_dict["keen"] = {
253		"timestamp": payload["job"]["finished_at"]["isotimestamp"]
254		}
255
256		return payload_as_dict
257
258
259		def add_project_info_list(payload):
260		"""
261		Add project info to a list of dictionaries.
262
263		Param payload: list of dictionaries
264		"""
265		# check if payload is a list, throws an exception if it isn't
266		is_list(payload, "payload")
267
268		payload_as_list = []
269
270		# loop over dicts in payload and add project info to each one
271		for event_dict in payload:
272		payload_as_list.append(add_project_info_dict(event_dict))
273
274		return payload_as_list
275
276
277		def get_dashboard_keen_config(repo):
278		"""
279		Generate the Keen.io settings for the configuration of the dashboard.
280
281		The dashboard is Javascript powered HTML file that contains the
282		graphs generated by Keen.io.
283		"""
284		# initialise config settings
285		keen_config = {}
286
287		if not has_project_id() or not has_master_key():
288		logger.warning("Keen.io related config settings could not be created,"
289		" keen.project_id and/or keen.master_key"
290		" are not defined.")
291		return keen_config
292
293		# set keen project ID
294		if keen.project_id is None:
295		keen_project_id = os.environ["KEEN_PROJECT_ID"]
296		else:
297		keen_project_id = keen.project_id
298		keen_config['projectId'] = str(keen_project_id)
299
300		# generate read key
301		read_key = generate_read_key(repo)
302		if read_key is not None:
303		# convert bytes to string
304		if isinstance(read_key, bytes):
305		read_key = read_key.decode('utf-8')
306
307		keen_config['readKey'] = str(read_key)
308
309		# return keen config settings
310		return keen_config
311
312
313	View Code Duplication	def get_avg_buildtime(repo=None, interval=None):
		0 ignored issues – show Duplication introduced 2016-05-15 19:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
314		"""
315		Query Keen.io database and retrieve average build time.
316
317		Parameters :
318		- repo : repo name (fe. buildtimetrend/service)
319		- interval : timeframe, possible values : 'week', 'month', 'year',
320		anything else defaults to 'week'
321		"""
322		if repo is None or not is_readable():
323		return -1
324
325		interval_data = check_time_interval(interval)
326
327		try:
328		return keen.average(
329		"build_jobs",
330		target_property="job.duration",
331		timeframe=interval_data['timeframe'],
332		max_age=interval_data['max_age'],
333		filters=[get_repo_filter(repo)]
334		)
335		except requests.ConnectionError:
336		logger.error("Connection to Keen.io API failed")
337		return -1
338		except keen.exceptions.KeenApiError as msg:
339		logger.error("Error in keenio.get_avg_buildtime() : " + str(msg))
340		return -1
341
342
343	View Code Duplication	def get_total_build_jobs(repo=None, interval=None):
		0 ignored issues – show Duplication introduced 2016-05-15 19:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
344		"""
345		Query Keen.io database and retrieve total number of build jobs.
346
347		Parameters :
348		- repo : repo name (fe. buildtimetrend/service)
349		- interval : timeframe, possible values : 'week', 'month', 'year',
350		anything else defaults to 'week'
351		"""
352		if repo is None or not is_readable():
353		return -1
354
355		interval_data = check_time_interval(interval)
356
357		try:
358		return keen.count_unique(
359		"build_jobs",
360		target_property="job.job",
361		timeframe=interval_data['timeframe'],
362		max_age=interval_data['max_age'],
363		filters=[get_repo_filter(repo)]
364		)
365		except requests.ConnectionError:
366		logger.error("Connection to Keen.io API failed")
367		return -1
368		except keen.exceptions.KeenApiError as msg:
369		logger.error("Error in keenio.get_total_build_jobs() : " + str(msg))
370		return -1
371
372
373	View Code Duplication	def get_passed_build_jobs(repo=None, interval=None):
		0 ignored issues – show Duplication introduced 2016-05-15 19:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
374		"""
375		Query Keen.io database and retrieve total number of build jobs that passed.
376
377		Parameters :
378		- repo : repo name (fe. buildtimetrend/service)
379		- interval : timeframe, possible values : 'week', 'month', 'year',
380		anything else defaults to 'week'
381		"""
382		if repo is None or not is_readable():
383		return -1
384
385		interval_data = check_time_interval(interval)
386
387		try:
388		return keen.count_unique(
389		"build_jobs",
390		target_property="job.job",
391		timeframe=interval_data['timeframe'],
392		max_age=interval_data['max_age'],
393		filters=[
394		get_repo_filter(repo),
395		{
396		"property_name": "job.result",
397		"operator": "eq",
398		"property_value": "passed"
399		}
400		]
401		)
402		except requests.ConnectionError:
403		logger.error("Connection to Keen.io API failed")
404		return -1
405		except keen.exceptions.KeenApiError as msg:
406		logger.error("Error in keenio.get_passed_build_jobs() : " + str(msg))
407		return -1
408
409
410		def get_pct_passed_build_jobs(repo=None, interval=None):
411		"""
412		Calculate percentage of passed build jobs.
413
414		Parameters :
415		- repo : repo name (fe. buildtimetrend/service)
416		- interval : timeframe, possible values : 'week', 'month', 'year',
417		anything else defaults to 'week'
418		"""
419		total_jobs = get_total_build_jobs(repo, interval)
420		passed_jobs = get_passed_build_jobs(repo, interval)
421
422		logger.debug("passed/total build jobs : %d/%d", passed_jobs, total_jobs)
423
424		# calculate percentage if at least one job was executed
425		# passed is a valid number (not -1)
426		if total_jobs > 0 and passed_jobs >= 0:
427		return int(float(passed_jobs) / float(total_jobs) * 100.0)
428
429		return -1
430
431
432		def get_result_color(value=0, ok_thershold=90, warning_thershold=70):
433		"""
434		Get color code that corresponds to the result.
435
436		Parameters:
437		- value : value to check
438		- ok_thershold : OK threshold value
439		- warning_thershold : warning thershold value
440		"""
441		if not(isinstance(value, (int, float)) and
442		isinstance(ok_thershold, (int, float)) and
443		isinstance(warning_thershold, (int, float))):
444		return "lightgrey"
445
446		# check thresholds
447		if value >= ok_thershold:
448		return "green"
449		elif value >= warning_thershold:
450		return "yellow"
451		else:
452		return "red"
453
454
455	View Code Duplication	def get_total_builds(repo=None, interval=None):
		0 ignored issues – show Duplication introduced 2016-05-15 19:57 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
456		"""
457		Query Keen.io database and retrieve total number of builds.
458
459		Parameters :
460		- repo : repo name (fe. buildtimetrend/service)
461		- interval : timeframe, possible values : 'week', 'month', 'year',
462		anything else defaults to 'week'
463		"""
464		if repo is None or not is_readable():
465		return -1
466
467		interval_data = check_time_interval(interval)
468
469		try:
470		return keen.count_unique(
471		"build_jobs",
472		target_property="job.build",
473		timeframe=interval_data['timeframe'],
474		max_age=interval_data['max_age'],
475		filters=[get_repo_filter(repo)]
476		)
477		except requests.ConnectionError:
478		logger.error("Connection to Keen.io API failed")
479		return -1
480		except keen.exceptions.KeenApiError as msg:
481		logger.error("Error in keenio.get_total_builds() : " + str(msg))
482		return -1
483
484
485		def get_latest_buildtime(repo=None):
486		"""
487		Query Keen.io database and retrieve buildtime duration of last build.
488
489		Parameters :
490		- repo : repo name (fe. buildtimetrend/python-lib)
491		"""
492		if repo is None or not is_readable():
493		return -1
494
495		try:
496		result = keen.extraction(
497		"build_jobs",
498		property_names="job.duration",
499		latest=1,
500		filters=[get_repo_filter(repo)]
501		)
502		except requests.ConnectionError:
503		logger.error("Connection to Keen.io API failed")
504		return -1
505		except keen.exceptions.KeenApiError as msg:
506		logger.error("Error in keenio.get_latest_buildtime() : " + str(msg))
507		return -1
508
509		if is_list(result) and len(result) > 0 and \
510		check_dict(result[0], None, ['job']) and \
511		check_dict(result[0]['job'], None, ['duration']):
512		return result[0]['job']['duration']
513
514		return -1
515
516
517		def get_days_since_fail(repo=None):
518		"""
519		Query Keen.io database and retrieve time since last failed buildjob.
520
521		Parameters :
522		- repo : repo name (fe. buildtimetrend/python-lib)
523		"""
524		if repo is None or not is_readable():
525		return -1
526
527		try:
528		failed_timestamp = keen.maximum(
529		"build_jobs",
530		target_property="job.finished_at.timestamp_seconds",
531		filters=[
532		get_repo_filter(repo),
533		{
534		"operator": "ne",
535		"property_name": "job.result",
536		"property_value": "passed"
537		}
538		]
539		)
540		except requests.ConnectionError:
541		logger.error("Connection to Keen.io API failed")
542		return -1
543		except keen.exceptions.KeenApiError as msg:
544		logger.error("Error in keenio.get_days_since_fail() : " + str(msg))
545		return -1
546
547		if isinstance(failed_timestamp, (int, float)) and failed_timestamp > 0:
548		dt_failed = datetime.fromtimestamp(failed_timestamp)
549		dt_now = datetime.now()
550		return math.floor((dt_now - dt_failed).total_seconds() / (3600 * 24))
551
552		return -1
553
554
555		def has_build_id(repo=None, build_id=None):
556		"""
557		Check if build_id exists in Keen.io database.
558
559		Parameters :
560		- repo : repo name (fe. buildtimetrend/python-lib)
561		- build_id : ID of the build
562		"""
563		if repo is None or build_id is None:
564		logger.error("Repo or build_id is not set")
565		raise ValueError("Repo or build_id is not set")
566		if not is_readable():
567		raise SystemError("Keen.io Project ID or API Read Key is not set")
568
569		try:
570		count = keen.count(
571		"build_jobs",
572		filters=[get_repo_filter(repo), {
573		"property_name": "job.build",
574		"operator": "eq",
575		"property_value": str(build_id)}]
576		)
577		except requests.ConnectionError:
578		logger.error("Connection to Keen.io API failed")
579		raise SystemError("Connection to Keen.io API failed")
580		except keen.exceptions.KeenApiError as msg:
581		logger.error("Error in keenio.has_build_id : " + str(msg))
582		raise SystemError(msg)
583
584		return count > 0
585
586
587		def get_all_projects():
588		"""Query Keen.io database and retrieve a list of all projects."""
589		if not is_readable():
590		return []
591
592		try:
593		result = keen.select_unique(
594		"build_jobs",
595		"buildtime_trend.project_name",
596		max_age=3600 * 24 # cache for 24 hours
597		)
598		except requests.ConnectionError:
599		logger.error("Connection to Keen.io API failed")
600		return []
601		except keen.exceptions.KeenApiError as msg:
602		logger.error("Error in keenio.get_all_projects() : " + str(msg))
603		return []
604
605		if is_list(result):
606		return result
607
608		return []
609
610
611		def get_repo_filter(repo=None):
612		"""
613		Return filter for analysis request.
614
615		Parameters
616		- repo : repo slug name, fe. buildtimetrend/python-lib
617		"""
618		if repo is None:
619		return None
620
621		return {
622		"property_name": "{}.project_name".format(KEEN_PROJECT_INFO_NAME),
623		"operator": "eq",
624		"property_value": str(repo)
625		}
626
627
628		def check_time_interval(interval=None):
629		"""
630		Check time interval and returns corresponding parameters.
631
632		Parameters :
633		- interval : timeframe, possible values : 'week', 'month', 'year',
634		anything else defaults to 'week'
635		"""
636		if is_string(interval):
637		# convert to lowercase
638		interval = interval.lower()
639
640		if interval in TIME_INTERVALS:
641		return TIME_INTERVALS[interval]
642
643		return TIME_INTERVALS['week']
644

buildtimetrend / python-lib

Push — master ( 889440...e965d2 )

buildtimetrend.get_latest_buildtime() F

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like