Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
7 changes: 7 additions & 0 deletions CHANGES.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,10 @@
6.1.0 2026-08-18
================

- Added support for the Global Profile API:
- `client.get_global_profile()` to fetch a user's Global Profile
- `client.get_global_profile_by_attributes()` to look up a Global Profile by email and/or phone

6.0.0 2025-05-05
================

Expand Down
21 changes: 21 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -227,4 +227,25 @@ try:
except sift.client.ApiException:
# request failed
pass

# Get the Global Profile for a user
try:
response = client.get_global_profile(
user_id,
global_only=False,
include_own_data=True,
)
except sift.client.ApiException:
# request failed
pass

# Look up a Global Profile by email and/or phone
try:
response = client.get_global_profile_by_attributes(
email="buyer@gmail.com",
phone="+15555550100",
)
except sift.client.ApiException:
# request failed
pass
```
128 changes: 128 additions & 0 deletions sift/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,16 @@ def _psp_merchant_id_url(self, account_id: str, merchant_id: str) -> str:
f"/accounts/{_q(account_id)}/psp_management/merchants/{_q(merchant_id)}"
)

def _global_profile_url(self, account_id: str, user_id: str) -> str:
return self._v3_api(
f"/accounts/{_q(account_id)}/global_profile/users/{_q(user_id)}"
)

def _global_profile_lookup_url(self, account_id: str) -> str:
return self._v3_api(
f"/accounts/{_q(account_id)}/global_profile/lookup"
)

def _verification_send_url(self) -> str:
return self._v1_api("/verification/send")

Expand Down Expand Up @@ -1535,6 +1545,124 @@ def get_a_psp_merchant_profile(

return Response(response)

def get_global_profile(
self,
user_id: str,
global_only: bool = False,
include_own_data: bool = True,
timeout: float | tuple[float, float] | None = None,
) -> Response:
"""Gets the Global Profile for a user.

Args:
user_id:
The ID of the user to fetch the Global Profile for.

global_only (optional):
If True, excludes the requesting tenant's own network
connections from the response. [default: False]

include_own_data (optional):
If True, includes the requested user's own feature values
in the response. [default: True]

timeout (optional):
How many seconds to wait for the server to send data before
giving up, as a float, or a (connect timeout, read timeout) tuple.

Returns:
A sift.client.Response object if the call to the Sift API is successful

Raises:
ApiException: If the call to the Sift API is not successful
"""
_assert_non_empty_str(self.account_id, "account_id")
_assert_non_empty_str(user_id, "user_id", error_cls=ValueError)

params: dict[str, t.Any] = {
"global_only": "true" if global_only else "false",
"include_own_data": "true" if include_own_data else "false",
}

if timeout is None:
timeout = self.timeout

url = self._global_profile_url(self.account_id, user_id)

try:
response = self.session.get(
url,
params=params,
auth=self._auth,
headers=self._default_headers(),
timeout=timeout,
)
except requests.exceptions.RequestException as e:
raise ApiException(str(e), url)

return Response(response)

def get_global_profile_by_attributes(
self,
email: str | None = None,
phone: str | None = None,
timeout: float | tuple[float, float] | None = None,
) -> Response:
"""Looks up the Global Profile for a user by email and/or phone.

Args:
email (optional):
The email address to look up. At least one of `email` or
`phone` must be provided.

phone (optional):
The phone number to look up. At least one of `email` or
`phone` must be provided.

timeout (optional):
How many seconds to wait for the server to send data before
giving up, as a float, or a (connect timeout, read timeout) tuple.

Returns:
A sift.client.Response object if the call to the Sift API is successful

Raises:
ApiException: If the call to the Sift API is not successful
"""
_assert_non_empty_str(self.account_id, "account_id")

email_stripped = email.strip() if email else ""
phone_stripped = phone.strip() if phone else ""

if not email_stripped and not phone_stripped:
raise ValueError("must provide at least one of 'email' or 'phone'")

properties: dict[str, t.Any] = {}

if email_stripped:
properties["email"] = email_stripped

if phone_stripped:
properties["phone"] = phone_stripped

if timeout is None:
timeout = self.timeout

url = self._global_profile_lookup_url(self.account_id)

try:
response = self.session.post(
url,
data=json.dumps(properties),
auth=self._auth,
headers=self._post_headers(),
timeout=timeout,
)
except requests.exceptions.RequestException as e:
raise ApiException(str(e), url)

return Response(response)

def verification_send(
self,
properties: Mapping[str, t.Any],
Expand Down
2 changes: 1 addition & 1 deletion sift/version.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,2 @@
VERSION = "6.0.0"
VERSION = "6.1.0"
API_VERSION = "205"
138 changes: 138 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,23 @@ def response_with_data_header() -> dict[str, t.Any]:
return {"content-type": "application/json; charset=UTF-8"}


def global_profile_response_json() -> str:
return """
{
"status": 0,
"error_message": "OK",
"error_code": null,
"lookback_months": 12,
"profile_summary": {
"identity_found": true,
"has_links": true,
"link_count": 7,
"linked_accounts_count_per_industry": {"finances": 3, "internet": 4}
}
}
"""


class TestSiftPythonClient(TestCase):

def setUp(self) -> None:
Expand Down Expand Up @@ -1616,6 +1633,127 @@ def test_get_psp_merchant_profile_id(self) -> None:
assert isinstance(response.body, dict)
assert "address" in response.body

def test_get_global_profile(self) -> None:
"""Test the GET /v3/accounts/{accountId}/global_profile/users/{userId}"""
test_timeout = 5
mock_response = mock.Mock()
mock_response.content = global_profile_response_json()
mock_response.json.return_value = json.loads(mock_response.content)
mock_response.status_code = 200
mock_response.headers = response_with_data_header()

with mock.patch.object(self.sift_client.session, "get") as mock_get:
mock_get.return_value = mock_response

response = self.sift_client.get_global_profile(
"example_user",
global_only=True,
include_own_data=False,
timeout=test_timeout,
)

mock_get.assert_called_with(
"https://api.sift.com/v3/accounts/ACCT/global_profile/users/example_user",
params={
"global_only": "true",
"include_own_data": "false",
},
headers=mock.ANY,
auth=mock.ANY,
timeout=test_timeout,
)
self.assertIsInstance(response, sift.client.Response)
assert response.is_ok()
assert isinstance(response.body, dict)
assert response.body["profile_summary"]["identity_found"] is True

def test_get_global_profile_default_params(self) -> None:
mock_response = mock.Mock()
mock_response.content = '{"status": 0, "error_message": "OK"}'
mock_response.json.return_value = json.loads(mock_response.content)
mock_response.status_code = 200
mock_response.headers = response_with_data_header()

with mock.patch.object(self.sift_client.session, "get") as mock_get:
mock_get.return_value = mock_response

self.sift_client.get_global_profile("example_user")

mock_get.assert_called_with(
"https://api.sift.com/v3/accounts/ACCT/global_profile/users/example_user",
params={
"global_only": "false",
"include_own_data": "true",
},
headers=mock.ANY,
auth=mock.ANY,
timeout=mock.ANY,
)

def test_get_global_profile_requires_user_id(self) -> None:
with self.assertRaises(ValueError):
self.sift_client.get_global_profile("")

def test_get_global_profile_by_attributes(self) -> None:
"""Test the POST /v3/accounts/{accountId}/global_profile/lookup"""
test_timeout = 5
mock_response = mock.Mock()
mock_response.content = global_profile_response_json()
mock_response.json.return_value = json.loads(mock_response.content)
mock_response.status_code = 200
mock_response.headers = response_with_data_header()

with mock.patch.object(self.sift_client.session, "post") as mock_post:
mock_post.return_value = mock_response

response = self.sift_client.get_global_profile_by_attributes(
email="buyer@example.com",
phone="+15555550100",
timeout=test_timeout,
)

mock_post.assert_called_with(
"https://api.sift.com/v3/accounts/ACCT/global_profile/lookup",
data=json.dumps(
{"email": "buyer@example.com", "phone": "+15555550100"}
),
headers=mock.ANY,
auth=mock.ANY,
timeout=test_timeout,
)
self.assertIsInstance(response, sift.client.Response)
assert response.is_ok()
assert isinstance(response.body, dict)
assert response.body["profile_summary"]["identity_found"] is True

def test_get_global_profile_by_attributes_email_only(self) -> None:
mock_response = mock.Mock()
mock_response.content = '{"status": 0, "error_message": "OK"}'
mock_response.json.return_value = json.loads(mock_response.content)
mock_response.status_code = 200
mock_response.headers = response_with_data_header()

with mock.patch.object(self.sift_client.session, "post") as mock_post:
mock_post.return_value = mock_response

self.sift_client.get_global_profile_by_attributes(
email="buyer@example.com"
)

mock_post.assert_called_with(
"https://api.sift.com/v3/accounts/ACCT/global_profile/lookup",
data=json.dumps({"email": "buyer@example.com"}),
headers=mock.ANY,
auth=mock.ANY,
timeout=mock.ANY,
)

def test_get_global_profile_by_attributes_requires_email_or_phone(
self,
) -> None:
with self.assertRaises(ValueError):
self.sift_client.get_global_profile_by_attributes()

def test_create_psp_merchant_profile(self) -> None:
mock_response = mock.Mock()
mock_response.content = valid_psp_merchant_properties_response()
Expand Down
Loading