buildtimetrend.has_read_key() - Code Metrics - Inspection of "rename keenio.keen_is_*() to keenio.is_*()" - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( c86d1d...684985 )

by Dieter

created 2016-03-26 19:45 UTC

buildtimetrend.has_read_key() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	7
rs	9.4285

# 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 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 keen.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 keen.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
        keen_add_event("build_jobs", {"job": buildjob.to_dict()})

        # store build stages
        if data_detail in ("full", "extended"):
            keen_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
        )
        keen_add_event("build_jobs", {"job": buildjob.to_dict()})
        if data_detail in ("full", "extended"):
            keen_add_events("build_substages", buildjob.stages_to_list())


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

buildtimetrend / python-lib

Push — master ( c86d1d...684985 )

buildtimetrend.has_read_key() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like