19558051963

Committed 20 Nov 2025 05:48PM UTC coverage: 86.859% (-0.05%) from 86.907%

Build # 19558051963

Build Type

push

github

Committed by

web-flow

Commit Message

Sns:v2 publish (#13399)

Run Details

199 of 279 new or added lines in 5 files covered. (71.33%)

168 existing lines in 9 files now uncovered.

68851 of 79268 relevant lines covered (86.86%)

0.87 hits per line

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

87.04

/localstack-core/localstack/utils/analytics/client.py

"""
Client for the analytics backend.
"""

import logging
from typing import Any

import requests

from localstack import config, constants
from localstack.utils.http import get_proxies
from localstack.utils.time import now

from .events import Event, EventMetadata
from .metadata import ClientMetadata, get_session_id

LOG = logging.getLogger(__name__)


class SessionResponse:
    response: dict[str, Any]
    status: int

    def __init__(self, response: dict[str, Any], status: int = 200):
        self.response = response
        self.status = status

    def track_events(self) -> bool:
        return self.response.get("track_events")

    def __repr__(self):
        return f"SessionResponse({self.status},{self.response!r})"


class AnalyticsClient:
    api: str

    def __init__(self, api=None):
        self.api = (api or constants.ANALYTICS_API).rstrip("/")
        self.debug = config.DEBUG_ANALYTICS

        self.endpoint_session = self.api + "/session"
        self.endpoint_events = self.api + "/events"

        self.localstack_session_id = get_session_id()
        self.session = requests.Session()

    def close(self):
        self.session.close()

    def start_session(self, metadata: ClientMetadata) -> SessionResponse:
        # FIXME: re-using Event as request object this way is kind of a hack
        request = Event(
            "session", EventMetadata(self.localstack_session_id, str(now())), payload=metadata
        )

        response = self.session.post(
            self.endpoint_session,
            headers=self._create_headers(),
            json=request.asdict(),
            proxies=get_proxies(),
        )

        # 403 errors may indicate that track_events=False
        if response.ok or response.status_code == 403:
            return SessionResponse(response.json(), status=response.status_code)

        raise ValueError(
            f"error during session initiation with analytics backend. code: {response.status_code}"
        )

    # TODO: naming seems confusing since this doesn't actually append, but directly sends all passed events via HTTP
    def append_events(self, events: list[Event]):
        # TODO: add compression to append_events
        #  it would maybe be useful to compress analytics data, but it's unclear how that will
        #  affect performance and what the benefit is. need to measure first.

        endpoint = self.endpoint_events

        if not events:
            return

        docs = []
        for event in events:
            try:
                docs.append(event.asdict())
            except Exception:
                if self.debug:
                    LOG.exception("error while recording event %s", event)

        headers = self._create_headers()

        if self.debug:
            LOG.debug("posting to %s events %s", endpoint, docs)

        # FIXME: fault tolerance/timeouts
        response = self.session.post(
            endpoint, json={"events": docs}, headers=headers, proxies=get_proxies()
        )

        if self.debug:
            LOG.debug("response from %s was: %s %s", endpoint, response.status_code, response.text)

        # TODO: Add response type to analytics client
        return response

    def _create_headers(self) -> dict[str, str]:
        return {
            "User-Agent": "localstack/" + constants.VERSION,
            "Localstack-Session-ID": self.localstack_session_id,
        }

1	"""
2	Client for the analytics backend.
3	"""
4
5	import logging	1✔
6	from typing import Any	1✔
7
8	import requests	1✔
9
10	from localstack import config, constants	1✔
11	from localstack.utils.http import get_proxies	1✔
12	from localstack.utils.time import now	1✔
13
14	from .events import Event, EventMetadata	1✔
15	from .metadata import ClientMetadata, get_session_id	1✔
16
17	LOG = logging.getLogger(__name__)	1✔
18
19
20	class SessionResponse:	1✔
21	response: dict[str, Any]	1✔
22	status: int	1✔
23
24	def __init__(self, response: dict[str, Any], status: int = 200):	1✔
25	self.response = response	1✔
26	self.status = status	1✔
27
28	def track_events(self) -> bool:	1✔
29	return self.response.get("track_events")	1✔
30
31	def __repr__(self):
32	return f"SessionResponse({self.status},{self.response!r})"
33
34
35	class AnalyticsClient:	1✔
36	api: str	1✔
37
38	def __init__(self, api=None):	1✔
39	self.api = (api or constants.ANALYTICS_API).rstrip("/")	1✔
40	self.debug = config.DEBUG_ANALYTICS	1✔
41
42	self.endpoint_session = self.api + "/session"	1✔
43	self.endpoint_events = self.api + "/events"	1✔
44
45	self.localstack_session_id = get_session_id()	1✔
46	self.session = requests.Session()	1✔
47
48	def close(self):	1✔
UNCOV 49	self.session.close()	×
50
51	def start_session(self, metadata: ClientMetadata) -> SessionResponse:	1✔
52	# FIXME: re-using Event as request object this way is kind of a hack
53	request = Event(	1✔
54	"session", EventMetadata(self.localstack_session_id, str(now())), payload=metadata
55	)
56
57	response = self.session.post(	1✔
58	self.endpoint_session,
59	headers=self._create_headers(),
60	json=request.asdict(),
61	proxies=get_proxies(),
62	)
63
64	# 403 errors may indicate that track_events=False
65	if response.ok or response.status_code == 403:	1✔
66	return SessionResponse(response.json(), status=response.status_code)	1✔
67
68	raise ValueError(	1✔
69	f"error during session initiation with analytics backend. code: {response.status_code}"
70	)
71
72	# TODO: naming seems confusing since this doesn't actually append, but directly sends all passed events via HTTP
73	def append_events(self, events: list[Event]):	1✔
74	# TODO: add compression to append_events
75	# it would maybe be useful to compress analytics data, but it's unclear how that will
76	# affect performance and what the benefit is. need to measure first.
77
78	endpoint = self.endpoint_events	1✔
79
80	if not events:	1✔
UNCOV 81	return	×
82
83	docs = []	1✔
84	for event in events:	1✔
85	try:	1✔
86	docs.append(event.asdict())	1✔
87	except Exception:	×
88	if self.debug:	×
89	LOG.exception("error while recording event %s", event)	×
90
91	headers = self._create_headers()	1✔
92
93	if self.debug:	1✔
94	LOG.debug("posting to %s events %s", endpoint, docs)	×
95
96	# FIXME: fault tolerance/timeouts
97	response = self.session.post(	1✔
98	endpoint, json={"events": docs}, headers=headers, proxies=get_proxies()
99	)
100
101	if self.debug:	1✔
102	LOG.debug("response from %s was: %s %s", endpoint, response.status_code, response.text)	×
103
104	# TODO: Add response type to analytics client
105	return response	1✔
106
107	def _create_headers(self) -> dict[str, str]:	1✔
108	return {	1✔
109	"User-Agent": "localstack/" + constants.VERSION,
110	"Localstack-Session-ID": self.localstack_session_id,
111	}

localstack / localstack / 19558051963

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