Compare commits

..
Author SHA1 Message Date
James HawkinsandGitHub df0dc10f36 Update README.md 2020-02-18 18:20:20 -08:00
42 changed files with 938 additions and 7977 deletions
-65
View File
@@ -1,65 +0,0 @@
name: CI
on:
- pull_request
jobs:
code-quality:
name: Code quality checks
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
with:
fetch-depth: 1
- name: Set up Python 3.8
uses: actions/setup-python@v2
with:
python-version: 3.8
- uses: actions/cache@v1
with:
path: ~/.cache/pip
key: ${{ runner.os }}-pip-${{ hashFiles('setup.py') }}
restore-keys: |
${{ runner.os }}-pip-
- name: Install dev dependencies
run: |
python -m pip install -e .[dev]
if: steps.cache.outputs.cache-hit != 'true'
- name: Check formatting with black
run: |
black --check .
- name: Lint with flake8
run: |
flake8 posthog --ignore E501
- name: Check import order with isort
run: |
isort --check-only .
tests:
name: Python tests
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v1
with:
fetch-depth: 1
- name: Set up Python 3.7
uses: actions/setup-python@v1
with:
python-version: 3.7
- name: Install requirements.txt dependencies with pip
run: |
python -m pip install -e .[test]
- name: Run posthog tests
run: |
pytest --verbose --timeout=30
-38
View File
@@ -1,38 +0,0 @@
name: 'Release'
on:
- workflow_dispatch
jobs:
release:
name: Publish release
runs-on: ubuntu-20.04
env:
TWINE_USERNAME: __token__
TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
steps:
- name: Checkout the repository
uses: actions/checkout@v2
with:
fetch-depth: 0
token: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
- name: Set up Python
uses: actions/setup-python@v2
- name: Detect version
run: echo "REPO_VERSION=$(python3 posthog/version.py)" >> $GITHUB_ENV
- name: Prepare for building release
run: pip install -U pip setuptools wheel twine
- name: Push release to PyPI
run: make release && make release_analytics
- name: Create GitHub release
uses: actions/create-release@v1
env:
GITHUB_TOKEN: ${{ secrets.POSTHOG_BOT_GITHUB_TOKEN }}
with:
tag_name: v${{ env.REPO_VERSION }}
release_name: ${{ env.REPO_VERSION }}
+5 -8
View File
@@ -1,17 +1,14 @@
**sublime**
*.pyc
dist/
dist
*.egg-info
dist
MANIFEST
build/
.eggs/
build
.eggs
.coverage
.vscode/
env/
venv/
env
flake8.out
pylint.out
posthog-analytics
.idea
.python-version
.coverage
-9
View File
@@ -1,9 +0,0 @@
repos:
- repo: https://github.com/psf/black
rev: stable
hooks:
- id: black
- repo: https://github.com/pycqa/isort
rev: 5.7.0
hooks:
- id: isort
-156
View File
@@ -1,156 +0,0 @@
## 3.3.4 - 2024-01-30
1. Update type hints for module variables to work with newer versions of mypy
## 3.3.3 - 2024-01-26
1. Remove new relative date operators, combine into regular date operators
## 3.3.2 - 2024-01-19
1. Return success/failure with all capture calls from module functions
## 3.3.1 - 2024-01-10
1. Make sure we don't override any existing feature flag properties when adding locally evaluated feature flag properties.
## 3.3.0 - 2024-01-09
1. When local evaluation is enabled, we automatically add flag information to all events sent to PostHog, whenever possible. This makes it easier to use these events in experiments.
## 3.2.0 - 2024-01-09
1. Numeric property handling for feature flags now does the expected: When passed in a number, we do a numeric comparison. When passed in a string, we do a string comparison. Previously, we always did a string comparison.
2. Add support for relative date operators for local evaluation.
## 3.1.0 - 2023-12-04
1. Increase maximum event size and batch size
## 3.0.2 - 2023-08-17
1. Returns the current flag property with $feature_flag_called events, to make it easier to use in experiments
## 3.0.1 - 2023-04-21
1. Restore how feature flags work when the client library is disabled: All requests return `None` and no events are sent when the client is disabled.
2. Add a `feature_flag_definitions()` debug option, which returns currently loaded feature flag definitions. You can use this to more cleverly decide when to request local evaluation of feature flags.
## 3.0.0 - 2023-04-14
Breaking change:
All events by default now send the `$geoip_disable` property to disable geoip lookup in app. This is because usually we don't
want to update person properties to take the server's location.
The same now happens for feature flag requests, where we discard the IP address of the server for matching on geoip properties like city, country, continent.
To restore previous behaviour, you can set the default to False like so:
```python
posthog.disable_geoip = False
# // and if using client instantiation:
posthog = Posthog('api_key', disable_geoip=False)
```
## 2.5.0 - 2023-04-10
1. Add option for instantiating separate client object
## 2.4.2 - 2023-03-30
1. Update backoff dependency for posthoganalytics package to be the same as posthog package
## 2.4.1 - 2023-03-17
1. Removes accidental print call left in for decide response
## 2.4.0 - 2023-03-14
1. Support evaluating all cohorts in feature flags for local evaluation
## 2.3.1 - 2023-02-07
1. Log instead of raise error on posthog personal api key errors
2. Remove upper bound on backoff dependency
## 2.3.0 - 2023-01-31
1. Add support for returning payloads of matched feature flags
## 2.2.0 - 2022-11-14
Changes:
1. Add support for feature flag variant overrides with local evaluation
## 2.1.2 - 2022-09-15
Changes:
1. Fixes issues with date comparison.
## 2.1.1 - 2022-09-14
Changes:
1. Feature flags local evaluation now supports date property filters as well. Accepts both strings and datetime objects.
## 2.1.0 - 2022-08-11
Changes:
1. Feature flag defaults have been removed
2. Setup logging only when debug mode is enabled.
## 2.0.1 - 2022-08-04
- Make poll_interval configurable
- Add `send_feature_flag_events` parameter to feature flag calls, which determine whether the `$feature_flag_called` event should be sent or not.
- Add `only_evaluate_locally` parameter to feature flag calls, which determines whether the feature flag should only be evaluated locally or not.
## 2.0.0 - 2022-08-02
Breaking changes:
1. The minimum version requirement for PostHog servers is now 1.38. If you're using PostHog Cloud, you satisfy this requirement automatically.
2. Feature flag defaults apply only when there's an error fetching feature flag results. Earlier, if the default was set to `True`, even if a flag resolved to `False`, the default would override this.
**Note: These are removed in 2.0.2**
3. Feature flag remote evaluation doesn't require a personal API key.
New Changes:
1. You can now evaluate feature flags locally (i.e. without sending a request to your PostHog servers) by setting a personal API key, and passing in groups and person properties to `is_feature_enabled` and `get_feature_flag` calls.
2. Introduces a `get_all_flags` method that returns all feature flags. This is useful for when you want to seed your frontend with some initial flags, given a user ID.
## 1.4.9 - 2022-06-13
- Support for sending feature flags with capture calls
## 1.4.8 - 2022-05-12
- Support multi variate feature flags
## 1.4.7 - 2022-04-25
- Allow feature flags usage without project_api_key
## 1.4.1 - 2021-05-28
- Fix packaging issues with Sentry integrations
## 1.4.0 - 2021-05-18
- Improve support for `project_api_key` (#32)
- Resolve polling issues with feature flags (#29)
- Add Sentry (and Sentry+Django) integrations (#13)
- Fix feature flag issue with no percentage rollout (#30)
## 1.3.1 - 2021-05-07
- Add `$set` and `$set_once` support (#23)
- Add distinct ID to `$create_alias` event (#27)
- Add `UUID` to `ID_TYPES` (#26)
## 1.2.1 - 2021-02-05
Initial release logged in CHANGELOG.md.
+1 -1
View File
@@ -1,4 +1,4 @@
Copyright (c) 2023 PostHog (part of Hiberly Inc)
Copyright (c) 2020 PostHog (part of Hiberly Inc)
Copyright (c) 2013 Segment Inc. friends@segment.com
+3 -10
View File
@@ -1,10 +1,7 @@
lint:
test:
pylint --rcfile=.pylintrc --reports=y --exit-zero analytics | tee pylint.out
flake8 --max-complexity=10 --statistics analytics > flake8.out || true
test:
coverage run -m pytest
coverage report
coverage run --branch --include=analytics/\* --omit=*/test* setup.py test
release:
rm -rf dist/*
@@ -29,8 +26,4 @@ release_analytics:
e2e_test:
.buildscripts/e2e.sh
django_example:
python -m pip install -e ".[sentry]"
cd sentry_django_example && python manage.py runserver 8080
.PHONY: test lint release e2e_test
.PHONY: test release e2e_test
+2 -43
View File
@@ -1,46 +1,5 @@
# PostHog Python
[![PyPI](https://img.shields.io/pypi/v/posthog)](https://pypi.org/project/posthog/)
Please see the main [PostHog docs](https://github.com/PostHog/posthog/wiki).
Please see the [Python integration docs](https://posthog.com/docs/integrations/python-integration) for details.
## Development
### Testing Locally
1. Run `python3 -m venv env` (creates virtual environment called "env")
2. Run `source env/bin/activate` (activates the virtual environment)
3. Run `python3 -m pip install -e ".[test]"` (installs the package in develop mode, along with test dependencies)
4. Run `make test`
1. To run a specific test do `pytest -k test_no_api_key`
### Running Locally
Assuming you have a [local version of PostHog](https://posthog.com/docs/developing-locally) running, you can run `python3 example.py` to see the library in action.
### Running the Django Sentry Integration Locally
There's a sample Django project included, called `sentry_django_example`, which explains how to use PostHog with Sentry.
There's 2 places of importance (Changes required are all marked with TODO in the sample project directory)
1. Settings.py
1. Input your Sentry DSN
2. Input your Sentry Org and ProjectID details into `PosthogIntegration()`
3. Add `POSTHOG_DJANGO` to settings.py. This allows the `PosthogDistinctIdMiddleware` to get the distinct_ids
2. urls.py
1. This includes the `sentry-debug/` endpoint, which generates an exception
To run things: `make django_example`. This installs the posthog-python library with the sentry-sdk add-on, and then runs the django app.
Also start the PostHog app locally.
Then navigate to `http://127.0.0.1:8080/sentry-debug/` and you should get an event in both Sentry and PostHog, with links to each other.
### Releasing Versions
Updated are released using GitHub Actions: after bumping `version.py` in `master` and adding to `CHANGELOG.md`, go to [our release workflow's page](https://github.com/PostHog/posthog-python/actions/workflows/release.yaml) and dispatch it manually, using workflow from `master`.
## Questions?
### [Join our Slack community.](https://join.slack.com/t/posthogusers/shared_invite/enQtOTY0MzU5NjAwMDY3LTc2MWQ0OTZlNjhkODk3ZDI3NDVjMDE1YjgxY2I4ZjI4MzJhZmVmNjJkN2NmMGJmMzc2N2U3Yjc3ZjI5NGFlZDQ)
Specifically, the Python integration details are [here](https://github.com/PostHog/posthog/wiki/Python-integration).
+7 -88
View File
@@ -1,101 +1,20 @@
# PostHog Python library example
# Import the library
import time
import posthog
posthog.debug = True
# You can find this key on the /setup page in PostHog
posthog.project_api_key = "phc_gtWmTq3Pgl06u4sZY3TRcoQfp42yfuXHKoe8ZVSR6Kh"
posthog.personal_api_key = "phx_fiRCOQkTA3o2ePSdLrFDAILLHjMu2Mv52vUi8MNruIm"
posthog.api_key = '<your key>'
# Where you host PostHog, with no trailing /.
# You can remove this line if you're using posthog.com
posthog.host = "http://localhost:8000"
posthog.poll_interval = 10
print(
posthog.feature_enabled(
"person-on-events-enabled",
"12345",
groups={"organization": str("0182ee91-8ef7-0000-4cb9-fedc5f00926a")},
group_properties={
"organization": {
"id": "0182ee91-8ef7-0000-4cb9-fedc5f00926a",
"created_at": "2022-06-30 11:44:52.984121+00:00",
}
},
only_evaluate_locally=True,
)
)
posthog.host = 'http://127.0.0.1:8000'
# Capture an event
posthog.capture("distinct_id", "event", {"property1": "value", "property2": "value"}, send_feature_flags=True)
posthog.capture('distinct_id', 'event', {'property1': 'value', 'property2': 'value'})
print(posthog.feature_enabled("beta-feature", "distinct_id"))
print(posthog.feature_enabled("beta-feature-groups", "distinct_id", groups={"company": "id:5"}))
# Alias a previous distinct id with a new one
posthog.alias('distinct_id', 'new_distinct_id')
print(posthog.feature_enabled("beta-feature", "distinct_id"))
# get payload
print(posthog.get_feature_flag_payload("beta-feature", "distinct_id"))
print(posthog.get_all_flags_and_payloads("distinct_id"))
exit()
# # Alias a previous distinct id with a new one
posthog.alias("distinct_id", "new_distinct_id")
posthog.capture("new_distinct_id", "event2", {"property1": "value", "property2": "value"})
posthog.capture(
"new_distinct_id", "event-with-groups", {"property1": "value", "property2": "value"}, groups={"company": "id:5"}
)
# # Add properties to the person
posthog.identify("new_distinct_id", {"email": "something@something.com"})
# Add properties to a group
posthog.group_identify("company", "id:5", {"employees": 11})
# properties set only once to the person
posthog.set_once("new_distinct_id", {"self_serve_signup": True})
posthog.set_once(
"new_distinct_id", {"self_serve_signup": False}
) # this will not change the property (because it was already set)
posthog.set("new_distinct_id", {"current_browser": "Chrome"})
posthog.set("new_distinct_id", {"current_browser": "Firefox"})
# #############################################################################
# Make sure you have a personal API key for the examples below
# Local Evaluation
# If flag has City=Sydney, this call doesn't go to `/decide`
print(posthog.feature_enabled("test-flag", "distinct_id_random_22", person_properties={"$geoip_city_name": "Sydney"}))
print(
posthog.feature_enabled(
"test-flag",
"distinct_id_random_22",
person_properties={"$geoip_city_name": "Sydney"},
only_evaluate_locally=True,
)
)
print(posthog.get_all_flags("distinct_id_random_22"))
print(posthog.get_all_flags("distinct_id_random_22", only_evaluate_locally=True))
print(
posthog.get_all_flags(
"distinct_id_random_22", person_properties={"$geoip_city_name": "Sydney"}, only_evaluate_locally=True
)
)
posthog.shutdown()
# Add properties to the person
posthog.identify('distinct_id', {'email': 'something@something.com'})
+34 -382
View File
@@ -1,224 +1,66 @@
import datetime # noqa: F401
from typing import Callable, Dict, Optional, Tuple # noqa: F401
from posthog.client import Client
from posthog.version import VERSION
from posthog.client import Client
from typing import Optional, Dict, Callable
__version__ = VERSION
"""Settings."""
api_key = None # type: Optional[str]
host = None # type: Optional[str]
on_error = None # type: Optional[Callable]
debug = False # type: bool
send = True # type: bool
sync_mode = False # type: bool
disabled = False # type: bool
personal_api_key = None # type: Optional[str]
project_api_key = None # type: Optional[str]
poll_interval = 30 # type: int
disable_geoip = True # type: bool
api_key: str = None
host: str = None
on_error: Callable = None
debug: bool = False
send: bool = True
sync_mode:bool = False
default_client = None # type: Optional[Client]
default_client = None
def capture(
distinct_id, # type: str
event, # type: str
properties=None, # type: Optional[Dict]
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
groups=None, # type: Optional[Dict]
send_feature_flags=False,
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
def capture(distinct_id: str, event: str, properties: Optional[Dict]=None, context: Optional[Dict]=None,
timestamp: Optional[str]=None, message_id: Optional[str]=None) -> None:
"""
Capture allows you to capture anything a user does within your system, which you can later use in PostHog to find patterns in usage, work out which features to improve or where people are giving up.
A `capture` call requires
- `distinct id` which uniquely identifies your user
- `event name` to specify the event
- `event name` to make sure
- We recommend using [verb] [noun], like `movie played` or `movie updated` to easily identify what your events mean later on.
Optionally you can submit
- `properties`, which can be a dict with any information you'd like to add
- `groups`, which is a dict of group type -> group key mappings
For example:
```python
posthog.capture('distinct id', 'opened app')
posthog.capture('distinct id', 'movie played', {'movie_id': '123', 'category': 'romcom'})
posthog.capture('distinct id', 'purchase', groups={'company': 'id:5'})
```
"""
return _proxy(
"capture",
distinct_id=distinct_id,
event=event,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
groups=groups,
send_feature_flags=send_feature_flags,
disable_geoip=disable_geoip,
)
_proxy('capture', distinct_id=distinct_id, event=event, properties=properties, context=context, timestamp=timestamp, message_id=message_id)
def identify(
distinct_id, # type: str
properties=None, # type: Optional[Dict]
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
def identify(distinct_id: str, properties: Optional[Dict]=None, context: Optional[Dict]=None, timestamp: Optional[str]=None,
message_id=None) -> None:
"""
Identify lets you add metadata on your users so you can more easily identify who they are in PostHog, and even do things like segment users by these properties.
An `identify` call requires
- `distinct id` which uniquely identifies your user
- `properties` with a dict with any key: value pairs
- `properties` with a dict with any key: value pairs
For example:
```python
posthog.identify('distinct id', {
posthog.capture('distinct id', {
'email': 'dwayne@gmail.com',
'name': 'Dwayne Johnson'
})
```
"""
return _proxy(
"identify",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
_proxy('identify', distinct_id=distinct_id, properties=properties, context=context, timestamp=timestamp, message_id=message_id)
def group(*args, **kwargs):
"""Send a group call."""
_proxy('group', *args, **kwargs)
def set(
distinct_id, # type: str
properties=None, # type: Optional[Dict]
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
"""
Set properties on a user record.
This will overwrite previous people property values, just like `identify`.
A `set` call requires
- `distinct id` which uniquely identifies your user
- `properties` with a dict with any key: value pairs
For example:
```python
posthog.set('distinct id', {
'current_browser': 'Chrome',
})
```
"""
return _proxy(
"set",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
def set_once(
distinct_id, # type: str
properties=None, # type: Optional[Dict]
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
"""
Set properties on a user record, only if they do not yet exist.
This will not overwrite previous people property values, unlike `identify`.
A `set_once` call requires
- `distinct id` which uniquely identifies your user
- `properties` with a dict with any key: value pairs
For example:
```python
posthog.set_once('distinct id', {
'referred_by': 'friend',
})
```
"""
return _proxy(
"set_once",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
def group_identify(
group_type, # type: str
group_key, # type: str
properties=None, # type: Optional[Dict]
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
"""
Set properties on a group
A `group_identify` call requires
- `group_type` type of your group
- `group_key` unique identifier of the group
- `properties` with a dict with any key: value pairs
For example:
```python
posthog.group_identify('company', 5, {
'employees': 11,
})
```
"""
return _proxy(
"group_identify",
group_type=group_type,
group_key=group_key,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
def alias(
previous_id, # type: str
distinct_id, # type: str
context=None, # type: Optional[Dict]
timestamp=None, # type: Optional[datetime.datetime]
uuid=None, # type: Optional[str]
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> Tuple[bool, dict]
def alias(previous_id: str, distinct_id: str, context: Optional[Dict]=None, timestamp: Optional[str]=None, message_id: Optional[str]=None) -> None:
"""
To marry up whatever a user does before they sign up or log in with what they do after you need to make an alias call. This will allow you to answer questions like "Which marketing channels leads to users churning after a month?" or "What do users do on our website before signing up?"
@@ -235,232 +77,42 @@ def alias(
posthog.alias('anonymous session id', 'distinct id')
```
"""
return _proxy(
"alias",
previous_id=previous_id,
distinct_id=distinct_id,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
def feature_enabled(
key, # type: str
distinct_id, # type: str
groups={}, # type: dict
person_properties={}, # type: dict
group_properties={}, # type: dict
only_evaluate_locally=False, # type: bool
send_feature_flag_events=True, # type: bool
disable_geoip=None, # type: Optional[bool]
):
# type: (...) -> bool
"""
Use feature flags to enable or disable features for users.
For example:
```python
if posthog.feature_enabled('beta feature', 'distinct id'):
# do something
if posthog.feature_enabled('groups feature', 'distinct id', groups={"organization": "5"}):
# do something
```
You can call `posthog.load_feature_flags()` before to make sure you're not doing unexpected requests.
"""
return _proxy(
"feature_enabled",
key=key,
distinct_id=distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
send_feature_flag_events=send_feature_flag_events,
disable_geoip=disable_geoip,
)
def get_feature_flag(
key, # type: str
distinct_id, # type: str
groups={}, # type: dict
person_properties={}, # type: dict
group_properties={}, # type: dict
only_evaluate_locally=False, # type: bool
send_feature_flag_events=True, # type: bool
disable_geoip=None, # type: Optional[bool]
):
"""
Get feature flag variant for users. Used with experiments.
Example:
```python
if posthog.get_feature_flag('beta-feature', 'distinct_id') == 'test-variant':
# do test variant code
if posthog.get_feature_flag('beta-feature', 'distinct_id') == 'control':
# do control code
```
`groups` are a mapping from group type to group key. So, if you have a group type of "organization" and a group key of "5",
you would pass groups={"organization": "5"}.
`group_properties` take the format: { group_type_name: { group_properties } }
So, for example, if you have the group type "organization" and the group key "5", with the properties name, and employee count,
you'll send these as:
```python
group_properties={"organization": {"name": "PostHog", "employees": 11}}
```
"""
return _proxy(
"get_feature_flag",
key=key,
distinct_id=distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
send_feature_flag_events=send_feature_flag_events,
disable_geoip=disable_geoip,
)
def get_all_flags(
distinct_id, # type: str
groups={}, # type: dict
person_properties={}, # type: dict
group_properties={}, # type: dict
only_evaluate_locally=False, # type: bool
disable_geoip=None, # type: Optional[bool]
):
"""
Get all flags for a given user.
Example:
```python
flags = posthog.get_all_flags('distinct_id')
```
flags are key-value pairs where the key is the flag key and the value is the flag variant, or True, or False.
"""
return _proxy(
"get_all_flags",
distinct_id=distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
disable_geoip=disable_geoip,
)
def get_feature_flag_payload(
key,
distinct_id,
match_value=None,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
send_feature_flag_events=True,
disable_geoip=None, # type: Optional[bool]
):
return _proxy(
"get_feature_flag_payload",
key=key,
distinct_id=distinct_id,
match_value=match_value,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
send_feature_flag_events=send_feature_flag_events,
disable_geoip=disable_geoip,
)
def get_all_flags_and_payloads(
distinct_id,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
disable_geoip=None, # type: Optional[bool]
):
return _proxy(
"get_all_flags_and_payloads",
distinct_id=distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
disable_geoip=disable_geoip,
)
def feature_flag_definitions():
"""Returns loaded feature flags, if any. Helpful for debugging what flag information you have loaded."""
return _proxy("feature_flag_definitions")
def load_feature_flags():
"""Load feature flag definitions from PostHog."""
return _proxy("load_feature_flags")
_proxy('alias', previous_id=previous_id, distinct_id=distinct_id, context=context, timestamp=timestamp, message_id=message_id)
def page(*args, **kwargs):
"""Send a page call."""
_proxy("page", *args, **kwargs)
_proxy('page', *args, **kwargs)
def screen(*args, **kwargs):
"""Send a screen call."""
_proxy("screen", *args, **kwargs)
_proxy('screen', *args, **kwargs)
def flush():
"""Tell the client to flush."""
_proxy("flush")
_proxy('flush')
def join():
"""Block program until the client clears the queue"""
_proxy("join")
_proxy('join')
def shutdown():
"""Flush all messages and cleanly shutdown the client"""
_proxy("flush")
_proxy("join")
_proxy('flush')
_proxy('join')
def _proxy(method, *args, **kwargs):
"""Create an analytics client if one doesn't exist and send to it."""
global default_client
if not default_client:
default_client = Client(
api_key,
host=host,
debug=debug,
on_error=on_error,
send=send,
sync_mode=sync_mode,
personal_api_key=personal_api_key,
project_api_key=project_api_key,
poll_interval=poll_interval,
disabled=disabled,
disable_geoip=disable_geoip,
)
# always set incase user changes it
default_client.disabled = disabled
default_client.debug = debug
default_client = Client(api_key, host=host, debug=debug,
on_error=on_error, send=send,
sync_mode=sync_mode)
fn = getattr(default_client, method)
return fn(*args, **kwargs)
class Posthog(Client):
pass
fn(*args, **kwargs)
+135 -648
View File
@@ -1,17 +1,15 @@
import atexit
from datetime import datetime
from uuid import uuid4
import logging
import numbers
from datetime import datetime, timedelta
from uuid import UUID
import atexit
from dateutil.tz import tzutc
from six import string_types
from posthog.utils import guess_timezone, clean
from posthog.consumer import Consumer
from posthog.feature_flags import InconclusiveMatchError, match_feature_flag_properties
from posthog.poller import Poller
from posthog.request import APIError, batch_post, decide, get
from posthog.utils import SizeLimitedDict, clean, guess_timezone
from posthog.request import post
from posthog.version import VERSION
try:
@@ -20,43 +18,21 @@ except ImportError:
import Queue as queue
ID_TYPES = (numbers.Number, string_types, UUID)
MAX_DICT_SIZE = 50_000
ID_TYPES = (numbers.Number, string_types)
class Client(object):
"""Create a new PostHog client."""
log = logging.getLogger('posthog')
log = logging.getLogger("posthog")
def __init__(self, api_key=None, host=None, debug=False,
max_queue_size=10000, send=True, on_error=None, flush_at=100,
flush_interval=0.5, gzip=False, max_retries=3,
sync_mode=False, timeout=15, thread=1):
require('api_key', api_key, string_types)
def __init__(
self,
api_key=None,
host=None,
debug=False,
max_queue_size=10000,
send=True,
on_error=None,
flush_at=100,
flush_interval=0.5,
gzip=False,
max_retries=3,
sync_mode=False,
timeout=15,
thread=1,
poll_interval=30,
personal_api_key=None,
project_api_key=None,
disabled=False,
disable_geoip=True,
):
self.queue = queue.Queue(max_queue_size)
# api_key: This should be the Team API Key (token), public
self.api_key = project_api_key or api_key
require("api_key", self.api_key, string_types)
self.api_key = api_key
self.on_error = on_error
self.debug = debug
self.send = send
@@ -64,25 +40,9 @@ class Client(object):
self.host = host
self.gzip = gzip
self.timeout = timeout
self.feature_flags = None
self.feature_flags_by_key = None
self.group_type_mapping = None
self.cohorts = None
self.poll_interval = poll_interval
self.poller = None
self.distinct_ids_feature_flags_reported = SizeLimitedDict(MAX_DICT_SIZE, set)
self.disabled = disabled
self.disable_geoip = disable_geoip
# personal_api_key: This should be a generated Personal API Key, private
self.personal_api_key = personal_api_key
if debug:
# Ensures that debug level messages are logged when debug mode is on.
# Otherwise, defaults to WARNING level. See https://docs.python.org/3/howto/logging.html#what-happens-if-no-configuration-is-provided
logging.basicConfig()
self.log.setLevel(logging.DEBUG)
else:
self.log.setLevel(logging.WARNING)
if sync_mode:
self.consumers = None
@@ -98,15 +58,9 @@ class Client(object):
for n in range(thread):
self.consumers = []
consumer = Consumer(
self.queue,
self.api_key,
host=host,
on_error=on_error,
flush_at=flush_at,
flush_interval=flush_interval,
gzip=gzip,
retries=max_retries,
timeout=timeout,
self.queue, api_key, host=host, on_error=on_error,
flush_at=flush_at, flush_interval=flush_interval,
gzip=gzip, retries=max_retries, timeout=timeout,
)
self.consumers.append(consumer)
@@ -114,276 +68,183 @@ class Client(object):
if send:
consumer.start()
def identify(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
def identify(self, distinct_id=None, properties=None, context=None, timestamp=None,
message_id=None):
properties = properties or {}
context = context or {}
require("distinct_id", distinct_id, ID_TYPES)
require("properties", properties, dict)
require('distinct_id', distinct_id, ID_TYPES)
require('properties', properties, dict)
msg = {
"timestamp": timestamp,
"context": context,
"distinct_id": distinct_id,
"$set": properties,
"event": "$identify",
"uuid": uuid,
'timestamp': timestamp,
'context': context,
'type': 'identify',
'distinct_id': distinct_id,
'$set': properties,
'event': '$identify',
'messageId': message_id,
}
return self._enqueue(msg, disable_geoip)
return self._enqueue(msg)
def get_feature_variants(
self, distinct_id, groups=None, person_properties=None, group_properties=None, disable_geoip=None
):
resp_data = self.get_decide(distinct_id, groups, person_properties, group_properties, disable_geoip)
return resp_data["featureFlags"]
def get_feature_payloads(
self, distinct_id, groups=None, person_properties=None, group_properties=None, disable_geoip=None
):
resp_data = self.get_decide(distinct_id, groups, person_properties, group_properties, disable_geoip)
return resp_data["featureFlagPayloads"]
def get_decide(self, distinct_id, groups=None, person_properties=None, group_properties=None, disable_geoip=None):
require("distinct_id", distinct_id, ID_TYPES)
if disable_geoip is None:
disable_geoip = self.disable_geoip
if groups:
require("groups", groups, dict)
else:
groups = {}
request_data = {
"distinct_id": distinct_id,
"groups": groups,
"person_properties": person_properties,
"group_properties": group_properties,
"disable_geoip": disable_geoip,
}
resp_data = decide(self.api_key, self.host, timeout=10, **request_data)
return resp_data
def capture(
self,
distinct_id=None,
event=None,
properties=None,
context=None,
timestamp=None,
uuid=None,
groups=None,
send_feature_flags=False,
disable_geoip=None,
):
def capture(self, distinct_id=None, event=None, properties=None, context=None,
timestamp=None, message_id=None):
properties = properties or {}
context = context or {}
require("distinct_id", distinct_id, ID_TYPES)
require("properties", properties, dict)
require("event", event, string_types)
require('distinct_id', distinct_id, ID_TYPES)
require('properties', properties, dict)
require('event', event, string_types)
msg = {
"properties": properties,
"timestamp": timestamp,
"context": context,
"distinct_id": distinct_id,
"event": event,
"uuid": uuid,
'properties': properties,
'timestamp': timestamp,
'context': context,
'distinct_id': distinct_id,
'type': 'capture',
'event': event,
'messageId': message_id,
}
if groups:
require("groups", groups, dict)
msg["properties"]["$groups"] = groups
return self._enqueue(msg)
extra_properties = {}
feature_variants = {}
if send_feature_flags:
try:
feature_variants = self.get_feature_variants(distinct_id, groups, disable_geoip=disable_geoip)
except Exception as e:
self.log.exception(f"[FEATURE FLAGS] Unable to get feature variants: {e}")
elif self.feature_flags:
# Local evaluation is enabled, flags are loaded, so try and get all flags we can without going to the server
feature_variants = self.get_all_flags(
distinct_id, groups=(groups or {}), disable_geoip=disable_geoip, only_evaluate_locally=True
)
for feature, variant in feature_variants.items():
extra_properties[f"$feature/{feature}"] = variant
active_feature_flags = [key for (key, value) in feature_variants.items() if value is not False]
if active_feature_flags:
extra_properties["$active_feature_flags"] = active_feature_flags
if extra_properties:
msg["properties"] = {**extra_properties, **msg["properties"]}
return self._enqueue(msg, disable_geoip)
def set(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
properties = properties or {}
def alias(self, previous_id=None, distinct_id=None, context=None,
timestamp=None, message_id=None):
context = context or {}
require("distinct_id", distinct_id, ID_TYPES)
require("properties", properties, dict)
require('previous_id', previous_id, ID_TYPES)
require('distinct_id', distinct_id, ID_TYPES)
msg = {
"timestamp": timestamp,
"context": context,
"distinct_id": distinct_id,
"$set": properties,
"event": "$set",
"uuid": uuid,
}
return self._enqueue(msg, disable_geoip)
def set_once(self, distinct_id=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
properties = properties or {}
context = context or {}
require("distinct_id", distinct_id, ID_TYPES)
require("properties", properties, dict)
msg = {
"timestamp": timestamp,
"context": context,
"distinct_id": distinct_id,
"$set_once": properties,
"event": "$set_once",
"uuid": uuid,
}
return self._enqueue(msg, disable_geoip)
def group_identify(
self,
group_type=None,
group_key=None,
properties=None,
context=None,
timestamp=None,
uuid=None,
disable_geoip=None,
):
properties = properties or {}
context = context or {}
require("group_type", group_type, ID_TYPES)
require("group_key", group_key, ID_TYPES)
require("properties", properties, dict)
msg = {
"event": "$groupidentify",
"properties": {
"$group_type": group_type,
"$group_key": group_key,
"$group_set": properties,
'properties': {
'distinct_id': previous_id,
'alias': distinct_id,
},
"distinct_id": "${}_{}".format(group_type, group_key),
"timestamp": timestamp,
"context": context,
"uuid": uuid,
'timestamp': timestamp,
'context': context,
'type': 'alias',
'event': '$create_alias'
}
return self._enqueue(msg, disable_geoip)
return self._enqueue(msg)
def alias(self, previous_id=None, distinct_id=None, context=None, timestamp=None, uuid=None, disable_geoip=None):
def group(self, distinct_id=None, group_id=None, traits=None, context=None,
timestamp=None, message_id=None):
traits = traits or {}
context = context or {}
require("previous_id", previous_id, ID_TYPES)
require("distinct_id", distinct_id, ID_TYPES)
require('distinct_id', distinct_id, ID_TYPES)
require('group_id', group_id, ID_TYPES)
require('traits', traits, dict)
msg = {
"properties": {
"distinct_id": previous_id,
"alias": distinct_id,
},
"timestamp": timestamp,
"context": context,
"event": "$create_alias",
"distinct_id": previous_id,
'timestamp': timestamp,
'groupId': group_id,
'context': context,
'distinct_id': distinct_id,
'traits': traits,
'type': 'group',
'messageId': message_id,
}
return self._enqueue(msg, disable_geoip)
return self._enqueue(msg)
def page(
self, distinct_id=None, url=None, properties=None, context=None, timestamp=None, uuid=None, disable_geoip=None
):
def page(self, distinct_id=None, category=None, name=None, properties=None,
context=None, timestamp=None, message_id=None):
properties = properties or {}
context = context or {}
require("distinct_id", distinct_id, ID_TYPES)
require("properties", properties, dict)
require('distinct_id', distinct_id, ID_TYPES)
require('properties', properties, dict)
require("url", url, string_types)
properties["$current_url"] = url
if name:
require('name', name, string_types)
if category:
require('category', category, string_types)
msg = {
"event": "$pageview",
"properties": properties,
"timestamp": timestamp,
"context": context,
"distinct_id": distinct_id,
"uuid": uuid,
'properties': properties,
'timestamp': timestamp,
'category': category,
'context': context,
'distinct_id': distinct_id,
'type': 'page',
'name': name,
'messageId': message_id,
}
return self._enqueue(msg, disable_geoip)
return self._enqueue(msg)
def _enqueue(self, msg, disable_geoip):
def screen(self, distinct_id=None, category=None, name=None, properties=None,
context=None, timestamp=None, message_id=None):
properties = properties or {}
context = context or {}
require('distinct_id', distinct_id, ID_TYPES)
require('properties', properties, dict)
if name:
require('name', name, string_types)
if category:
require('category', category, string_types)
msg = {
'properties': properties,
'timestamp': timestamp,
'category': category,
'context': context,
'distinct_id': distinct_id,
'type': 'screen',
'name': name,
'messageId': message_id,
}
return self._enqueue(msg)
def _enqueue(self, msg):
"""Push a new `msg` onto the queue, return `(success, msg)`"""
if self.disabled:
return False, "disabled"
timestamp = msg["timestamp"]
timestamp = msg['timestamp']
if timestamp is None:
timestamp = datetime.utcnow().replace(tzinfo=tzutc())
message_id = msg.get('messageId')
if message_id is None:
message_id = uuid4()
require("timestamp", timestamp, datetime)
require("context", msg["context"], dict)
require('type', msg['type'], string_types)
require('timestamp', timestamp, datetime)
require('context', msg['context'], dict)
# add common
timestamp = guess_timezone(timestamp)
msg["timestamp"] = timestamp.isoformat()
msg['timestamp'] = timestamp.isoformat()
msg['messageId'] = stringify_id(message_id)
if not msg.get('properties'):
msg['properties'] = {}
msg['properties']['$lib'] = 'posthog-python'
msg['properties']['$lib_version'] = VERSION
# only send if "uuid" is truthy
if "uuid" in msg:
uuid = msg.pop("uuid")
if uuid:
msg["uuid"] = stringify_id(uuid)
if not msg.get("properties"):
msg["properties"] = {}
msg["properties"]["$lib"] = "posthog-python"
msg["properties"]["$lib_version"] = VERSION
if disable_geoip is None:
disable_geoip = self.disable_geoip
if disable_geoip:
msg["properties"]["$geoip_disable"] = True
msg["distinct_id"] = stringify_id(msg.get("distinct_id", None))
msg['distinct_id'] = stringify_id(msg.get('distinct_id', None))
msg = clean(msg)
self.log.debug("queueing: %s", msg)
self.log.debug('queueing: %s', msg)
# if send is False, return msg as if it was successfully queued
if not self.send:
return True, msg
if self.sync_mode:
self.log.debug("enqueued with blocking %s.", msg["event"])
batch_post(self.api_key, self.host, gzip=self.gzip, timeout=self.timeout, batch=[msg])
self.log.debug('enqueued with blocking %s.', msg['type'])
post(self.api_key, self.host, gzip=self.gzip,
timeout=self.timeout, batch=[msg])
return True, msg
try:
self.queue.put(msg, block=False)
self.log.debug("enqueued %s.", msg["event"])
self.log.debug('enqueued %s.', msg['type'])
return True, msg
except queue.Full:
self.log.warning("analytics-python queue is full")
self.log.warning('analytics-python queue is full')
return False, msg
def flush(self):
@@ -392,7 +253,7 @@ class Client(object):
size = queue.qsize()
queue.join()
# Note that this message may not be precise, because of threading.
self.log.debug("successfully flushed about %s items.", size)
self.log.debug('successfully flushed about %s items.', size)
def join(self):
"""Ends the consumer thread once the queue is empty.
@@ -406,390 +267,16 @@ class Client(object):
# consumer thread has not started
pass
if self.poller:
self.poller.stop()
def shutdown(self):
"""Flush all messages and cleanly shutdown the client"""
self.flush()
self.join()
def _load_feature_flags(self):
try:
response = get(
self.personal_api_key,
f"/api/feature_flag/local_evaluation/?token={self.api_key}&send_cohorts",
self.host,
timeout=10,
)
self.feature_flags = response["flags"] or []
self.feature_flags_by_key = {
flag["key"]: flag for flag in self.feature_flags if flag.get("key") is not None
}
self.group_type_mapping = response["group_type_mapping"] or {}
self.cohorts = response["cohorts"] or {}
except APIError as e:
if e.status == 401:
self.log.error(
"[FEATURE FLAGS] Error loading feature flags: To use feature flags, please set a valid personal_api_key. More information: https://posthog.com/docs/api/overview"
)
if self.debug:
raise APIError(
status=401,
message="You are using a write-only key with feature flags. "
"To use feature flags, please set a personal_api_key "
"More information: https://posthog.com/docs/api/overview",
)
else:
self.log.error(f"[FEATURE FLAGS] Error loading feature flags: {e}")
except Exception as e:
self.log.warning(
"[FEATURE FLAGS] Fetching feature flags failed with following error. We will retry in %s seconds."
% self.poll_interval
)
self.log.warning(e)
self._last_feature_flag_poll = datetime.utcnow().replace(tzinfo=tzutc())
def load_feature_flags(self):
if not self.personal_api_key:
self.log.warning("[FEATURE FLAGS] You have to specify a personal_api_key to use feature flags.")
self.feature_flags = []
return
self._load_feature_flags()
if not (self.poller and self.poller.is_alive()):
self.poller = Poller(interval=timedelta(seconds=self.poll_interval), execute=self._load_feature_flags)
self.poller.start()
def _compute_flag_locally(
self,
feature_flag,
distinct_id,
*,
groups={},
person_properties={},
group_properties={},
warn_on_unknown_groups=True,
):
if feature_flag.get("ensure_experience_continuity", False):
raise InconclusiveMatchError("Flag has experience continuity enabled")
if not feature_flag.get("active"):
return False
flag_filters = feature_flag.get("filters") or {}
aggregation_group_type_index = flag_filters.get("aggregation_group_type_index")
if aggregation_group_type_index is not None:
group_name = self.group_type_mapping.get(str(aggregation_group_type_index))
if not group_name:
self.log.warning(
f"[FEATURE FLAGS] Unknown group type index {aggregation_group_type_index} for feature flag {feature_flag['key']}"
)
# failover to `/decide/`
raise InconclusiveMatchError("Flag has unknown group type index")
if group_name not in groups:
# Group flags are never enabled in `groups` aren't passed in
# don't failover to `/decide/`, since response will be the same
if warn_on_unknown_groups:
self.log.warning(
f"[FEATURE FLAGS] Can't compute group feature flag: {feature_flag['key']} without group names passed in"
)
else:
self.log.debug(
f"[FEATURE FLAGS] Can't compute group feature flag: {feature_flag['key']} without group names passed in"
)
return False
focused_group_properties = group_properties[group_name]
return match_feature_flag_properties(feature_flag, groups[group_name], focused_group_properties)
else:
return match_feature_flag_properties(feature_flag, distinct_id, person_properties, self.cohorts)
def feature_enabled(
self,
key,
distinct_id,
*,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
send_feature_flag_events=True,
disable_geoip=None,
):
response = self.get_feature_flag(
key,
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
send_feature_flag_events=send_feature_flag_events,
disable_geoip=disable_geoip,
)
if response is None:
return None
return bool(response)
def get_feature_flag(
self,
key,
distinct_id,
*,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
send_feature_flag_events=True,
disable_geoip=None,
):
require("key", key, string_types)
require("distinct_id", distinct_id, ID_TYPES)
require("groups", groups, dict)
if self.disabled:
return None
person_properties, group_properties = self._add_local_person_and_group_properties(
distinct_id, groups, person_properties, group_properties
)
if self.feature_flags is None and self.personal_api_key:
self.load_feature_flags()
response = None
# If loading in previous line failed
if self.feature_flags:
for flag in self.feature_flags:
if flag["key"] == key:
try:
response = self._compute_flag_locally(
flag,
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
)
self.log.debug(f"Successfully computed flag locally: {key} -> {response}")
except InconclusiveMatchError as e:
self.log.debug(f"Failed to compute flag {key} locally: {e}")
continue
except Exception as e:
self.log.exception(f"[FEATURE FLAGS] Error while computing variant locally: {e}")
continue
flag_was_locally_evaluated = response is not None
if not flag_was_locally_evaluated and not only_evaluate_locally:
try:
feature_flags = self.get_feature_variants(
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
disable_geoip=disable_geoip,
)
response = feature_flags.get(key)
if response is None:
response = False
self.log.debug(f"Successfully computed flag remotely: #{key} -> #{response}")
except Exception as e:
self.log.exception(f"[FEATURE FLAGS] Unable to get flag remotely: {e}")
feature_flag_reported_key = f"{key}_{str(response)}"
if (
feature_flag_reported_key not in self.distinct_ids_feature_flags_reported[distinct_id]
and send_feature_flag_events # noqa: W503
):
self.capture(
distinct_id,
"$feature_flag_called",
{
"$feature_flag": key,
"$feature_flag_response": response,
"locally_evaluated": flag_was_locally_evaluated,
f"$feature/{key}": response,
},
groups=groups,
disable_geoip=disable_geoip,
)
self.distinct_ids_feature_flags_reported[distinct_id].add(feature_flag_reported_key)
return response
def get_feature_flag_payload(
self,
key,
distinct_id,
*,
match_value=None,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
send_feature_flag_events=True,
disable_geoip=None,
):
if self.disabled:
return None
if match_value is None:
match_value = self.get_feature_flag(
key,
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
send_feature_flag_events=send_feature_flag_events,
only_evaluate_locally=True,
disable_geoip=disable_geoip,
)
response = None
if match_value is not None:
response = self._compute_payload_locally(key, match_value)
if response is None and not only_evaluate_locally:
decide_payloads = self.get_feature_payloads(
distinct_id, groups, person_properties, group_properties, disable_geoip
)
response = decide_payloads.get(str(key).lower(), None)
return response
def _compute_payload_locally(self, key, match_value):
payload = None
if self.feature_flags_by_key is None:
return payload
flag_definition = self.feature_flags_by_key.get(key) or {}
flag_filters = flag_definition.get("filters") or {}
flag_payloads = flag_filters.get("payloads") or {}
payload = flag_payloads.get(str(match_value).lower(), None)
return payload
def get_all_flags(
self,
distinct_id,
*,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
disable_geoip=None,
):
flags = self.get_all_flags_and_payloads(
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
only_evaluate_locally=only_evaluate_locally,
disable_geoip=disable_geoip,
)
return flags["featureFlags"]
def get_all_flags_and_payloads(
self,
distinct_id,
*,
groups={},
person_properties={},
group_properties={},
only_evaluate_locally=False,
disable_geoip=None,
):
if self.disabled:
return {"featureFlags": None, "featureFlagPayloads": None}
person_properties, group_properties = self._add_local_person_and_group_properties(
distinct_id, groups, person_properties, group_properties
)
flags, payloads, fallback_to_decide = self._get_all_flags_and_payloads_locally(
distinct_id, groups=groups, person_properties=person_properties, group_properties=group_properties
)
response = {"featureFlags": flags, "featureFlagPayloads": payloads}
if fallback_to_decide and not only_evaluate_locally:
try:
flags_and_payloads = self.get_decide(
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
disable_geoip=disable_geoip,
)
response = flags_and_payloads
except Exception as e:
self.log.exception(f"[FEATURE FLAGS] Unable to get feature flags and payloads: {e}")
return response
def _get_all_flags_and_payloads_locally(
self, distinct_id, *, groups={}, person_properties={}, group_properties={}, warn_on_unknown_groups=False
):
require("distinct_id", distinct_id, ID_TYPES)
require("groups", groups, dict)
if self.feature_flags is None and self.personal_api_key:
self.load_feature_flags()
flags = {}
payloads = {}
fallback_to_decide = False
# If loading in previous line failed
if self.feature_flags:
for flag in self.feature_flags:
try:
flags[flag["key"]] = self._compute_flag_locally(
flag,
distinct_id,
groups=groups,
person_properties=person_properties,
group_properties=group_properties,
warn_on_unknown_groups=warn_on_unknown_groups,
)
matched_payload = self._compute_payload_locally(flag["key"], flags[flag["key"]])
if matched_payload:
payloads[flag["key"]] = matched_payload
except InconclusiveMatchError:
# No need to log this, since it's just telling us to fall back to `/decide`
fallback_to_decide = True
except Exception as e:
self.log.exception(f"[FEATURE FLAGS] Error while computing variant and payload: {e}")
fallback_to_decide = True
else:
fallback_to_decide = True
return flags, payloads, fallback_to_decide
def feature_flag_definitions(self):
return self.feature_flags
def _add_local_person_and_group_properties(self, distinct_id, groups, person_properties, group_properties):
all_person_properties = {"distinct_id": distinct_id, **(person_properties or {})}
all_group_properties = {}
if groups:
for group_name in groups:
all_group_properties[group_name] = {
"$group_key": groups[group_name],
**(group_properties.get(group_name) or {}),
}
return all_person_properties, all_group_properties
def require(name, field, data_type):
"""Require that the named `field` has the right `data_type`"""
if not isinstance(field, data_type):
msg = "{0} must have {1}, got: {2}".format(name, data_type, field)
msg = '{0} must have {1}, got: {2}'.format(name, data_type, field)
raise AssertionError(msg)
+30 -35
View File
@@ -1,42 +1,30 @@
import json
import logging
from threading import Thread
import backoff
import monotonic
import backoff
import json
from posthog.request import APIError, DatetimeSerializer, batch_post
from posthog.request import post, APIError, DatetimeSerializer
try:
from queue import Empty
except ImportError:
from Queue import Empty
MAX_MSG_SIZE = 32 << 10
MAX_MSG_SIZE = 900 * 1024 # 900KiB per event
# The maximum request body size is currently 20MiB, let's be conservative
# in case we want to lower it in the future.
BATCH_SIZE_LIMIT = 5 * 1024 * 1024
# Our servers only accept batches less than 500KB. Here limit is set slightly
# lower to leave space for extra data that will be added later, eg. "sentAt".
BATCH_SIZE_LIMIT = 475000
class Consumer(Thread):
"""Consumes the messages from the client's queue."""
log = logging.getLogger('posthog')
log = logging.getLogger("posthog")
def __init__(
self,
queue,
api_key,
flush_at=100,
host=None,
on_error=None,
flush_interval=0.5,
gzip=False,
retries=10,
timeout=15,
):
def __init__(self, queue, api_key, flush_at=100, host=None,
on_error=None, flush_interval=0.5, gzip=False, retries=10,
timeout=15):
"""Create a consumer thread."""
Thread.__init__(self)
# Make consumer a daemon thread so that it doesn't block program exit
@@ -58,11 +46,11 @@ class Consumer(Thread):
def run(self):
"""Runs the consumer."""
self.log.debug("consumer is running...")
self.log.debug('consumer is running...')
while self.running:
self.upload()
self.log.debug("consumer exited.")
self.log.debug('consumer exited.')
def pause(self):
"""Pause the consumer."""
@@ -79,7 +67,7 @@ class Consumer(Thread):
self.request(batch)
success = True
except Exception as e:
self.log.error("error uploading: %s", e)
self.log.error('error uploading: %s', e)
success = False
if self.on_error:
self.on_error(e, batch)
@@ -102,15 +90,19 @@ class Consumer(Thread):
if elapsed >= self.flush_interval:
break
try:
item = queue.get(block=True, timeout=self.flush_interval - elapsed)
item_size = len(json.dumps(item, cls=DatetimeSerializer).encode())
item = queue.get(
block=True, timeout=self.flush_interval - elapsed)
item_size = len(json.dumps(
item, cls=DatetimeSerializer).encode())
if item_size > MAX_MSG_SIZE:
self.log.error("Item exceeds 900kib limit, dropping. (%s)", str(item))
self.log.error(
'Item exceeds 32kb limit, dropping. (%s)', str(item))
continue
items.append(item)
total_size += item_size
if total_size >= BATCH_SIZE_LIMIT:
self.log.debug("hit batch size limit (size: %d)", total_size)
self.log.debug(
'hit batch size limit (size: %d)', total_size)
break
except Empty:
break
@@ -118,22 +110,25 @@ class Consumer(Thread):
return items
def request(self, batch):
"""Attempt to upload the batch and retry before raising an error"""
"""Attempt to upload the batch and retry before raising an error """
def fatal_exception(exc):
if isinstance(exc, APIError):
# retry on server errors and client errors
# with 429 status code (rate limited),
# don't retry on other client errors
if exc.status == "N/A":
return False
return (400 <= exc.status < 500) and exc.status != 429
else:
# retry on all other errors (eg. network)
return False
@backoff.on_exception(backoff.expo, Exception, max_tries=self.retries + 1, giveup=fatal_exception)
@backoff.on_exception(
backoff.expo,
Exception,
max_tries=self.retries + 1,
giveup=fatal_exception)
def send_request():
batch_post(self.api_key, self.host, gzip=self.gzip, timeout=self.timeout, batch=batch)
post(self.api_key, self.host, gzip=self.gzip,
timeout=self.timeout, batch=batch)
send_request()
-331
View File
@@ -1,331 +0,0 @@
import datetime
import hashlib
import logging
import re
from typing import Optional
from dateutil import parser
from dateutil.relativedelta import relativedelta
from posthog.utils import convert_to_datetime_aware, is_valid_regex
__LONG_SCALE__ = float(0xFFFFFFFFFFFFFFF)
log = logging.getLogger("posthog")
class InconclusiveMatchError(Exception):
pass
# This function takes a distinct_id and a feature flag key and returns a float between 0 and 1.
# Given the same distinct_id and key, it'll always return the same float. These floats are
# uniformly distributed between 0 and 1, so if we want to show this feature to 20% of traffic
# we can do _hash(key, distinct_id) < 0.2
def _hash(key, distinct_id, salt=""):
hash_key = f"{key}.{distinct_id}{salt}"
hash_val = int(hashlib.sha1(hash_key.encode("utf-8")).hexdigest()[:15], 16)
return hash_val / __LONG_SCALE__
def get_matching_variant(flag, distinct_id):
hash_value = _hash(flag["key"], distinct_id, salt="variant")
for variant in variant_lookup_table(flag):
if hash_value >= variant["value_min"] and hash_value < variant["value_max"]:
return variant["key"]
return None
def variant_lookup_table(feature_flag):
lookup_table = []
value_min = 0
multivariates = ((feature_flag.get("filters") or {}).get("multivariate") or {}).get("variants") or []
for variant in multivariates:
value_max = value_min + variant["rollout_percentage"] / 100
lookup_table.append({"value_min": value_min, "value_max": value_max, "key": variant["key"]})
value_min = value_max
return lookup_table
def match_feature_flag_properties(flag, distinct_id, properties, cohort_properties=None):
flag_conditions = (flag.get("filters") or {}).get("groups") or []
is_inconclusive = False
cohort_properties = cohort_properties or {}
# Stable sort conditions with variant overrides to the top. This ensures that if overrides are present, they are
# evaluated first, and the variant override is applied to the first matching condition.
sorted_flag_conditions = sorted(
flag_conditions,
key=lambda condition: 0 if condition.get("variant") else 1,
)
for condition in sorted_flag_conditions:
try:
# if any one condition resolves to True, we can shortcircuit and return
# the matching variant
if is_condition_match(flag, distinct_id, condition, properties, cohort_properties):
variant_override = condition.get("variant")
# Some filters can be explicitly set to null, which require accessing variants like so
flag_variants = ((flag.get("filters") or {}).get("multivariate") or {}).get("variants") or []
if variant_override and variant_override in [variant["key"] for variant in flag_variants]:
variant = variant_override
else:
variant = get_matching_variant(flag, distinct_id)
return variant or True
except InconclusiveMatchError:
is_inconclusive = True
if is_inconclusive:
raise InconclusiveMatchError("Can't determine if feature flag is enabled or not with given properties")
# We can only return False when either all conditions are False, or
# no condition was inconclusive.
return False
def is_condition_match(feature_flag, distinct_id, condition, properties, cohort_properties):
rollout_percentage = condition.get("rollout_percentage")
if len(condition.get("properties") or []) > 0:
for prop in condition.get("properties"):
property_type = prop.get("type")
if property_type == "cohort":
matches = match_cohort(prop, properties, cohort_properties)
else:
matches = match_property(prop, properties)
if not matches:
return False
if rollout_percentage is None:
return True
if rollout_percentage is not None and _hash(feature_flag["key"], distinct_id) > (rollout_percentage / 100):
return False
return True
def match_property(property, property_values) -> bool:
# only looks for matches where key exists in override_property_values
# doesn't support operator is_not_set
key = property.get("key")
operator = property.get("operator") or "exact"
value = property.get("value")
if key not in property_values:
raise InconclusiveMatchError("can't match properties without a given property value")
if operator == "is_not_set":
raise InconclusiveMatchError("can't match properties with operator is_not_set")
override_value = property_values[key]
if operator in ("exact", "is_not"):
def compute_exact_match(value, override_value):
if isinstance(value, list):
return str(override_value).lower() in [str(val).lower() for val in value]
return str(value).lower() == str(override_value).lower()
if operator == "exact":
return compute_exact_match(value, override_value)
else:
return not compute_exact_match(value, override_value)
if operator == "is_set":
return key in property_values
if operator == "icontains":
return str(value).lower() in str(override_value).lower()
if operator == "not_icontains":
return str(value).lower() not in str(override_value).lower()
if operator == "regex":
return is_valid_regex(str(value)) and re.compile(str(value)).search(str(override_value)) is not None
if operator == "not_regex":
return is_valid_regex(str(value)) and re.compile(str(value)).search(str(override_value)) is None
if operator in ("gt", "gte", "lt", "lte"):
# :TRICKY: We adjust comparison based on the override value passed in,
# to make sure we handle both numeric and string comparisons appropriately.
def compare(lhs, rhs, operator):
if operator == "gt":
return lhs > rhs
elif operator == "gte":
return lhs >= rhs
elif operator == "lt":
return lhs < rhs
elif operator == "lte":
return lhs <= rhs
else:
raise ValueError(f"Invalid operator: {operator}")
parsed_value = None
try:
parsed_value = float(value) # type: ignore
except Exception:
pass
if parsed_value is not None and override_value is not None:
if isinstance(override_value, str):
return compare(override_value, str(value), operator)
else:
return compare(override_value, parsed_value, operator)
else:
return compare(str(override_value), str(value), operator)
if operator in ["is_date_before", "is_date_after"]:
try:
parsed_date = relative_date_parse_for_feature_flag_matching(str(value))
if not parsed_date:
parsed_date = parser.parse(str(value))
parsed_date = convert_to_datetime_aware(parsed_date)
except Exception as e:
raise InconclusiveMatchError("The date set on the flag is not a valid format") from e
if not parsed_date:
raise InconclusiveMatchError("The date set on the flag is not a valid format")
if isinstance(override_value, datetime.datetime):
override_date = convert_to_datetime_aware(override_value)
if operator == "is_date_before":
return override_date < parsed_date
else:
return override_date > parsed_date
elif isinstance(override_value, datetime.date):
if operator == "is_date_before":
return override_value < parsed_date.date()
else:
return override_value > parsed_date.date()
elif isinstance(override_value, str):
try:
override_date = parser.parse(override_value)
override_date = convert_to_datetime_aware(override_date)
if operator == "is_date_before":
return override_date < parsed_date
else:
return override_date > parsed_date
except Exception:
raise InconclusiveMatchError("The date provided is not a valid format")
else:
raise InconclusiveMatchError("The date provided must be a string or date object")
# if we get here, we don't know how to handle the operator
raise InconclusiveMatchError(f"Unknown operator {operator}")
def match_cohort(property, property_values, cohort_properties) -> bool:
# Cohort properties are in the form of property groups like this:
# {
# "cohort_id": {
# "type": "AND|OR",
# "values": [{
# "key": "property_name", "value": "property_value"
# }]
# }
# }
cohort_id = str(property.get("value"))
if cohort_id not in cohort_properties:
raise InconclusiveMatchError("can't match cohort without a given cohort property value")
property_group = cohort_properties[cohort_id]
return match_property_group(property_group, property_values, cohort_properties)
def match_property_group(property_group, property_values, cohort_properties) -> bool:
if not property_group:
return True
property_group_type = property_group.get("type")
properties = property_group.get("values")
if not properties or len(properties) == 0:
# empty groups are no-ops, always match
return True
error_matching_locally = False
if "values" in properties[0]:
# a nested property group
for prop in properties:
try:
matches = match_property_group(prop, property_values, cohort_properties)
if property_group_type == "AND":
if not matches:
return False
else:
# OR group
if matches:
return True
except InconclusiveMatchError as e:
log.debug(f"Failed to compute property {prop} locally: {e}")
error_matching_locally = True
if error_matching_locally:
raise InconclusiveMatchError("Can't match cohort without a given cohort property value")
# if we get here, all matched in AND case, or none matched in OR case
return property_group_type == "AND"
else:
for prop in properties:
try:
if prop.get("type") == "cohort":
matches = match_cohort(prop, property_values, cohort_properties)
else:
matches = match_property(prop, property_values)
negation = prop.get("negation", False)
if property_group_type == "AND":
# if negated property, do the inverse
if not matches and not negation:
return False
if matches and negation:
return False
else:
# OR group
if matches and not negation:
return True
if not matches and negation:
return True
except InconclusiveMatchError as e:
log.debug(f"Failed to compute property {prop} locally: {e}")
error_matching_locally = True
if error_matching_locally:
raise InconclusiveMatchError("can't match cohort without a given cohort property value")
# if we get here, all matched in AND case, or none matched in OR case
return property_group_type == "AND"
def relative_date_parse_for_feature_flag_matching(value: str) -> Optional[datetime.datetime]:
regex = r"^-?(?P<number>[0-9]+)(?P<interval>[a-z])$"
match = re.search(regex, value)
parsed_dt = datetime.datetime.now(datetime.timezone.utc)
if match:
number = int(match.group("number"))
if number >= 10_000:
# Guard against overflow, disallow numbers greater than 10_000
return None
interval = match.group("interval")
if interval == "h":
parsed_dt = parsed_dt - relativedelta(hours=number)
elif interval == "d":
parsed_dt = parsed_dt - relativedelta(days=number)
elif interval == "w":
parsed_dt = parsed_dt - relativedelta(weeks=number)
elif interval == "m":
parsed_dt = parsed_dt - relativedelta(months=number)
elif interval == "y":
parsed_dt = parsed_dt - relativedelta(years=number)
else:
return None
return parsed_dt
else:
return None
-20
View File
@@ -1,20 +0,0 @@
import threading
class Poller(threading.Thread):
def __init__(self, interval, execute, *args, **kwargs):
threading.Thread.__init__(self)
self.daemon = True # Make daemon to not interfere with program exit
self.stopped = threading.Event()
self.interval = interval
self.execute = execute
self.args = args
self.kwargs = kwargs
def stop(self):
self.stopped.set()
self.join()
def run(self):
while not self.stopped.wait(self.interval.total_seconds()):
self.execute(*self.args, **self.kwargs)
+34 -63
View File
@@ -1,98 +1,69 @@
import json
import logging
from datetime import date, datetime
from gzip import GzipFile
from io import BytesIO
from typing import Any, Optional, Union
import requests
from dateutil.tz import tzutc
import logging
import json
from gzip import GzipFile
from requests.auth import HTTPBasicAuth
from requests import sessions
from io import BytesIO
from posthog.utils import remove_trailing_slash
from posthog.version import VERSION
from posthog.utils import remove_trailing_slash
_session = requests.sessions.Session()
DEFAULT_HOST = "https://app.posthog.com"
USER_AGENT = "posthog-python/" + VERSION
_session = sessions.Session()
def post(
api_key: str, host: Optional[str] = None, path=None, gzip: bool = False, timeout: int = 15, **kwargs
) -> requests.Response:
def post(api_key, host=None, gzip=False, timeout=15, **kwargs):
"""Post the `kwargs` to the API"""
log = logging.getLogger("posthog")
log = logging.getLogger('posthog')
body = kwargs
body["sentAt"] = datetime.utcnow().replace(tzinfo=tzutc()).isoformat()
url = remove_trailing_slash(host or DEFAULT_HOST) + path
body["api_key"] = api_key
url = remove_trailing_slash(host or 'https://t.posthog.com') + '/batch/'
body['api_key'] = api_key
data = json.dumps(body, cls=DatetimeSerializer)
log.debug("making request: %s", data)
headers = {"Content-Type": "application/json", "User-Agent": USER_AGENT}
log.debug('making request: %s', data)
headers = {
'Content-Type': 'application/json',
'User-Agent': 'analytics-python/' + VERSION
}
if gzip:
headers["Content-Encoding"] = "gzip"
headers['Content-Encoding'] = 'gzip'
buf = BytesIO()
with GzipFile(fileobj=buf, mode="w") as gz:
with GzipFile(fileobj=buf, mode='w') as gz:
# 'data' was produced by json.dumps(),
# whose default encoding is utf-8.
gz.write(data.encode("utf-8"))
gz.write(data.encode('utf-8'))
data = buf.getvalue()
res = _session.post(url, data=data, headers=headers, timeout=timeout)
res = _session.post(url, data=data,
headers=headers, timeout=timeout)
if res.status_code == 200:
log.debug("data uploaded successfully")
log.debug('data uploaded successfully')
return res
return res
def _process_response(
res: requests.Response, success_message: str, *, return_json: bool = True
) -> Union[requests.Response, Any]:
log = logging.getLogger("posthog")
if res.status_code == 200:
log.debug(success_message)
return res.json() if return_json else res
try:
payload = res.json()
log.debug("received response: %s", payload)
raise APIError(res.status_code, payload["detail"])
except (KeyError, ValueError):
raise APIError(res.status_code, res.text)
def decide(api_key: str, host: Optional[str] = None, gzip: bool = False, timeout: int = 15, **kwargs) -> Any:
"""Post the `kwargs to the decide API endpoint"""
res = post(api_key, host, "/decide/?v=3", gzip, timeout, **kwargs)
return _process_response(res, success_message="Feature flags decided successfully")
def batch_post(
api_key: str, host: Optional[str] = None, gzip: bool = False, timeout: int = 15, **kwargs
) -> requests.Response:
"""Post the `kwargs` to the batch API endpoint for events"""
res = post(api_key, host, "/batch/", gzip, timeout, **kwargs)
return _process_response(res, success_message="data uploaded successfully", return_json=False)
def get(api_key: str, url: str, host: Optional[str] = None, timeout: Optional[int] = None) -> requests.Response:
url = remove_trailing_slash(host or DEFAULT_HOST) + url
res = requests.get(url, headers={"Authorization": "Bearer %s" % api_key, "User-Agent": USER_AGENT}, timeout=timeout)
return _process_response(res, success_message=f"GET {url} completed successfully")
log.debug('received response: %s', payload)
raise APIError(res.status_code, payload['code'], payload['message'])
except ValueError:
raise APIError(res.status_code, 'unknown', res.text)
class APIError(Exception):
def __init__(self, status: Union[int, str], message: str):
def __init__(self, status, code, message):
self.message = message
self.status = status
self.code = code
def __str__(self):
msg = "[PostHog] {0} ({1})"
return msg.format(self.message, self.status)
msg = "[PostHog] {0}: {1} ({2})"
return msg.format(self.code, self.message, self.status)
class DatetimeSerializer(json.JSONEncoder):
def default(self, obj: Any):
def default(self, obj):
if isinstance(obj, (date, datetime)):
return obj.isoformat()
-1
View File
@@ -1 +0,0 @@
POSTHOG_ID_TAG = "posthog_distinct_id"
-28
View File
@@ -1,28 +0,0 @@
from django.conf import settings
from sentry_sdk import configure_scope
from posthog.sentry import POSTHOG_ID_TAG
GET_DISTINCT_ID = getattr(settings, "POSTHOG_DJANGO", {}).get("distinct_id")
def get_distinct_id(request):
if not GET_DISTINCT_ID:
return None
try:
return GET_DISTINCT_ID(request)
except: # noqa: E722
return None
class PosthogDistinctIdMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
with configure_scope() as scope:
distinct_id = get_distinct_id(request)
if distinct_id:
scope.set_tag(POSTHOG_ID_TAG, distinct_id)
response = self.get_response(request)
return response
-52
View File
@@ -1,52 +0,0 @@
from sentry_sdk._types import MYPY
from sentry_sdk.hub import Hub
from sentry_sdk.integrations import Integration
from sentry_sdk.scope import add_global_event_processor
from sentry_sdk.utils import Dsn
import posthog
from posthog.request import DEFAULT_HOST
from posthog.sentry import POSTHOG_ID_TAG
if MYPY:
from typing import Optional # noqa: F401
from sentry_sdk._types import Event, Hint # noqa: F401
class PostHogIntegration(Integration):
identifier = "posthog-python"
organization = None # The Sentry organization, used to send a direct link from PostHog to Sentry
project_id = None # The Sentry project id, used to send a direct link from PostHog to Sentry
prefix = "https://sentry.io/organizations/" # URL of a hosted sentry instance (default: https://sentry.io/organizations/)
@staticmethod
def setup_once():
@add_global_event_processor
def processor(event, hint):
# type: (Event, Optional[Hint]) -> Optional[Event]
if Hub.current.get_integration(PostHogIntegration) is not None:
if event.get("level") != "error":
return event
if event.get("tags", {}).get(POSTHOG_ID_TAG):
posthog_distinct_id = event["tags"][POSTHOG_ID_TAG]
event["tags"]["PostHog URL"] = f"{posthog.host or DEFAULT_HOST}/person/{posthog_distinct_id}"
properties = {
"$sentry_event_id": event["event_id"],
"$sentry_exception": event["exception"],
}
if PostHogIntegration.organization:
project_id = PostHogIntegration.project_id or (
not not Hub.current.client.dsn and Dsn(Hub.current.client.dsn).project_id
)
if project_id:
properties["$sentry_url"] = (
f"{PostHogIntegration.prefix}{PostHogIntegration.organization}/issues/?project={project_id}&query={event['event_id']}"
)
posthog.capture(posthog_distinct_id, "$exception", properties)
return event
+4 -4
View File
@@ -1,12 +1,12 @@
import logging
import pkgutil
import sys
import unittest
import pkgutil
import logging
import sys
def all_names():
for _, modname, _ in pkgutil.iter_modules(__path__):
yield "posthog.test." + modname
yield 'analytics.test.' + modname
def all():
+314
View File
@@ -0,0 +1,314 @@
from datetime import date, datetime
import unittest
import six
import mock
import time
from posthog.version import VERSION
from posthog.client import Client
class TestClient(unittest.TestCase):
def fail(self, e, batch):
"""Mark the failure handler"""
self.failed = True
def setUp(self):
self.failed = False
self.client = Client('testsecret', on_error=self.fail)
def test_requires_api_key(self):
self.assertRaises(AssertionError, Client)
def test_empty_flush(self):
self.client.flush()
def test_basic_track(self):
client = self.client
success, msg = client.track('distinct_id', 'python test event')
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['event'], 'python test event')
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertTrue(isinstance(msg['messageId'], str))
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['properties'], {})
self.assertEqual(msg['type'], 'track')
def test_stringifies_distinct_id(self):
# A large number that loses precision in node:
# node -e "console.log(157963456373623802 + 1)" > 157963456373623800
client = self.client
success, msg = client.track(
distinct_id=157963456373623802, event='python test event')
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['distinct_id'], '157963456373623802')
def test_advanced_track(self):
client = self.client
success, msg = client.track(
'distinct_id', 'python test event', {'property': 'value'},
{'ip': '192.168.0.1'}, datetime(2014, 9, 3),
'messageId')
self.assertTrue(success)
self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00')
self.assertEqual(msg['properties'], {'property': 'value'})
self.assertEqual(msg['context']['ip'], '192.168.0.1')
self.assertEqual(msg['event'], 'python test event')
self.assertEqual(msg['properties']['$lib'], 'posthog-python')
self.assertEqual(msg['properties']['$lib_version'], VERSION)
self.assertEqual(msg['messageId'], 'messageId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'track')
def test_basic_identify(self):
client = self.client
success, msg = client.identify('distinct_id', {'trait': 'value'})
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['traits'], {'trait': 'value'})
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertTrue(isinstance(msg['messageId'], str))
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'identify')
def test_advanced_identify(self):
client = self.client
success, msg = client.identify(
'distinct_id', {'trait': 'value'}, {'ip': '192.168.0.1'},
datetime(2014, 9, 3), 'messageId')
self.assertTrue(success)
self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00')
self.assertEqual(msg['context']['ip'], '192.168.0.1')
self.assertEqual(msg['traits'], {'trait': 'value'})
self.assertEqual(msg['context']['library'], {
'name': 'analytics-python',
'version': VERSION
})
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertEqual(msg['messageId'], 'messageId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'identify')
def test_basic_group(self):
client = self.client
success, msg = client.group('distinct_id', 'groupId')
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['groupId'], 'groupId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'group')
def test_advanced_group(self):
client = self.client
success, msg = client.group(
'distinct_id', 'groupId', {'trait': 'value'}, {'ip': '192.168.0.1'},
datetime(2014, 9, 3), 'messageId')
self.assertTrue(success)
self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00')
self.assertEqual(msg['context']['ip'], '192.168.0.1')
self.assertEqual(msg['traits'], {'trait': 'value'})
self.assertEqual(msg['context']['library'], {
'name': 'analytics-python',
'version': VERSION
})
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertEqual(msg['messageId'], 'messageId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'group')
def test_basic_alias(self):
client = self.client
success, msg = client.alias('previousId', 'distinct_id')
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['previousId'], 'previousId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
def test_basic_page(self):
client = self.client
success, msg = client.page('distinct_id', name='name')
self.assertFalse(self.failed)
client.flush()
self.assertTrue(success)
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'page')
self.assertEqual(msg['name'], 'name')
def test_advanced_page(self):
client = self.client
success, msg = client.page(
'distinct_id', 'category', 'name', {'property': 'value'},
{'ip': '192.168.0.1'}, datetime(2014, 9, 3), 'messageId')
self.assertTrue(success)
self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00')
self.assertEqual(msg['context']['ip'], '192.168.0.1')
self.assertEqual(msg['properties'], {'property': 'value'})
self.assertEqual(msg['context']['library'], {
'name': 'analytics-python',
'version': VERSION
})
self.assertEqual(msg['category'], 'category')
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertEqual(msg['messageId'], 'messageId')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'page')
self.assertEqual(msg['name'], 'name')
def test_basic_screen(self):
client = self.client
success, msg = client.screen('distinct_id', name='name')
client.flush()
self.assertTrue(success)
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'screen')
self.assertEqual(msg['name'], 'name')
def test_advanced_screen(self):
client = self.client
success, msg = client.screen(
'distinct_id', 'category', 'name', {'property': 'value'},
{'ip': '192.168.0.1'}, datetime(2014, 9, 3), 'messageId')
self.assertTrue(success)
self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00')
self.assertEqual(msg['context']['ip'], '192.168.0.1')
self.assertEqual(msg['properties'], {'property': 'value'})
self.assertEqual(msg['context']['library'], {
'name': 'analytics-python',
'version': VERSION
})
self.assertTrue(isinstance(msg['timestamp'], str))
self.assertEqual(msg['messageId'], 'messageId')
self.assertEqual(msg['category'], 'category')
self.assertEqual(msg['distinct_id'], 'distinct_id')
self.assertEqual(msg['type'], 'screen')
self.assertEqual(msg['name'], 'name')
def test_flush(self):
client = self.client
# set up the consumer with more requests than a single batch will allow
for i in range(1000):
success, msg = client.identify('distinct_id', {'trait': 'value'})
# We can't reliably assert that the queue is non-empty here; that's
# a race condition. We do our best to load it up though.
client.flush()
# Make sure that the client queue is empty after flushing
self.assertTrue(client.queue.empty())
def test_shutdown(self):
client = self.client
# set up the consumer with more requests than a single batch will allow
for i in range(1000):
success, msg = client.identify('distinct_id', {'trait': 'value'})
client.shutdown()
# we expect two things after shutdown:
# 1. client queue is empty
# 2. consumer thread has stopped
self.assertTrue(client.queue.empty())
for consumer in client.consumers:
self.assertFalse(consumer.is_alive())
def test_synchronous(self):
client = Client('testsecret', sync_mode=True)
success, message = client.identify('distinct_id')
self.assertFalse(client.consumers)
self.assertTrue(client.queue.empty())
self.assertTrue(success)
def test_overflow(self):
client = Client('testsecret', max_queue_size=1)
# Ensure consumer thread is no longer uploading
client.join()
for i in range(10):
client.identify('distinct_id')
success, msg = client.identify('distinct_id')
# Make sure we are informed that the queue is at capacity
self.assertFalse(success)
def test_success_on_invalid_api_key(self):
client = Client('bad_key', on_error=self.fail)
client.track('distinct_id', 'event')
client.flush()
self.assertFalse(self.failed)
def test_unicode(self):
Client(six.u('unicode_key'))
def test_numeric_distinct_id(self):
self.client.track(1234, 'python event')
self.client.flush()
self.assertFalse(self.failed)
def test_debug(self):
Client('bad_key', debug=True)
def test_identify_with_date_object(self):
client = self.client
success, msg = client.identify(
'distinct_id',
{
'birthdate': date(1981, 2, 2),
},
)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg['traits'], {'birthdate': date(1981, 2, 2)})
def test_gzip(self):
client = Client('testsecret', on_error=self.fail, gzip=True)
for _ in range(10):
client.identify('distinct_id', {'trait': 'value'})
client.flush()
self.assertFalse(self.failed)
def test_user_defined_flush_at(self):
client = Client('testsecret', on_error=self.fail,
flush_at=10, flush_interval=3)
def mock_post_fn(*args, **kwargs):
self.assertEquals(len(kwargs['batch']), 10)
# the post function should be called 2 times, with a batch size of 10
# each time.
with mock.patch('analytics.consumer.post', side_effect=mock_post_fn) \
as mock_post:
for _ in range(20):
client.identify('distinct_id', {'trait': 'value'})
time.sleep(1)
self.assertEquals(mock_post.call_count, 2)
def test_user_defined_timeout(self):
client = Client('testsecret', timeout=10)
for consumer in client.consumers:
self.assertEquals(consumer.timeout, 10)
def test_default_timeout_15(self):
client = Client('testsecret')
for consumer in client.consumers:
self.assertEquals(consumer.timeout, 15)
@@ -1,23 +1,22 @@
import json
import time
import unittest
import mock
import time
import json
try:
from queue import Queue
except ImportError:
from Queue import Queue
from posthog.consumer import MAX_MSG_SIZE, Consumer
from posthog.consumer import Consumer, MAX_MSG_SIZE
from posthog.request import APIError
from posthog.test.test_utils import TEST_API_KEY
class TestConsumer(unittest.TestCase):
def test_next(self):
q = Queue()
consumer = Consumer(q, "")
consumer = Consumer(q, '')
q.put(1)
next = consumer.next()
self.assertEqual(next, [1])
@@ -25,7 +24,7 @@ class TestConsumer(unittest.TestCase):
def test_next_limit(self):
q = Queue()
flush_at = 50
consumer = Consumer(q, "", flush_at)
consumer = Consumer(q, '', flush_at)
for i in range(10000):
q.put(i)
next = consumer.next()
@@ -33,8 +32,8 @@ class TestConsumer(unittest.TestCase):
def test_dropping_oversize_msg(self):
q = Queue()
consumer = Consumer(q, "")
oversize_msg = {"m": "x" * MAX_MSG_SIZE}
consumer = Consumer(q, '')
oversize_msg = {'m': 'x' * MAX_MSG_SIZE}
q.put(oversize_msg)
next = consumer.next()
self.assertEqual(next, [])
@@ -42,8 +41,12 @@ class TestConsumer(unittest.TestCase):
def test_upload(self):
q = Queue()
consumer = Consumer(q, TEST_API_KEY)
track = {"type": "track", "event": "python event", "distinct_id": "distinct_id"}
consumer = Consumer(q, 'testsecret')
track = {
'type': 'track',
'event': 'python event',
'distinct_id': 'distinct_id'
}
q.put(track)
success = consumer.upload()
self.assertTrue(success)
@@ -54,11 +57,16 @@ class TestConsumer(unittest.TestCase):
# The consumer should upload _n_ times.
q = Queue()
flush_interval = 0.3
consumer = Consumer(q, TEST_API_KEY, flush_at=10, flush_interval=flush_interval)
with mock.patch("posthog.consumer.batch_post") as mock_post:
consumer = Consumer(q, 'testsecret', flush_at=10,
flush_interval=flush_interval)
with mock.patch('analytics.consumer.post') as mock_post:
consumer.start()
for i in range(0, 3):
track = {"type": "track", "event": "python event %d" % i, "distinct_id": "distinct_id"}
track = {
'type': 'track',
'event': 'python event %d' % i,
'distinct_id': 'distinct_id'
}
q.put(track)
time.sleep(flush_interval * 1.1)
self.assertEqual(mock_post.call_count, 3)
@@ -69,30 +77,45 @@ class TestConsumer(unittest.TestCase):
q = Queue()
flush_interval = 0.5
flush_at = 10
consumer = Consumer(q, TEST_API_KEY, flush_at=flush_at, flush_interval=flush_interval)
with mock.patch("posthog.consumer.batch_post") as mock_post:
consumer = Consumer(q, 'testsecret', flush_at=flush_at,
flush_interval=flush_interval)
with mock.patch('analytics.consumer.post') as mock_post:
consumer.start()
for i in range(0, flush_at * 2):
track = {"type": "track", "event": "python event %d" % i, "distinct_id": "distinct_id"}
track = {
'type': 'track',
'event': 'python event %d' % i,
'distinct_id': 'distinct_id'
}
q.put(track)
time.sleep(flush_interval * 1.1)
self.assertEqual(mock_post.call_count, 2)
def test_request(self):
consumer = Consumer(None, TEST_API_KEY)
track = {"type": "track", "event": "python event", "distinct_id": "distinct_id"}
consumer = Consumer(None, 'testsecret')
track = {
'type': 'track',
'event': 'python event',
'distinct_id': 'distinct_id'
}
consumer.request([track])
def _test_request_retry(self, consumer, expected_exception, exception_count):
def _test_request_retry(self, consumer,
expected_exception, exception_count):
def mock_post(*args, **kwargs):
mock_post.call_count += 1
if mock_post.call_count <= exception_count:
raise expected_exception
mock_post.call_count = 0
with mock.patch("posthog.consumer.batch_post", mock.Mock(side_effect=mock_post)):
track = {"type": "track", "event": "python event", "distinct_id": "distinct_id"}
with mock.patch('analytics.consumer.post',
mock.Mock(side_effect=mock_post)):
track = {
'type': 'track',
'event': 'python event',
'distinct_id': 'distinct_id'
}
# request() should succeed if the number of exceptions raised is
# less than the retries paramater.
if exception_count <= consumer.retries:
@@ -107,63 +130,69 @@ class TestConsumer(unittest.TestCase):
self.assertEqual(exc, expected_exception)
else:
self.fail(
"request() should raise an exception if still failing after %d retries" % consumer.retries
)
"request() should raise an exception if still failing "
"after %d retries" % consumer.retries)
def test_request_retry(self):
# we should retry on general errors
consumer = Consumer(None, TEST_API_KEY)
self._test_request_retry(consumer, Exception("generic exception"), 2)
consumer = Consumer(None, 'testsecret')
self._test_request_retry(consumer, Exception('generic exception'), 2)
# we should retry on server errors
consumer = Consumer(None, TEST_API_KEY)
self._test_request_retry(consumer, APIError(500, "Internal Server Error"), 2)
consumer = Consumer(None, 'testsecret')
self._test_request_retry(consumer, APIError(
500, 'code', 'Internal Server Error'), 2)
# we should retry on HTTP 429 errors
consumer = Consumer(None, TEST_API_KEY)
self._test_request_retry(consumer, APIError(429, "Too Many Requests"), 2)
consumer = Consumer(None, 'testsecret')
self._test_request_retry(consumer, APIError(
429, 'code', 'Too Many Requests'), 2)
# we should NOT retry on other client errors
consumer = Consumer(None, TEST_API_KEY)
api_error = APIError(400, "Client Errors")
consumer = Consumer(None, 'testsecret')
api_error = APIError(400, 'code', 'Client Errors')
try:
self._test_request_retry(consumer, api_error, 1)
except APIError:
pass
else:
self.fail("request() should not retry on client errors")
self.fail('request() should not retry on client errors')
# test for number of exceptions raise > retries value
consumer = Consumer(None, TEST_API_KEY, retries=3)
self._test_request_retry(consumer, APIError(500, "Internal Server Error"), 3)
consumer = Consumer(None, 'testsecret', retries=3)
self._test_request_retry(consumer, APIError(
500, 'code', 'Internal Server Error'), 3)
def test_pause(self):
consumer = Consumer(None, TEST_API_KEY)
consumer = Consumer(None, 'testsecret')
consumer.pause()
self.assertFalse(consumer.running)
def test_max_batch_size(self):
q = Queue()
consumer = Consumer(q, TEST_API_KEY, flush_at=100000, flush_interval=3)
properties = {}
for n in range(0, 500):
properties[str(n)] = "one_long_property_value_to_build_a_big_event"
track = {"type": "track", "event": "python event", "distinct_id": "distinct_id", "properties": properties}
consumer = Consumer(
q, 'testsecret', flush_at=100000, flush_interval=3)
track = {
'type': 'track',
'event': 'python event',
'distinct_id': 'distinct_id'
}
msg_size = len(json.dumps(track).encode())
# Let's capture 8MB of data to trigger two batches
n_msgs = int(8_000_000 / msg_size)
# number of messages in a maximum-size batch
n_msgs = int(475000 / msg_size)
def mock_post_fn(_, data, **kwargs):
res = mock.Mock()
res.status_code = 200
request_size = len(data.encode())
# Batches close after the first message bringing it bigger than BATCH_SIZE_LIMIT, let's add 10% of margin
self.assertTrue(request_size < (5 * 1024 * 1024) * 1.1, "batch size (%d) higher than limit" % request_size)
self.assertTrue(len(data.encode()) < 500000,
'batch size (%d) exceeds 500KB limit'
% len(data.encode()))
return res
with mock.patch("posthog.request._session.post", side_effect=mock_post_fn) as mock_post:
with mock.patch('analytics.request._session.post',
side_effect=mock_post_fn) as mock_post:
consumer.start()
for _ in range(0, n_msgs + 2):
q.put(track)
q.join()
self.assertEqual(mock_post.call_count, 2)
self.assertEquals(mock_post.call_count, 2)
+49
View File
@@ -0,0 +1,49 @@
import unittest
import analytics
class TestModule(unittest.TestCase):
def failed(self):
self.failed = True
def setUp(self):
self.failed = False
analytics.api_key = 'testsecret'
analytics.on_error = self.failed
def test_no_api_key(self):
analytics.api_key = None
self.assertRaises(Exception, analytics.track)
def test_no_host(self):
analytics.host = None
self.assertRaises(Exception, analytics.track)
def test_track(self):
analytics.track('distinct_id', 'python module event')
analytics.flush()
def test_identify(self):
analytics.identify('distinct_id', {'email': 'user@email.com'})
analytics.flush()
def test_group(self):
analytics.group('distinct_id', 'groupId')
analytics.flush()
def test_alias(self):
analytics.alias('previousId', 'distinct_id')
analytics.flush()
def test_page(self):
analytics.page('distinct_id')
analytics.flush()
def test_screen(self):
analytics.screen('distinct_id')
analytics.flush()
def test_flush(self):
analytics.flush()
+53
View File
@@ -0,0 +1,53 @@
from datetime import datetime, date
import unittest
import json
import requests
from posthog.request import post, DatetimeSerializer
class TestRequests(unittest.TestCase):
def test_valid_request(self):
res = post(batch=[{
'distinct_id': 'distinct_id',
'event': 'python event',
'type': 'track'
}])
self.assertEqual(res.status_code, 200)
def test_invalid_request_error(self):
self.assertRaises(Exception, post, 'testsecret',
'https://t.posthog.com', False, '[{]')
def test_invalid_host(self):
self.assertRaises(Exception, post, 'testsecret',
't.posthog.com/', batch=[])
def test_datetime_serialization(self):
data = {'created': datetime(2012, 3, 4, 5, 6, 7, 891011)}
result = json.dumps(data, cls=DatetimeSerializer)
self.assertEqual(result, '{"created": "2012-03-04T05:06:07.891011"}')
def test_date_serialization(self):
today = date.today()
data = {'created': today}
result = json.dumps(data, cls=DatetimeSerializer)
expected = '{"created": "%s"}' % today.isoformat()
self.assertEqual(result, expected)
def test_should_not_timeout(self):
res = post(batch=[{
'distinct_id': 'distinct_id',
'event': 'python event',
'type': 'track'
}], timeout=15)
self.assertEqual(res.status_code, 200)
def test_should_timeout(self):
with self.assertRaises(requests.ReadTimeout):
post(batch=[{
'distinct_id': 'distinct_id',
'event': 'python event',
'type': 'track'
}], timeout=0.0001)
-892
View File
@@ -1,892 +0,0 @@
import time
import unittest
from datetime import datetime
from uuid import uuid4
import mock
import six
from posthog.client import Client
from posthog.test.test_utils import FAKE_TEST_API_KEY
from posthog.version import VERSION
class TestClient(unittest.TestCase):
@classmethod
def setUpClass(cls):
# This ensures no real HTTP POST requests are made
cls.client_post_patcher = mock.patch("posthog.client.batch_post")
cls.consumer_post_patcher = mock.patch("posthog.consumer.batch_post")
cls.client_post_patcher.start()
cls.consumer_post_patcher.start()
@classmethod
def tearDownClass(cls):
cls.client_post_patcher.stop()
cls.consumer_post_patcher.stop()
def set_fail(self, e, batch):
"""Mark the failure handler"""
print("FAIL", e, batch) # noqa: T201
self.failed = True
def setUp(self):
self.failed = False
self.client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail)
def test_requires_api_key(self):
self.assertRaises(AssertionError, Client)
def test_empty_flush(self):
self.client.flush()
def test_basic_capture(self):
client = self.client
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
def test_basic_capture_with_uuid(self):
client = self.client
uuid = str(uuid4())
success, msg = client.capture("distinct_id", "python test event", uuid=uuid)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertEqual(msg["uuid"], uuid)
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
def test_basic_capture_with_project_api_key(self):
client = Client(project_api_key=FAKE_TEST_API_KEY, on_error=self.set_fail)
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
@mock.patch("posthog.client.decide")
def test_basic_capture_with_feature_flags(self, patch_decide):
patch_decide.return_value = {"featureFlags": {"beta-feature": "random-variant"}}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY)
success, msg = client.capture("distinct_id", "python test event", send_feature_flags=True)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["properties"]["$feature/beta-feature"], "random-variant")
self.assertEqual(msg["properties"]["$active_feature_flags"], ["beta-feature"])
self.assertEqual(patch_decide.call_count, 1)
@mock.patch("posthog.client.decide")
def test_basic_capture_with_locally_evaluated_feature_flags(self, patch_decide):
patch_decide.return_value = {"featureFlags": {"beta-feature": "random-variant"}}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY)
multivariate_flag = {
"id": 1,
"name": "Beta Feature",
"key": "beta-feature-local",
"is_simple_flag": False,
"active": True,
"rollout_percentage": 100,
"filters": {
"groups": [
{
"properties": [
{"key": "email", "type": "person", "value": "test@posthog.com", "operator": "exact"}
],
"rollout_percentage": 100,
},
{
"rollout_percentage": 50,
},
],
"multivariate": {
"variants": [
{"key": "first-variant", "name": "First Variant", "rollout_percentage": 50},
{"key": "second-variant", "name": "Second Variant", "rollout_percentage": 25},
{"key": "third-variant", "name": "Third Variant", "rollout_percentage": 25},
]
},
"payloads": {"first-variant": "some-payload", "third-variant": {"a": "json"}},
},
}
basic_flag = {
"id": 1,
"name": "Beta Feature",
"key": "person-flag",
"is_simple_flag": True,
"active": True,
"filters": {
"groups": [
{
"properties": [
{
"key": "region",
"operator": "exact",
"value": ["USA"],
"type": "person",
}
],
"rollout_percentage": 100,
}
],
"payloads": {"true": 300},
},
}
false_flag = {
"id": 1,
"name": "Beta Feature",
"key": "false-flag",
"is_simple_flag": True,
"active": True,
"filters": {
"groups": [
{
"properties": [],
"rollout_percentage": 0,
}
],
"payloads": {"true": 300},
},
}
client.feature_flags = [multivariate_flag, basic_flag, false_flag]
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["properties"]["$feature/beta-feature-local"], "third-variant")
self.assertEqual(msg["properties"]["$feature/false-flag"], False)
self.assertEqual(msg["properties"]["$active_feature_flags"], ["beta-feature-local"])
assert "$feature/beta-feature" not in msg["properties"]
self.assertEqual(patch_decide.call_count, 0)
# test that flags are not evaluated without local evaluation
client.feature_flags = []
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
assert "$feature/beta-feature" not in msg["properties"]
assert "$feature/beta-feature-local" not in msg["properties"]
assert "$feature/false-flag" not in msg["properties"]
assert "$active_feature_flags" not in msg["properties"]
@mock.patch("posthog.client.decide")
def test_dont_override_capture_with_local_flags(self, patch_decide):
patch_decide.return_value = {"featureFlags": {"beta-feature": "random-variant"}}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY)
multivariate_flag = {
"id": 1,
"name": "Beta Feature",
"key": "beta-feature-local",
"is_simple_flag": False,
"active": True,
"rollout_percentage": 100,
"filters": {
"groups": [
{
"properties": [
{"key": "email", "type": "person", "value": "test@posthog.com", "operator": "exact"}
],
"rollout_percentage": 100,
},
{
"rollout_percentage": 50,
},
],
"multivariate": {
"variants": [
{"key": "first-variant", "name": "First Variant", "rollout_percentage": 50},
{"key": "second-variant", "name": "Second Variant", "rollout_percentage": 25},
{"key": "third-variant", "name": "Third Variant", "rollout_percentage": 25},
]
},
"payloads": {"first-variant": "some-payload", "third-variant": {"a": "json"}},
},
}
basic_flag = {
"id": 1,
"name": "Beta Feature",
"key": "person-flag",
"is_simple_flag": True,
"active": True,
"filters": {
"groups": [
{
"properties": [
{
"key": "region",
"operator": "exact",
"value": ["USA"],
"type": "person",
}
],
"rollout_percentage": 100,
}
],
"payloads": {"true": 300},
},
}
client.feature_flags = [multivariate_flag, basic_flag]
success, msg = client.capture(
"distinct_id", "python test event", {"$feature/beta-feature-local": "my-custom-variant"}
)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["properties"]["$feature/beta-feature-local"], "my-custom-variant")
self.assertEqual(msg["properties"]["$active_feature_flags"], ["beta-feature-local"])
assert "$feature/beta-feature" not in msg["properties"]
assert "$feature/person-flag" not in msg["properties"]
self.assertEqual(patch_decide.call_count, 0)
@mock.patch("posthog.client.decide")
def test_basic_capture_with_feature_flags_returns_active_only(self, patch_decide):
patch_decide.return_value = {
"featureFlags": {"beta-feature": "random-variant", "alpha-feature": True, "off-feature": False}
}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY)
success, msg = client.capture("distinct_id", "python test event", send_feature_flags=True)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertTrue(msg["properties"]["$geoip_disable"])
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["properties"]["$feature/beta-feature"], "random-variant")
self.assertEqual(msg["properties"]["$feature/alpha-feature"], True)
self.assertEqual(msg["properties"]["$active_feature_flags"], ["beta-feature", "alpha-feature"])
self.assertEqual(patch_decide.call_count, 1)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="distinct_id",
groups={},
person_properties=None,
group_properties=None,
disable_geoip=True,
)
@mock.patch("posthog.client.decide")
def test_basic_capture_with_feature_flags_and_disable_geoip_returns_correctly(self, patch_decide):
patch_decide.return_value = {
"featureFlags": {"beta-feature": "random-variant", "alpha-feature": True, "off-feature": False}
}
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY, disable_geoip=True
)
success, msg = client.capture("distinct_id", "python test event", send_feature_flags=True, disable_geoip=False)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertTrue("$geoip_disable" not in msg["properties"])
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["properties"]["$feature/beta-feature"], "random-variant")
self.assertEqual(msg["properties"]["$feature/alpha-feature"], True)
self.assertEqual(msg["properties"]["$active_feature_flags"], ["beta-feature", "alpha-feature"])
self.assertEqual(patch_decide.call_count, 1)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="distinct_id",
groups={},
person_properties=None,
group_properties=None,
disable_geoip=False,
)
@mock.patch("posthog.client.decide")
def test_basic_capture_with_feature_flags_switched_off_doesnt_send_them(self, patch_decide):
patch_decide.return_value = {"featureFlags": {"beta-feature": "random-variant"}}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, personal_api_key=FAKE_TEST_API_KEY)
success, msg = client.capture("distinct_id", "python test event", send_feature_flags=False)
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertTrue("$feature/beta-feature" not in msg["properties"])
self.assertTrue("$active_feature_flags" not in msg["properties"])
self.assertEqual(patch_decide.call_count, 0)
def test_stringifies_distinct_id(self):
# A large number that loses precision in node:
# node -e "console.log(157963456373623802 + 1)" > 157963456373623800
client = self.client
success, msg = client.capture(distinct_id=157963456373623802, event="python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["distinct_id"], "157963456373623802")
def test_advanced_capture(self):
client = self.client
success, msg = client.capture(
"distinct_id",
"python test event",
{"property": "value"},
{"ip": "192.168.0.1"},
datetime(2014, 9, 3),
"new-uuid",
)
self.assertTrue(success)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["properties"]["property"], "value")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
self.assertEqual(msg["event"], "python test event")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertEqual(msg["uuid"], "new-uuid")
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertTrue("$groups" not in msg["properties"])
def test_groups_capture(self):
success, msg = self.client.capture(
"distinct_id",
"test_event",
groups={"company": "id:5", "instance": "app.posthog.com"},
)
self.assertTrue(success)
self.assertEqual(msg["properties"]["$groups"], {"company": "id:5", "instance": "app.posthog.com"})
def test_basic_identify(self):
client = self.client
success, msg = client.identify("distinct_id", {"trait": "value"})
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["$set"]["trait"], "value")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_advanced_identify(self):
client = self.client
success, msg = client.identify(
"distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
)
self.assertTrue(success)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
self.assertEqual(msg["$set"]["trait"], "value")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertEqual(msg["uuid"], "new-uuid")
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_basic_set(self):
client = self.client
success, msg = client.set("distinct_id", {"trait": "value"})
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["$set"]["trait"], "value")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_advanced_set(self):
client = self.client
success, msg = client.set(
"distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
)
self.assertTrue(success)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
self.assertEqual(msg["$set"]["trait"], "value")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertEqual(msg["uuid"], "new-uuid")
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_basic_set_once(self):
client = self.client
success, msg = client.set_once("distinct_id", {"trait": "value"})
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["$set_once"]["trait"], "value")
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_advanced_set_once(self):
client = self.client
success, msg = client.set_once(
"distinct_id", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
)
self.assertTrue(success)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
self.assertEqual(msg["$set_once"]["trait"], "value")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertEqual(msg["uuid"], "new-uuid")
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_basic_group_identify(self):
success, msg = self.client.group_identify("organization", "id:5")
self.assertTrue(success)
self.assertEqual(msg["event"], "$groupidentify")
self.assertEqual(msg["distinct_id"], "$organization_id:5")
self.assertEqual(
msg["properties"],
{
"$group_type": "organization",
"$group_key": "id:5",
"$group_set": {},
"$lib": "posthog-python",
"$lib_version": VERSION,
"$geoip_disable": True,
},
)
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertIsNone(msg.get("uuid"))
def test_advanced_group_identify(self):
success, msg = self.client.group_identify(
"organization", "id:5", {"trait": "value"}, {"ip": "192.168.0.1"}, datetime(2014, 9, 3), "new-uuid"
)
self.assertTrue(success)
self.assertEqual(msg["event"], "$groupidentify")
self.assertEqual(msg["distinct_id"], "$organization_id:5")
self.assertEqual(
msg["properties"],
{
"$group_type": "organization",
"$group_key": "id:5",
"$group_set": {"trait": "value"},
"$lib": "posthog-python",
"$lib_version": VERSION,
"$geoip_disable": True,
},
)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
def test_basic_alias(self):
client = self.client
success, msg = client.alias("previousId", "distinct_id")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["properties"]["distinct_id"], "previousId")
self.assertEqual(msg["properties"]["alias"], "distinct_id")
def test_basic_page(self):
client = self.client
success, msg = client.page("distinct_id", url="https://posthog.com/contact")
self.assertFalse(self.failed)
client.flush()
self.assertTrue(success)
self.assertEqual(msg["distinct_id"], "distinct_id")
self.assertEqual(msg["properties"]["$current_url"], "https://posthog.com/contact")
def test_basic_page_distinct_uuid(self):
client = self.client
distinct_id = uuid4()
success, msg = client.page(distinct_id, url="https://posthog.com/contact")
self.assertFalse(self.failed)
client.flush()
self.assertTrue(success)
self.assertEqual(msg["distinct_id"], str(distinct_id))
self.assertEqual(msg["properties"]["$current_url"], "https://posthog.com/contact")
def test_advanced_page(self):
client = self.client
success, msg = client.page(
"distinct_id",
"https://posthog.com/contact",
{"property": "value"},
{"ip": "192.168.0.1"},
datetime(2014, 9, 3),
"new-uuid",
)
self.assertTrue(success)
self.assertEqual(msg["timestamp"], "2014-09-03T00:00:00+00:00")
self.assertEqual(msg["context"]["ip"], "192.168.0.1")
self.assertEqual(msg["properties"]["$current_url"], "https://posthog.com/contact")
self.assertEqual(msg["properties"]["property"], "value")
self.assertEqual(msg["properties"]["$lib"], "posthog-python")
self.assertEqual(msg["properties"]["$lib_version"], VERSION)
self.assertTrue(isinstance(msg["timestamp"], str))
self.assertEqual(msg["uuid"], "new-uuid")
self.assertEqual(msg["distinct_id"], "distinct_id")
def test_flush(self):
client = self.client
# set up the consumer with more requests than a single batch will allow
for i in range(1000):
success, msg = client.identify("distinct_id", {"trait": "value"})
# We can't reliably assert that the queue is non-empty here; that's
# a race condition. We do our best to load it up though.
client.flush()
# Make sure that the client queue is empty after flushing
self.assertTrue(client.queue.empty())
def test_shutdown(self):
client = self.client
# set up the consumer with more requests than a single batch will allow
for i in range(1000):
success, msg = client.identify("distinct_id", {"trait": "value"})
client.shutdown()
# we expect two things after shutdown:
# 1. client queue is empty
# 2. consumer thread has stopped
self.assertTrue(client.queue.empty())
for consumer in client.consumers:
self.assertFalse(consumer.is_alive())
def test_synchronous(self):
client = Client(FAKE_TEST_API_KEY, sync_mode=True)
success, message = client.identify("distinct_id")
self.assertFalse(client.consumers)
self.assertTrue(client.queue.empty())
self.assertTrue(success)
def test_overflow(self):
client = Client(FAKE_TEST_API_KEY, max_queue_size=1)
# Ensure consumer thread is no longer uploading
client.join()
for i in range(10):
client.identify("distinct_id")
success, msg = client.identify("distinct_id")
# Make sure we are informed that the queue is at capacity
self.assertFalse(success)
def test_unicode(self):
Client(six.u("unicode_key"))
def test_numeric_distinct_id(self):
self.client.capture(1234, "python event")
self.client.flush()
self.assertFalse(self.failed)
def test_debug(self):
Client("bad_key", debug=True)
def test_gzip(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.fail, gzip=True)
for _ in range(10):
client.identify("distinct_id", {"trait": "value"})
client.flush()
self.assertFalse(self.failed)
def test_user_defined_flush_at(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.fail, flush_at=10, flush_interval=3)
def mock_post_fn(*args, **kwargs):
self.assertEqual(len(kwargs["batch"]), 10)
# the post function should be called 2 times, with a batch size of 10
# each time.
with mock.patch("posthog.consumer.batch_post", side_effect=mock_post_fn) as mock_post:
for _ in range(20):
client.identify("distinct_id", {"trait": "value"})
time.sleep(1)
self.assertEqual(mock_post.call_count, 2)
def test_user_defined_timeout(self):
client = Client(FAKE_TEST_API_KEY, timeout=10)
for consumer in client.consumers:
self.assertEqual(consumer.timeout, 10)
def test_default_timeout_15(self):
client = Client(FAKE_TEST_API_KEY)
for consumer in client.consumers:
self.assertEqual(consumer.timeout, 15)
def test_disabled(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disabled=True)
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertFalse(success)
self.assertFalse(self.failed)
self.assertEqual(msg, "disabled")
@mock.patch("posthog.client.decide")
def test_disabled_with_feature_flags(self, patch_decide):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disabled=True)
response = client.get_feature_flag("beta-feature", "12345")
self.assertIsNone(response)
patch_decide.assert_not_called()
response = client.feature_enabled("beta-feature", "12345")
self.assertIsNone(response)
patch_decide.assert_not_called()
response = client.get_all_flags("12345")
self.assertIsNone(response)
patch_decide.assert_not_called()
response = client.get_feature_flag_payload("key", "12345")
self.assertIsNone(response)
patch_decide.assert_not_called()
response = client.get_all_flags_and_payloads("12345")
self.assertEqual(response, {"featureFlags": None, "featureFlagPayloads": None})
patch_decide.assert_not_called()
# no capture calls
self.assertTrue(client.queue.empty())
def test_enabled_to_disabled(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disabled=False)
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertTrue(success)
self.assertFalse(self.failed)
self.assertEqual(msg["event"], "python test event")
client.disabled = True
success, msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertFalse(success)
self.assertFalse(self.failed)
self.assertEqual(msg, "disabled")
def test_disable_geoip_default_on_events(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disable_geoip=True)
_, capture_msg = client.capture("distinct_id", "python test event")
client.flush()
self.assertEqual(capture_msg["properties"]["$geoip_disable"], True)
_, identify_msg = client.identify("distinct_id", {"trait": "value"})
client.flush()
self.assertEqual(identify_msg["properties"]["$geoip_disable"], True)
def test_disable_geoip_override_on_events(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disable_geoip=False)
_, capture_msg = client.set("distinct_id", {"a": "b", "c": "d"}, disable_geoip=True)
client.flush()
self.assertEqual(capture_msg["properties"]["$geoip_disable"], True)
_, identify_msg = client.page("distinct_id", "http://a.com", {"trait": "value"}, disable_geoip=False)
client.flush()
self.assertEqual("$geoip_disable" not in identify_msg["properties"], True)
def test_disable_geoip_method_overrides_init_on_events(self):
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disable_geoip=True)
_, msg = client.capture("distinct_id", "python test event", disable_geoip=False)
client.flush()
self.assertTrue("$geoip_disable" not in msg["properties"])
@mock.patch("posthog.client.decide")
def test_disable_geoip_default_on_decide(self, patch_decide):
patch_decide.return_value = {
"featureFlags": {"beta-feature": "random-variant", "alpha-feature": True, "off-feature": False}
}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disable_geoip=False)
client.get_feature_flag("random_key", "some_id", disable_geoip=True)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="some_id",
groups={},
person_properties={"distinct_id": "some_id"},
group_properties={},
disable_geoip=True,
)
patch_decide.reset_mock()
client.feature_enabled("random_key", "feature_enabled_distinct_id", disable_geoip=True)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="feature_enabled_distinct_id",
groups={},
person_properties={"distinct_id": "feature_enabled_distinct_id"},
group_properties={},
disable_geoip=True,
)
patch_decide.reset_mock()
client.get_all_flags_and_payloads("all_flags_payloads_id")
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="all_flags_payloads_id",
groups={},
person_properties={"distinct_id": "all_flags_payloads_id"},
group_properties={},
disable_geoip=False,
)
@mock.patch("posthog.client.Poller")
@mock.patch("posthog.client.get")
def test_call_identify_fails(self, patch_get, patch_poll):
def raise_effect():
raise Exception("http exception")
patch_get.return_value.raiseError.side_effect = raise_effect
client = Client(FAKE_TEST_API_KEY, personal_api_key="test")
client.feature_flags = [{"key": "example", "is_simple_flag": False}]
self.assertFalse(client.feature_enabled("example", "distinct_id"))
@mock.patch("posthog.client.decide")
def test_default_properties_get_added_properly(self, patch_decide):
patch_decide.return_value = {
"featureFlags": {"beta-feature": "random-variant", "alpha-feature": True, "off-feature": False}
}
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, disable_geoip=False)
client.get_feature_flag(
"random_key",
"some_id",
groups={"company": "id:5", "instance": "app.posthog.com"},
person_properties={"x1": "y1"},
group_properties={"company": {"x": "y"}},
)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="some_id",
groups={"company": "id:5", "instance": "app.posthog.com"},
person_properties={"distinct_id": "some_id", "x1": "y1"},
group_properties={
"company": {"$group_key": "id:5", "x": "y"},
"instance": {"$group_key": "app.posthog.com"},
},
disable_geoip=False,
)
patch_decide.reset_mock()
client.get_feature_flag(
"random_key",
"some_id",
groups={"company": "id:5", "instance": "app.posthog.com"},
person_properties={"distinct_id": "override"},
group_properties={
"company": {
"$group_key": "group_override",
}
},
)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="some_id",
groups={"company": "id:5", "instance": "app.posthog.com"},
person_properties={"distinct_id": "override"},
group_properties={
"company": {"$group_key": "group_override"},
"instance": {"$group_key": "app.posthog.com"},
},
disable_geoip=False,
)
patch_decide.reset_mock()
# test nones
client.get_all_flags_and_payloads("some_id", groups={}, person_properties=None, group_properties=None)
patch_decide.assert_called_with(
"random_key",
None,
timeout=10,
distinct_id="some_id",
groups={},
person_properties={"distinct_id": "some_id"},
group_properties={},
disable_geoip=False,
)
File diff suppressed because it is too large Load Diff
-48
View File
@@ -1,48 +0,0 @@
import unittest
from posthog import Posthog
class TestModule(unittest.TestCase):
posthog = None
def _assert_enqueue_result(self, result):
self.assertEqual(type(result[0]), bool)
self.assertEqual(type(result[1]), dict)
def failed(self):
self.failed = True
def setUp(self):
self.failed = False
self.posthog = Posthog("testsecret", host="http://localhost:8000", on_error=self.failed)
def test_no_api_key(self):
self.posthog.api_key = None
self.assertRaises(Exception, self.posthog.capture)
def test_no_host(self):
self.posthog.host = None
self.assertRaises(Exception, self.posthog.capture)
def test_track(self):
res = self.posthog.capture("distinct_id", "python module event")
self._assert_enqueue_result(res)
self.posthog.flush()
def test_identify(self):
res = self.posthog.identify("distinct_id", {"email": "user@email.com"})
self._assert_enqueue_result(res)
self.posthog.flush()
def test_alias(self):
res = self.posthog.alias("previousId", "distinct_id")
self._assert_enqueue_result(res)
self.posthog.flush()
def test_page(self):
self.posthog.page("distinct_id", "https://posthog.com/contact")
self.posthog.flush()
def test_flush(self):
self.posthog.flush()
-44
View File
@@ -1,44 +0,0 @@
import json
import unittest
from datetime import date, datetime
import requests
from posthog.request import DatetimeSerializer, batch_post
from posthog.test.test_utils import TEST_API_KEY
class TestRequests(unittest.TestCase):
def test_valid_request(self):
res = batch_post(TEST_API_KEY, batch=[{"distinct_id": "distinct_id", "event": "python event", "type": "track"}])
self.assertEqual(res.status_code, 200)
def test_invalid_request_error(self):
self.assertRaises(Exception, batch_post, "testsecret", "https://t.posthog.com", False, "[{]")
def test_invalid_host(self):
self.assertRaises(Exception, batch_post, "testsecret", "t.posthog.com/", batch=[])
def test_datetime_serialization(self):
data = {"created": datetime(2012, 3, 4, 5, 6, 7, 891011)}
result = json.dumps(data, cls=DatetimeSerializer)
self.assertEqual(result, '{"created": "2012-03-04T05:06:07.891011"}')
def test_date_serialization(self):
today = date.today()
data = {"created": today}
result = json.dumps(data, cls=DatetimeSerializer)
expected = '{"created": "%s"}' % today.isoformat()
self.assertEqual(result, expected)
def test_should_not_timeout(self):
res = batch_post(
TEST_API_KEY, batch=[{"distinct_id": "distinct_id", "event": "python event", "type": "track"}], timeout=15
)
self.assertEqual(res.status_code, 200)
def test_should_timeout(self):
with self.assertRaises(requests.ReadTimeout):
batch_post(
"key", batch=[{"distinct_id": "distinct_id", "event": "python event", "type": "track"}], timeout=0.0001
)
-101
View File
@@ -1,101 +0,0 @@
import unittest
from datetime import date, datetime, timedelta
from decimal import Decimal
from uuid import UUID
import six
from dateutil.tz import tzutc
from posthog import utils
TEST_API_KEY = "kOOlRy2QlMY9jHZQv0bKz0FZyazBUoY8Arj0lFVNjs4"
FAKE_TEST_API_KEY = "random_key"
class TestUtils(unittest.TestCase):
def test_timezone_utils(self):
now = datetime.now()
utcnow = datetime.now(tz=tzutc())
self.assertTrue(utils.is_naive(now))
self.assertFalse(utils.is_naive(utcnow))
fixed = utils.guess_timezone(now)
self.assertFalse(utils.is_naive(fixed))
shouldnt_be_edited = utils.guess_timezone(utcnow)
self.assertEqual(utcnow, shouldnt_be_edited)
def test_clean(self):
simple = {
"decimal": Decimal("0.142857"),
"unicode": six.u("woo"),
"date": datetime.now(),
"long": 200000000,
"integer": 1,
"float": 2.0,
"bool": True,
"str": "woo",
"none": None,
}
complicated = {
"exception": Exception("This should show up"),
"timedelta": timedelta(microseconds=20),
"list": [1, 2, 3],
}
combined = dict(simple.items())
combined.update(complicated.items())
pre_clean_keys = combined.keys()
utils.clean(combined)
self.assertEqual(combined.keys(), pre_clean_keys)
# test UUID separately, as the UUID object doesn't equal its string representation according to Python
self.assertEqual(utils.clean(UUID("12345678123456781234567812345678")), "12345678-1234-5678-1234-567812345678")
def test_clean_with_dates(self):
dict_with_dates = {
"birthdate": date(1980, 1, 1),
"registration": datetime.utcnow(),
}
self.assertEqual(dict_with_dates, utils.clean(dict_with_dates))
def test_bytes(self):
if six.PY3:
item = bytes(10)
else:
item = bytearray(10)
utils.clean(item)
def test_clean_fn(self):
cleaned = utils.clean({"fn": lambda x: x, "number": 4})
self.assertEqual(cleaned["number"], 4)
# TODO: fixme, different behavior on python 2 and 3
if "fn" in cleaned:
self.assertEqual(cleaned["fn"], None)
def test_remove_slash(self):
self.assertEqual("http://posthog.io", utils.remove_trailing_slash("http://posthog.io/"))
self.assertEqual("http://posthog.io", utils.remove_trailing_slash("http://posthog.io"))
class TestSizeLimitedDict(unittest.TestCase):
def test_size_limited_dict(self):
size = 10
values = utils.SizeLimitedDict(size, lambda _: -1)
for i in range(100):
values[i] = i
self.assertEqual(values[i], i)
self.assertEqual(len(values), i % size + 1)
if i % size == 0:
# old numbers should've been removed
self.assertIsNone(values.get(i - 1))
self.assertIsNone(values.get(i - 3))
self.assertIsNone(values.get(i - 5))
self.assertIsNone(values.get(i - 9))
+78
View File
@@ -0,0 +1,78 @@
from datetime import date, datetime, timedelta
from decimal import Decimal
import unittest
from dateutil.tz import tzutc
import six
from posthog import utils
class TestUtils(unittest.TestCase):
def test_timezone_utils(self):
now = datetime.now()
utcnow = datetime.now(tz=tzutc())
self.assertTrue(utils.is_naive(now))
self.assertFalse(utils.is_naive(utcnow))
fixed = utils.guess_timezone(now)
self.assertFalse(utils.is_naive(fixed))
shouldnt_be_edited = utils.guess_timezone(utcnow)
self.assertEqual(utcnow, shouldnt_be_edited)
def test_clean(self):
simple = {
'decimal': Decimal('0.142857'),
'unicode': six.u('woo'),
'date': datetime.now(),
'long': 200000000,
'integer': 1,
'float': 2.0,
'bool': True,
'str': 'woo',
'none': None
}
complicated = {
'exception': Exception('This should show up'),
'timedelta': timedelta(microseconds=20),
'list': [1, 2, 3]
}
combined = dict(simple.items())
combined.update(complicated.items())
pre_clean_keys = combined.keys()
utils.clean(combined)
self.assertEqual(combined.keys(), pre_clean_keys)
def test_clean_with_dates(self):
dict_with_dates = {
'birthdate': date(1980, 1, 1),
'registration': datetime.utcnow(),
}
self.assertEqual(dict_with_dates, utils.clean(dict_with_dates))
def test_bytes(self):
if six.PY3:
item = bytes(10)
else:
item = bytearray(10)
utils.clean(item)
def test_clean_fn(self):
cleaned = utils.clean({'fn': lambda x: x, 'number': 4})
self.assertEqual(cleaned['number'], 4)
# TODO: fixme, different behavior on python 2 and 3
if 'fn' in cleaned:
self.assertEqual(cleaned['fn'], None)
def test_remove_slash(self):
self.assertEqual('http://posthog.io',
utils.remove_trailing_slash('http://posthog.io/'))
self.assertEqual('http://posthog.io',
utils.remove_trailing_slash('http://posthog.io'))
+13 -43
View File
@@ -1,15 +1,12 @@
from dateutil.tz import tzlocal, tzutc
from datetime import date, datetime
from decimal import Decimal
import logging
import numbers
import re
from collections import defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal
from uuid import UUID
import six
from dateutil.tz import tzlocal, tzutc
log = logging.getLogger("posthog")
log = logging.getLogger('posthog')
def is_naive(dt):
@@ -20,7 +17,8 @@ def is_naive(dt):
def total_seconds(delta):
"""Determines total seconds with python < 2.7 compat."""
# http://stackoverflow.com/questions/3694835/python-2-6-5-divide-timedelta-with-timedelta
return (delta.microseconds + (delta.seconds + delta.days * 24 * 3600) * 1e6) / 1e6
return (delta.microseconds
+ (delta.seconds + delta.days * 24 * 3600) * 1e6) / 1e6
def guess_timezone(dt):
@@ -41,7 +39,7 @@ def guess_timezone(dt):
def remove_trailing_slash(host):
if host.endswith("/"):
if host.endswith('/'):
return host[:-1]
return host
@@ -49,9 +47,8 @@ def remove_trailing_slash(host):
def clean(item):
if isinstance(item, Decimal):
return float(item)
if isinstance(item, UUID):
return str(item)
elif isinstance(item, (six.string_types, bool, numbers.Number, datetime, date, type(None))):
elif isinstance(item, (six.string_types, bool, numbers.Number, datetime,
date, type(None))):
return item
elif isinstance(item, (set, list, tuple)):
return _clean_list(item)
@@ -72,10 +69,9 @@ def _clean_dict(dict_):
data[k] = clean(v)
except TypeError:
log.warning(
'Dictionary values must be serializeable to JSON "%s" value %s of type %s is unsupported.',
k,
v,
type(v),
'Dictionary values must be serializeable to '
'JSON "%s" value %s of type %s is unsupported.',
k, v, type(v),
)
return data
@@ -86,32 +82,6 @@ def _coerce_unicode(cmplx):
except AttributeError as exception:
item = ":".join(exception)
item.decode("utf-8", "strict")
log.warning("Error decoding: %s", item)
log.warning('Error decoding: %s', item)
return None
return item
def is_valid_regex(value) -> bool:
try:
re.compile(value)
return True
except re.error:
return False
class SizeLimitedDict(defaultdict):
def __init__(self, max_size, *args, **kwargs):
super().__init__(*args, **kwargs)
self.max_size = max_size
def __setitem__(self, key, value):
if len(self) >= self.max_size:
self.clear()
super().__setitem__(key, value)
def convert_to_datetime_aware(date_obj):
if date_obj.tzinfo is None:
date_obj = date_obj.replace(tzinfo=timezone.utc)
return date_obj
+1 -4
View File
@@ -1,4 +1 @@
VERSION = "3.3.4"
if __name__ == "__main__":
print(VERSION, end="") # noqa: T201
VERSION = '1.0.8'
-10
View File
@@ -1,10 +0,0 @@
[tool.black]
line-length = 120
[tool.isort]
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 8
ensure_newline_before_comments = true
line_length = 120
virtual_env = "env"
View File
-22
View File
@@ -1,22 +0,0 @@
#!/usr/bin/env python
"""Django's command-line utility for administrative tasks."""
import os
import sys
def main():
"""Run administrative tasks."""
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "sentry_django_example.settings")
try:
from django.core.management import execute_from_command_line
except ImportError as exc:
raise ImportError(
"Couldn't import Django. Are you sure it's installed and "
"available on your PYTHONPATH environment variable? Did you "
"forget to activate a virtual environment?"
) from exc
execute_from_command_line(sys.argv)
if __name__ == "__main__":
main()
@@ -1,16 +0,0 @@
"""
ASGI config for sentry_django_example project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/3.2/howto/deployment/asgi/
"""
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "sentry_django_example.settings")
application = get_asgi_application()
@@ -1,167 +0,0 @@
"""
Django settings for sentry_django_example project.
Generated by 'django-admin startproject' using Django 3.2.2.
For more information on this file, see
https://docs.djangoproject.com/en/3.2/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/3.2/ref/settings/
"""
from pathlib import Path
from uuid import uuid4
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/3.2/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = "django-insecure-4kzfiq7vb(t0+jbl#vq)u=%06ouf)n*=l%730c8=tk(wkm9i9o"
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
ALLOWED_HOSTS = []
# PostHog Setup (can be a separate app)
import posthog
# You can find this key on the /setup page in PostHog
posthog.api_key = "LXP6nQXvo-2TCqGVrWvPah8uJIyVykoMmhnEkEBi5PA" # TODO: replace with your api key
posthog.personal_api_key = ""
# Where you host PostHog, with no trailing /.
# You can remove this line if you're using posthog.com
posthog.host = "http://127.0.0.1:8000"
from posthog.sentry.posthog_integration import PostHogIntegration
PostHogIntegration.organization = "posthog" # TODO: your sentry organization
# PostHogIntegration.prefix = # TODO: your self hosted Sentry url. (default: https://sentry.io/organizations/)
# Since Sentry doesn't allow Integrations configuration (see https://github.com/getsentry/sentry-python/blob/master/sentry_sdk/integrations/__init__.py#L171-L183)
# we work around this by setting static class variables beforehand
# Sentry Setup
import sentry_sdk
from sentry_sdk.integrations.django import DjangoIntegration
sentry_sdk.init(
dsn="https://27ac54f7f4cf484abf1335436b0c52e5@o344752.ingest.sentry.io/5624115", # TODO: your Sentry DSN here
integrations=[DjangoIntegration(), PostHogIntegration()],
# Set traces_sample_rate to 1.0 to capture 100%
# of transactions for performance monitoring.
# We recommend adjusting this value in production.
traces_sample_rate=1.0,
# If you wish to associate users to errors (assuming you are using
# django.contrib.auth) you may enable sending PII data.
send_default_pii=True,
)
POSTHOG_DJANGO = {
"distinct_id": lambda request: str(uuid4()) # TODO: your logic for generating unique ID, given the request object
}
# Application definition
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
]
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
"posthog.sentry.django.PosthogDistinctIdMiddleware",
]
ROOT_URLCONF = "sentry_django_example.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
WSGI_APPLICATION = "sentry_django_example.wsgi.application"
# Database
# https://docs.djangoproject.com/en/3.2/ref/settings/#databases
DATABASES = {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": BASE_DIR / "db.sqlite3",
}
}
# Password validation
# https://docs.djangoproject.com/en/3.2/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
},
{
"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
},
{
"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
},
{
"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
},
]
# Internationalization
# https://docs.djangoproject.com/en/3.2/topics/i18n/
LANGUAGE_CODE = "en-us"
TIME_ZONE = "UTC"
USE_I18N = True
USE_L10N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/3.2/howto/static-files/
STATIC_URL = "/static/"
# Default primary key field type
# https://docs.djangoproject.com/en/3.2/ref/settings/#default-auto-field
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
@@ -1,28 +0,0 @@
"""sentry_django_example URL Configuration
The `urlpatterns` list routes URLs to views. For more information please see:
https://docs.djangoproject.com/en/3.2/topics/http/urls/
Examples:
Function views
1. Add an import: from my_app import views
2. Add a URL to urlpatterns: path('', views.home, name='home')
Class-based views
1. Add an import: from other_app.views import Home
2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')
Including another URLconf
1. Import the include() function: from django.urls import include, path
2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import path
def trigger_error(request):
division_by_zero = 1 / 0
urlpatterns = [
path("admin/", admin.site.urls),
path("sentry-debug/", trigger_error),
]
@@ -1,16 +0,0 @@
"""
WSGI config for sentry_django_example project.
It exposes the WSGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/3.2/howto/deployment/wsgi/
"""
import os
from django.core.wsgi import get_wsgi_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "sentry_django_example.settings")
application = get_wsgi_application()
+25 -26
View File
@@ -1,3 +1,4 @@
import os
import sys
@@ -7,41 +8,39 @@ except ImportError:
from distutils.core import setup
# Don't import analytics-python module here, since deps may not be installed
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "posthog"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'posthog'))
from version import VERSION
long_description = """
long_description = '''
PostHog is developer-friendly, self-hosted product analytics. posthog-python is the python package.
"""
'''
install_requires = ["requests>=2.7,<3.0", "six>=1.5", "monotonic>=1.5", "backoff>=1.10.0", "python-dateutil>2.1"]
install_requires = [
"requests>=2.7,<3.0",
"six>=1.5",
"monotonic>=1.5",
"backoff==1.6.0",
"python-dateutil>2.1"
]
extras_require = {
"dev": [
"black",
"isort",
"flake8",
"flake8-print",
"pre-commit",
],
"test": ["mock>=2.0.0", "freezegun==0.3.15", "pylint", "flake8", "coverage", "pytest", "pytest-timeout"],
"sentry": ["sentry-sdk", "django"],
}
tests_require = [
"mock>=2.0.0"
]
setup(
name="posthog",
name='posthog',
version=VERSION,
url="https://github.com/posthog/posthog-python",
author="Posthog",
author_email="hey@posthog.com",
maintainer="PostHog",
maintainer_email="hey@posthog.com",
test_suite="posthog.test.all",
packages=["posthog", "posthog.test", "posthog.sentry"],
license="MIT License",
url='https://github.com/posthog/posthog-python',
author='Posthog',
author_email='hey@posthog.com',
maintainer='PostHog',
maintainer_email='hey@posthog.com',
test_suite='posthog.test.all',
packages=['posthog', 'posthog.test'],
license='MIT License',
install_requires=install_requires,
extras_require=extras_require,
description="Integrate PostHog into any python application.",
tests_require=tests_require,
description='Integrate PostHog into any python application.',
long_description=long_description,
classifiers=[
"Development Status :: 5 - Production/Stable",
+24 -18
View File
@@ -1,3 +1,4 @@
import os
import sys
@@ -7,34 +8,39 @@ except ImportError:
from distutils.core import setup
# Don't import module here, since deps may not be installed
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "posthoganalytics"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'posthoganalytics'))
from version import VERSION
long_description = """
long_description = '''
PostHog is developer-friendly, self-hosted product analytics. posthog-python is the python package.
"""
'''
install_requires = ["requests>=2.7,<3.0", "six>=1.5", "monotonic>=1.5", "backoff>=1.10.0", "python-dateutil>2.1"]
install_requires = [
"requests>=2.7,<3.0",
"six>=1.5",
"monotonic>=1.5",
"backoff==1.6.0",
"python-dateutil>2.1"
]
tests_require = ["mock>=2.0.0"]
tests_require = [
"mock>=2.0.0"
]
setup(
name="posthoganalytics",
name='posthoganalytics',
version=VERSION,
url="https://github.com/posthog/posthog-python",
author="Posthog",
author_email="hey@posthog.com",
maintainer="PostHog",
maintainer_email="hey@posthog.com",
test_suite="posthoganalytics.test.all",
packages=["posthoganalytics", "posthoganalytics.test", "posthoganalytics.sentry"],
license="MIT License",
url='https://github.com/posthog/posthog-python',
author='Posthog',
author_email='hey@posthog.com',
maintainer='PostHog',
maintainer_email='hey@posthog.com',
test_suite='posthoganalytics.test.all',
packages=['posthoganalytics', 'posthoganalytics.test'],
license='MIT License',
install_requires=install_requires,
tests_require=tests_require,
extras_require={
"sentry": ["sentry-sdk", "django"],
},
description="Integrate PostHog into any python application.",
description='Integrate PostHog into any python application.',
long_description=long_description,
classifiers=[
"Development Status :: 5 - Production/Stable",
+48 -59
View File
@@ -1,39 +1,42 @@
import analytics
import argparse
import json
import logging
import posthog
__name__ = "simulator.py"
__version__ = "0.0.1"
__description__ = "scripting simulator"
__name__ = 'simulator.py'
__version__ = '0.0.1'
__description__ = 'scripting simulator'
def json_hash(str):
if str:
return json.loads(str)
# posthog -method=<method> -posthog-write-key=<posthogWriteKey> [options]
# analytics -method=<method> -posthog-write-key=<posthogWriteKey> [options]
parser = argparse.ArgumentParser(description="send a posthog message")
parser = argparse.ArgumentParser(description='send a posthog message')
parser.add_argument("--writeKey", help="the posthog writeKey")
parser.add_argument("--type", help="The posthog message type")
parser.add_argument('--writeKey', help='the posthog writeKey')
parser.add_argument('--type', help='The posthog message type')
parser.add_argument("--distinct_id", help="the user id to send the event as")
parser.add_argument("--anonymousId", help="the anonymous user id to send the event as")
parser.add_argument("--context", help="additional context for the event (JSON-encoded)")
parser.add_argument('--distinct_id', help='the user id to send the event as')
parser.add_argument(
'--anonymousId', help='the anonymous user id to send the event as')
parser.add_argument(
'--context', help='additional context for the event (JSON-encoded)')
parser.add_argument("--event", help="the event name to send with the event")
parser.add_argument("--properties", help="the event properties to send (JSON-encoded)")
parser.add_argument('--event', help='the event name to send with the event')
parser.add_argument(
'--properties', help='the event properties to send (JSON-encoded)')
parser.add_argument("--name", help="name of the screen or page to send with the message")
parser.add_argument(
'--name', help='name of the screen or page to send with the message')
parser.add_argument("--traits", help="the identify/group traits to send (JSON-encoded)")
parser.add_argument(
'--traits', help='the identify/group traits to send (JSON-encoded)')
parser.add_argument("--groupId", help="the group id")
parser.add_argument('--groupId', help='the group id')
options = parser.parse_args()
@@ -42,69 +45,55 @@ def failed(status, msg):
raise Exception(msg)
def capture():
posthog.capture(
options.distinct_id,
options.event,
anonymous_id=options.anonymousId,
properties=json_hash(options.properties),
context=json_hash(options.context),
)
def track():
analytics.track(options.distinct_id, options.event, anonymous_id=options.anonymousId,
properties=json_hash(options.properties), context=json_hash(options.context))
def page():
posthog.page(
options.distinct_id,
name=options.name,
anonymous_id=options.anonymousId,
properties=json_hash(options.properties),
context=json_hash(options.context),
)
analytics.page(options.distinct_id, name=options.name, anonymous_id=options.anonymousId,
properties=json_hash(options.properties), context=json_hash(options.context))
def screen():
analytics.screen(options.distinct_id, name=options.name, anonymous_id=options.anonymousId,
properties=json_hash(options.properties), context=json_hash(options.context))
def identify():
posthog.identify(
options.distinct_id,
anonymous_id=options.anonymousId,
traits=json_hash(options.traits),
context=json_hash(options.context),
)
analytics.identify(options.distinct_id, anonymous_id=options.anonymousId,
traits=json_hash(options.traits), context=json_hash(options.context))
def set_once():
posthog.set_once(
options.distinct_id,
properties=json_hash(options.traits),
context=json_hash(options.context),
)
def set():
posthog.set(
options.distinct_id,
properties=json_hash(options.traits),
context=json_hash(options.context),
)
def group():
analytics.group(options.distinct_id, options.groupId, json_hash(options.traits),
json_hash(options.context), anonymous_id=options.anonymousId)
def unknown():
print()
posthog.api_key = options.writeKey
posthog.on_error = failed
posthog.debug = True
analytics.api_key = options.writeKey
analytics.on_error = failed
analytics.debug = True
log = logging.getLogger("posthog")
log = logging.getLogger('posthog')
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
log.addHandler(ch)
switcher = {"capture": capture, "page": page, "identify": identify, "set_once": set_once, "set": set}
switcher = {
"track": track,
"page": page,
"screen": screen,
"identify": identify,
"group": group
}
func = switcher.get(options.type)
if func:
func()
posthog.shutdown()
analytics.shutdown()
else:
print("Invalid Message Type " + options.type)