Skip to content

Add X-Incognia-Latency header with previous request latency #100

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
figueredo merged 2 commits into from
Apr 20, 2026
Merged

Add X-Incognia-Latency header with previous request latency #100

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
09f1e55
Add X-Incognia-Latency header with previous request latency
alangalvino Apr 15, 2026
af33dd5
Simplify IncogniaAPI post param handling
alangalvino Apr 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 37 additions & 10 deletions incognia/api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import datetime as dt
from typing import Optional, List
import time
from threading import Lock
from typing import Optional, List, Any

from .datetime_util import has_timezone, datetime_valid
from .endpoints import Endpoints
Expand All @@ -21,10 +23,34 @@
from .base_request import BaseRequest, JSON_CONTENT_HEADER


LATENCY_HEADER = 'X-Incognia-Latency'


class IncogniaAPI(metaclass=Singleton):
def __init__(self, client_id: str, client_secret: str):
self.__token_manager = TokenManager(client_id, client_secret)
self.__request = BaseRequest()
self.__last_latency_ms: Optional[int] = None
self.__last_latency_mutex = Lock()

def __get_last_latency(self) -> Optional[int]:
with self.__last_latency_mutex:
return self.__last_latency_ms

def __set_last_latency(self, latency_ms: int) -> None:
with self.__last_latency_mutex:
self.__last_latency_ms = latency_ms

def __post(self, url: str, headers: dict, data: bytes, params: Optional[Any] = None):
request_headers = dict(headers)
last_latency = self.__get_last_latency()
if last_latency is not None:
request_headers[LATENCY_HEADER] = str(last_latency)

start = time.monotonic()
response = self.__request.post(url, headers=request_headers, data=data, params=params)
self.__set_last_latency(int((time.monotonic() - start) * 1000))
return response

def __get_authorization_header(self) -> dict:
access_token, token_type = self.__token_manager.get()
Expand Down Expand Up @@ -66,7 +92,7 @@ def register_new_signup(self,
'custom_properties': custom_properties
}
data = encode(body)
return self.__request.post(Endpoints.SIGNUPS, headers=headers, data=data)
return self.__post(Endpoints.SIGNUPS, headers=headers, data=data)

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
Expand All @@ -93,7 +119,7 @@ def register_new_web_signup(self,
'person_id': person_id,
}
data = encode(body)
return self.__request.post(Endpoints.SIGNUPS, headers=headers, data=data)
return self.__post(Endpoints.SIGNUPS, headers=headers, data=data)

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
Expand Down Expand Up @@ -136,7 +162,8 @@ def register_feedback(self,
if expires_at is not None:
body['expires_at'] = expires_at.isoformat()
data = encode(body)
return self.__request.post(Endpoints.FEEDBACKS, headers=headers, data=data)
self.__post(Endpoints.FEEDBACKS, headers=headers, data=data)
return None

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
Expand Down Expand Up @@ -202,8 +229,8 @@ def register_payment(self,
'related_web_request_token': related_web_request_token,
}
data = encode(body)
return self.__request.post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)
return self.__post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
Expand Down Expand Up @@ -256,8 +283,8 @@ def register_login(self,
'related_web_request_token': related_web_request_token,
}
data = encode(body)
return self.__request.post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)
return self.__post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
Expand Down Expand Up @@ -291,8 +318,8 @@ def register_web_login(self,
'tenant_id': tenant_id,
}
data = encode(body)
return self.__request.post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)
return self.__post(Endpoints.TRANSACTIONS, headers=headers, params=params,
data=data)

except IncogniaHTTPError as e:
raise IncogniaHTTPError(e) from None
123 changes: 102 additions & 21 deletions tests/test_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,12 @@
from unittest import TestCase
from unittest.mock import patch, Mock

from incognia.api import IncogniaAPI
from incognia.api import IncogniaAPI, LATENCY_HEADER
from incognia.base_request import BaseRequest
from incognia.endpoints import Endpoints
from incognia.exceptions import IncogniaHTTPError, IncogniaError
from incognia.json_util import encode
from incognia.singleton import Singleton
from incognia.token_manager import TokenValues, TokenManager


Expand Down Expand Up @@ -283,12 +284,82 @@ class TestIncogniaAPI(TestCase):
})
DEFAULT_PARAMS: Final[None] = None

def setUp(self):
Singleton._instances.pop(IncogniaAPI, None)

def test_metaclass_singleton_should_always_return_the_same_instance(self):
api1 = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)
api2 = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)

self.assertEqual(api1, api2)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_new_signup_should_not_send_latency_header_on_first_api_call(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
mock_base_request_post.configure_mock(return_value=self.JSON_RESPONSE)

api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)
response = api.register_new_signup(request_token=self.REQUEST_TOKEN)

mock_token_manager_get.assert_called_once()
headers = mock_base_request_post.call_args.kwargs['headers']
self.assertNotIn(LATENCY_HEADER, headers)
self.assertEqual(response, self.JSON_RESPONSE)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_new_signup_should_send_latency_header_on_second_api_call(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
mock_base_request_post.configure_mock(return_value=self.JSON_RESPONSE)

api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)
api.register_new_signup(request_token=self.REQUEST_TOKEN)
response = api.register_new_signup(request_token=self.REQUEST_TOKEN)

self.assertEqual(mock_token_manager_get.call_count, 2)
first_headers = mock_base_request_post.call_args_list[0].kwargs['headers']
second_headers = mock_base_request_post.call_args_list[1].kwargs['headers']
self.assertNotIn(LATENCY_HEADER, first_headers)
self.assertIn(LATENCY_HEADER, second_headers)
self.assertGreaterEqual(int(second_headers[LATENCY_HEADER]), 0)
self.assertEqual(response, self.JSON_RESPONSE)
Comment thread
alangalvino marked this conversation as resolved.

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_feedback_should_send_latency_header_after_payment(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
mock_base_request_post.configure_mock(side_effect=[self.JSON_RESPONSE, None])

api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)
api.register_payment(self.REQUEST_TOKEN, self.ACCOUNT_ID)
api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE)

self.assertEqual(mock_token_manager_get.call_count, 2)
payment_headers = mock_base_request_post.call_args_list[0].kwargs['headers']
feedback_headers = mock_base_request_post.call_args_list[1].kwargs['headers']
self.assertNotIn(LATENCY_HEADER, payment_headers)
self.assertIn(LATENCY_HEADER, feedback_headers)
self.assertGreaterEqual(int(feedback_headers[LATENCY_HEADER]), 0)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_new_signup_should_send_latency_header_after_feedback(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
mock_base_request_post.configure_mock(side_effect=[None, self.JSON_RESPONSE])

api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)
api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE)
response = api.register_new_signup(request_token=self.REQUEST_TOKEN)

self.assertEqual(mock_token_manager_get.call_count, 2)
feedback_headers = mock_base_request_post.call_args_list[0].kwargs['headers']
signup_headers = mock_base_request_post.call_args_list[1].kwargs['headers']
self.assertNotIn(LATENCY_HEADER, feedback_headers)
self.assertIn(LATENCY_HEADER, signup_headers)
self.assertGreaterEqual(int(signup_headers[LATENCY_HEADER]), 0)
self.assertEqual(response, self.JSON_RESPONSE)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_new_signup_when_request_token_is_valid_should_return_a_valid_dict(
Expand All @@ -301,7 +372,8 @@ def test_register_new_signup_when_request_token_is_valid_should_return_a_valid_d
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_SIGNUP_DATA)
data=self.REGISTER_SIGNUP_DATA,
params=None)

self.assertEqual(response, self.JSON_RESPONSE)

Expand All @@ -317,7 +389,8 @@ def test_register_new_web_signup_when_request_token_is_valid_should_return_a_val
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_WEB_SIGNUP_DATA)
data=self.REGISTER_WEB_SIGNUP_DATA,
params=None)

self.assertEqual(response, self.JSON_RESPONSE)

Expand Down Expand Up @@ -345,7 +418,8 @@ def test_register_new_signup_when_request_token_is_valid_should_return_full_vali
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.FULL_REGISTER_SIGNUP_DATA)
data=self.FULL_REGISTER_SIGNUP_DATA,
params=None)

self.assertEqual(response, self.JSON_RESPONSE)

Expand All @@ -366,7 +440,8 @@ def test_register_new_web_signup_when_request_token_is_valid_should_return_full_
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.FULL_REGISTER_WEB_SIGNUP_DATA)
data=self.FULL_REGISTER_WEB_SIGNUP_DATA,
params=None)

self.assertEqual(response, self.JSON_RESPONSE)

Expand Down Expand Up @@ -405,7 +480,8 @@ def test_register_new_signup_when_request_token_is_invalid_should_raise_an_Incog
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_INVALID_SIGNUP_DATA)
data=self.REGISTER_INVALID_SIGNUP_DATA,
params=None)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
Expand All @@ -421,43 +497,48 @@ def test_register_new_web_signup_if_request_token_is_invalid_should_raise_an_Inc
mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.SIGNUPS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_INVALID_WEB_SIGNUP_DATA)
data=self.REGISTER_INVALID_WEB_SIGNUP_DATA,
params=None)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_feedback_when_required_fields_are_valid_should_work(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)

api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE)
response = api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE)

mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.FEEDBACKS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_VALID_FEEDBACK_DATA)
data=self.REGISTER_VALID_FEEDBACK_DATA,
params=None)
self.assertIsNone(response)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
def test_register_feedback_when_all_fields_are_valid_should_work(
self, mock_token_manager_get: Mock, mock_base_request_post: Mock):
api = IncogniaAPI(self.CLIENT_ID, self.CLIENT_SECRET)

api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE,
occurred_at=self.TIMESTAMP,
expires_at=self.TIMESTAMP,
external_id=self.EXTERNAL_ID,
login_id=self.LOGIN_ID,
payment_id=self.PAYMENT_ID,
signup_id=self.SIGNUP_ID,
account_id=self.ACCOUNT_ID,
installation_id=self.INSTALLATION_ID,
request_token=self.REQUEST_TOKEN,
person_id=self.PERSON_ID)
response = api.register_feedback(self.VALID_EVENT_FEEDBACK_TYPE,
occurred_at=self.TIMESTAMP,
expires_at=self.TIMESTAMP,
external_id=self.EXTERNAL_ID,
login_id=self.LOGIN_ID,
payment_id=self.PAYMENT_ID,
signup_id=self.SIGNUP_ID,
account_id=self.ACCOUNT_ID,
installation_id=self.INSTALLATION_ID,
request_token=self.REQUEST_TOKEN,
person_id=self.PERSON_ID)

mock_token_manager_get.assert_called()
mock_base_request_post.assert_called_with(Endpoints.FEEDBACKS,
headers=self.AUTH_AND_JSON_CONTENT_HEADERS,
data=self.REGISTER_VALID_FEEDBACK_DATA_FULL)
data=self.REGISTER_VALID_FEEDBACK_DATA_FULL,
params=None)
self.assertIsNone(response)

@patch.object(BaseRequest, 'post')
@patch.object(TokenManager, 'get', return_value=TOKEN_VALUES)
Expand Down
Loading