buildtimetrend.generate_write_key() - Code Metrics - Inspection of "rename keenio.keen_io_generate_*_key to keenio.gen..." - buildtimetrend/python-lib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 6bdeb6...d157c8 )

by Dieter

created 2016-02-04 20:09 UTC

buildtimetrend.generate_write_key() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	15
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
from buildtimetrend import logger
import copy
import keen
import requests
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 keen_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 keen_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 keen_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 keen_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 keen_is_writable():
    """Check if login keys for Keen IO API are set, to allow writing."""
    if keen_has_project_id() and keen_has_write_key():
        return True

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


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

buildtimetrend / python-lib

Push — master ( 6bdeb6...d157c8 )

buildtimetrend.generate_write_key() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like