3956992538

Build Type

github

Committed by GitHub

Commit Message

Merge bbb276e87 into ecca49d70

Pull Request Pull Request #312: Release/0.13.0

Run Details

511 of 582 new or added lines in 15 files covered. (87.8%)

2 existing lines in 1 file now uncovered.

2391 of 2522 relevant lines covered (94.81%)

11.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

47.83

/snowplow_tracker/snowplow.py

# """
#     snowplow.py

#     Copyright (c) 2013-2022 Snowplow Analytics Ltd. All rights reserved.

#     This program is licensed to you under the Apache License Version 2.0,
#     and you may not use this file except in compliance with the Apache License
#     Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
#     http://www.apache.org/licenses/LICENSE-2.0.

#     Unless required by applicable law or agreed to in writing,
#     software distributed under the Apache License Version 2.0 is distributed on
#     an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
#     express or implied. See the Apache License Version 2.0 for the specific
#     language governing permissions and limitations there under.

#     Authors: Jack Keene, Anuj More, Alex Dean, Fred Blundun, Paul Boocock
#     Copyright: Copyright (c) 2013-2022 Snowplow Analytics Ltd
#     License: Apache License Version 2.0
# """
import logging
from typing import Optional
from snowplow_tracker import (
    Tracker,
    Emitter,
    subject,
    EmitterConfiguration,
    TrackerConfiguration,
)
from snowplow_tracker.typing import Method

# Logging
logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

"""
Snowplow Class
"""


class Snowplow:
    _trackers = {}

    @staticmethod
    def create_tracker(
        namespace: str,
        endpoint: str,
        method: Method = "post",
        app_id: Optional[str] = None,
        subject: Optional[subject.Subject] = None,
        tracker_config: TrackerConfiguration = TrackerConfiguration(),
        emitter_config: EmitterConfiguration = EmitterConfiguration(),
    ) -> Tracker:
        """
        Create a Snowplow tracker with a namespace and collector URL

        :param  namespace:          Name of the tracker
        :type   namespace:          String
        :param  endpoint:           The collector URL
        :type   endpoint:           String
        :param  method:             The HTTP request method. Defaults to post.
        :type   method:             method
        :param  appId:              Application ID
        :type   appId:              String | None
        :param  subject:            Subject to be tracked
        :type   subject:            Subject | None
        :param  tracker_config:     Tracker configuration
        :type   tracker_config:     TrackerConfiguration
        :param  emitter_config:     Emitter configuration
        :type   emitter_config:     EmitterConfiguration
        :rtype                      Tracker
        """
        if endpoint is None:
            raise TypeError("Emitter or Collector URL must be provided")

        emitter = Emitter(
            endpoint,
            method=method,
            batch_size=emitter_config.batch_size,
            on_success=emitter_config.on_success,
            on_failure=emitter_config.on_failure,
            byte_limit=emitter_config.byte_limit,
            request_timeout=emitter_config.request_timeout,
            custom_retry_codes=emitter_config.custom_retry_codes,
        )

        tracker = Tracker(
            emitter,
            namespace=namespace,
            app_id=app_id,
            subject=subject,
            encode_base64=tracker_config.encode_base64,
            json_encoder=tracker_config.json_encoder,
        )

        return Snowplow.add_tracker(tracker)

    @classmethod
    def add_tracker(cls, tracker: Tracker) -> Tracker:
        """
        Add a Snowplow tracker to the Snowplow object

        :param  tracker:  Tracker object to add to Snowplow
        :type   tracker:  Tracker
        :rtype            Tracker
        """
        if not isinstance(tracker, Tracker):
            logger.info("Tracker not provided.")
            return None

        namespace = tracker.get_namespace()

        if namespace in cls._trackers.keys():
            raise TypeError("Tracker with this namespace already exists")

        cls._trackers[namespace] = tracker
        logger.info("Tracker with namespace: '" + namespace + "' added to Snowplow")
        return cls._trackers[namespace]

    @classmethod
    def remove_tracker(cls, tracker: Tracker):
        """
        Remove a Snowplow tracker from the Snowplow object if it exists

        :param  tracker:        Tracker object to remove from Snowplow
        :type   tracker:        Tracker | None
        """
        namespace = tracker.get_namespace()
        cls.remove_tracker_by_namespace(namespace)

    @classmethod
    def remove_tracker_by_namespace(cls, namespace: str):
        """
        Remove a Snowplow tracker from the Snowplow object using it's namespace if it exists

        :param  namespace:      Tracker namespace to remove from Snowplow
        :type   tracker:        String | None
        """
        if not cls._trackers.pop(namespace, False):
            logger.info("Tracker with namespace: '" + namespace + "' does not exist")
            return
        logger.info("Tracker with namespace: '" + namespace + "' removed from Snowplow")

    @classmethod
    def reset(cls):
        """
        Remove all active Snowplow trackers from the Snowplow object
        """
        cls._trackers = {}

    @classmethod
    def get_tracker(cls, namespace: str) -> Tracker:
        """
        Returns a Snowplow tracker from the Snowplow object if it exists
        :param  namespace:              Snowplow tracker namespace
        :type   namespace:              string
        :rtype:                         Tracker
        """
        if namespace in cls._trackers.keys():
            return cls._trackers[namespace]
        return None

1	# """
2	# snowplow.py
3
4	# Copyright (c) 2013-2022 Snowplow Analytics Ltd. All rights reserved.
5
6	# This program is licensed to you under the Apache License Version 2.0,
7	# and you may not use this file except in compliance with the Apache License
8	# Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
9	# http://www.apache.org/licenses/LICENSE-2.0.
10
11	# Unless required by applicable law or agreed to in writing,
12	# software distributed under the Apache License Version 2.0 is distributed on
13	# an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14	# express or implied. See the Apache License Version 2.0 for the specific
15	# language governing permissions and limitations there under.
16
17	# Authors: Jack Keene, Anuj More, Alex Dean, Fred Blundun, Paul Boocock
18	# Copyright: Copyright (c) 2013-2022 Snowplow Analytics Ltd
19	# License: Apache License Version 2.0
20	# """
21	import logging	12✔
22	from typing import Optional	12✔
23	from snowplow_tracker import (	12✔
24	Tracker,
25	Emitter,
26	subject,
27	EmitterConfiguration,
28	TrackerConfiguration,
29	)
30	from snowplow_tracker.typing import Method	12✔
31
32	# Logging
33	logging.basicConfig()	12✔
34	logger = logging.getLogger(__name__)	12✔
35	logger.setLevel(logging.INFO)	12✔
36
37	"""	4✔
38	Snowplow Class
39	"""
40
41
42	class Snowplow:	12✔
43	_trackers = {}	12✔
44
45	@staticmethod	12✔
46	def create_tracker(	12✔
47	namespace: str,
48	endpoint: str,
49	method: Method = "post",
50	app_id: Optional[str] = None,
51	subject: Optional[subject.Subject] = None,
52	tracker_config: TrackerConfiguration = TrackerConfiguration(),
53	emitter_config: EmitterConfiguration = EmitterConfiguration(),
54	) -> Tracker:
55	"""
56	Create a Snowplow tracker with a namespace and collector URL
57
58	:param namespace: Name of the tracker
59	:type namespace: String
60	:param endpoint: The collector URL
61	:type endpoint: String
62	:param method: The HTTP request method. Defaults to post.
63	:type method: method
64	:param appId: Application ID
65	:type appId: String \| None
66	:param subject: Subject to be tracked
67	:type subject: Subject \| None
68	:param tracker_config: Tracker configuration
69	:type tracker_config: TrackerConfiguration
70	:param emitter_config: Emitter configuration
71	:type emitter_config: EmitterConfiguration
72	:rtype Tracker
73	"""
NEW 74	if endpoint is None:	×
NEW 75	raise TypeError("Emitter or Collector URL must be provided")	×
76
NEW 77	emitter = Emitter(	×
78	endpoint,
79	method=method,
80	batch_size=emitter_config.batch_size,
81	on_success=emitter_config.on_success,
82	on_failure=emitter_config.on_failure,
83	byte_limit=emitter_config.byte_limit,
84	request_timeout=emitter_config.request_timeout,
85	custom_retry_codes=emitter_config.custom_retry_codes,
86	)
87
NEW 88	tracker = Tracker(	×
89	emitter,
90	namespace=namespace,
91	app_id=app_id,
92	subject=subject,
93	encode_base64=tracker_config.encode_base64,
94	json_encoder=tracker_config.json_encoder,
95	)
96
NEW 97	return Snowplow.add_tracker(tracker)	×
98
99	@classmethod	12✔
100	def add_tracker(cls, tracker: Tracker) -> Tracker:	12✔
101	"""
102	Add a Snowplow tracker to the Snowplow object
103
104	:param tracker: Tracker object to add to Snowplow
105	:type tracker: Tracker
106	:rtype Tracker
107	"""
NEW 108	if not isinstance(tracker, Tracker):	×
NEW 109	logger.info("Tracker not provided.")	×
NEW 110	return None	×
111
NEW 112	namespace = tracker.get_namespace()	×
113
NEW 114	if namespace in cls._trackers.keys():	×
NEW 115	raise TypeError("Tracker with this namespace already exists")	×
116
NEW 117	cls._trackers[namespace] = tracker	×
NEW 118	logger.info("Tracker with namespace: '" + namespace + "' added to Snowplow")	×
NEW 119	return cls._trackers[namespace]	×
120
121	@classmethod	12✔
122	def remove_tracker(cls, tracker: Tracker):	12✔
123	"""
124	Remove a Snowplow tracker from the Snowplow object if it exists
125
126	:param tracker: Tracker object to remove from Snowplow
127	:type tracker: Tracker \| None
128	"""
NEW 129	namespace = tracker.get_namespace()	×
NEW 130	cls.remove_tracker_by_namespace(namespace)	×
131
132	@classmethod	12✔
133	def remove_tracker_by_namespace(cls, namespace: str):	12✔
134	"""
135	Remove a Snowplow tracker from the Snowplow object using it's namespace if it exists
136
137	:param namespace: Tracker namespace to remove from Snowplow
138	:type tracker: String \| None
139	"""
NEW 140	if not cls._trackers.pop(namespace, False):	×
NEW 141	logger.info("Tracker with namespace: '" + namespace + "' does not exist")	×
NEW 142	return	×
NEW 143	logger.info("Tracker with namespace: '" + namespace + "' removed from Snowplow")	×
144
145	@classmethod	12✔
146	def reset(cls):	8✔
147	"""
148	Remove all active Snowplow trackers from the Snowplow object
149	"""
NEW 150	cls._trackers = {}	×
151
152	@classmethod	12✔
153	def get_tracker(cls, namespace: str) -> Tracker:	12✔
154	"""
155	Returns a Snowplow tracker from the Snowplow object if it exists
156	:param namespace: Snowplow tracker namespace
157	:type namespace: string
158	:rtype: Tracker
159	"""
NEW 160	if namespace in cls._trackers.keys():	×
NEW 161	return cls._trackers[namespace]	×
NEW 162	return None	×

snowplow / snowplow-python-tracker / 3956992538

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous