Compare commits

...
13 Commits
Author SHA1 Message Date
Dylan MartinandGitHub 722c88701b feat(flags): make the sendFeatureFlags parameter more declarative and ergonomic (#283) 2025-07-22 15:20:54 -07:00
Radu RaiceaandGitHub 6ab2856f8d feat(llmo): Use default PH client for LangChain (#293)
* feat(llmo): Use default PH client for langchain

* chore: Run formatter

* feat: Test the CallbackHandler without any PH client

* chore: Run formatter
2025-07-22 14:09:47 -07:00
7a8b09123c feat(llmo): Make it optional to pass posthog client (#291)
Co-authored-by: Peter Kirkham <peter@posthog.com>
2025-07-22 06:53:47 +00:00
David NewellandGitHub da09639428 fix: capture django processed exceptions (#287) 2025-07-16 21:54:01 +02:00
Vincent (Wen Yu) GeandGitHub 6a271026d1 Init reference doc generation (#280) 2025-07-15 13:53:13 -04:00
Phil HaackandGitHub 6d9247960f fix: Ignore new flag filter type in local evaluation (#285) 2025-07-11 16:33:23 +00:00
Dylan MartinandGitHub c4e09cdd40 feat(flags): decouple local evaluation from personal API keys; support decrypting remote config payloads without relying on the feature flags poller (#282) 2025-07-10 08:03:37 -07:00
Oliver BrowneandGitHub c61236b26a fix: add middleware setting for custom client (#281)
* Add middleware setting for custom client

* mypy

* comment
2025-07-09 17:06:20 +03:00
Dylan MartinandGitHub b965332698 feat(flags): add a flag_fallback_cache that tracks feature flag evaluation results and uses them as fallback values whenever the /flags API isn't available (#275) 2025-07-07 07:13:35 +00:00
Oliver BrowneandGitHub 4739945a82 fix: default send_feature_flags false for capture_exception (#278)
* default send_feature_flags false

* bump version
2025-07-02 22:21:25 +03:00
Oliver BrowneandGitHub 50ab10c858 fix(err): permit disabling person processing (#277)
* whoops

* bump version
2025-07-01 15:26:57 +03:00
Oliver BrowneGitHubgreptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>David Newell
37bd30194e feat: prep for 6.0.0, bunch of breaking changes (#273)
* alright

* fix exports

* Update posthog/test/test_before_send.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Update posthog/__init__.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* Update posthog/__init__.py

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

* fix tests after rebase

* update examples

* update mypy baseline

* Revert "update mypy baseline"

This reverts commit da395dd7cc075d1f5b1c03d748544e4abcd75bb9.

* try again

* whatever

* request user

* fix middleware

* getattr is_authenticated

* allow using custom client for exception capture

* update comments

* fix circular import

* type arguments, use TypedDict

* fix setup

* mypy

* whoops

* ok

* further mypy

* mypy sync

* docs and fixes

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>
Co-authored-by: David Newell <david@posthog.com>
2025-06-27 14:57:02 +03:00
Lucas RicoyandGitHub b41dc8568e docs: update release details on readme (#272) 2025-06-21 00:38:18 +01:00
44 changed files with 6244 additions and 4313 deletions
+1
View File
@@ -18,3 +18,4 @@ posthog-analytics
pyrightconfig.json
.env
.DS_Store
posthog-python-references.json
+59
View File
@@ -1,3 +1,62 @@
# 6.3.0 - 2025-07-22
- feat: Enhanced `send_feature_flags` parameter to accept `SendFeatureFlagsOptions` object for declarative control over local/remote evaluation and custom properties
# 6.2.1 - 2025-07-21
- feat: make `posthog_client` an optional argument in PostHog AI providers wrappers (`posthog.ai.*`), intuitively using the default client as the default
# 6.1.1 - 2025-07-16
- fix: correctly capture exceptions processed by Django from views or middleware
# 6.1.0 - 2025-07-10
- feat: decouple feature flag local evaluation from personal API keys; support decrypting remote config payloads without relying on the feature flags poller
# 6.0.4 - 2025-07-09
- fix: add POSTHOG_MW_CLIENT setting to django middleware, to support custom clients for exception capture.
# 6.0.3 - 2025-07-07
- feat: add a feature flag evaluation cache (local storage or redis) to support returning flag evaluations when the service is down
# 6.0.2 - 2025-07-02
- fix: send_feature_flags changed to default to false in `Client::capture_exception`
# 6.0.1
- fix: response `$process_person_profile` property when passed to capture
# 6.0.0
This release contains a number of major breaking changes:
- feat: make distinct_id an optional parameter in posthog.capture and related functions
- feat: make capture and related functions return `Optional[str]`, which is the UUID of the sent event, if it was sent
- fix: remove `identify` (prefer `posthog.set()`), and `page` and `screen` (prefer `posthog.capture()`)
- fix: delete exception-capture specific integrations module. Prefer the general-purpose django middleware as a replacement for the django `Integration`.
To migrate to this version, you'll mostly just need to switch to using named keyword arguments, rather than positional ones. For example:
```python
# Old calling convention
posthog.capture("user123", "button_clicked", {"button_id": "123"})
# New calling convention
posthog.capture(distinct_id="user123", event="button_clicked", properties={"button_id": "123"})
# Better pattern
with posthog.new_context():
posthog.identify_context("user123")
# The event name is the first argument, and can be passed positionally, or as a keyword argument in a later position
posthog.capture("button_pressed")
```
Generally, arguments are now appropriately typed, and docstrings have been updated. If something is unclear, please open an issue, or submit a PR!
# 5.4.0 - 2025-06-20
- feat: add support to session_id context on page method
+4 -2
View File
@@ -32,7 +32,7 @@ We recommend using [uv](https://docs.astral.sh/uv/). It's super fast.
```bash
uv python install 3.9.19
uv python pin 3.9.19
uv venv env
uv venv
source env/bin/activate
uv sync --extra dev --extra test
pre-commit install
@@ -45,7 +45,9 @@ Assuming you have a [local version of PostHog](https://posthog.com/docs/developi
### Releasing Versions
Updates 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`.
Updates are released automatically using GitHub Actions when `version.py` is updated on `master`. After bumping `version.py` in `master` and adding to `CHANGELOG.md`, the [release workflow](https://github.com/PostHog/posthog-python/blob/master/.github/workflows/release.yaml) will automatically trigger and deploy the new version.
If you need to check the latest runs or manually trigger a release, you can 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`.
### Testing changes locally with the PostHog app
Executable
+8
View File
@@ -0,0 +1,8 @@
#!/usr/bin/env bash
#/ Usage: bin/docs
#/ Description: Generate documentation for the PostHog Python SDK
source bin/helpers/_utils.sh
set_source_and_root_dir
ensure_virtual_env
exec python3 "$(dirname "$0")/docs_scripts/generate_json_schemas.py" "$@"
+81
View File
@@ -0,0 +1,81 @@
"""
Constants for PostHog Python SDK documentation generation.
"""
from typing import Dict, Union
# Types that are built-in to Python and don't need to be documented
NO_DOCS_TYPES = [
"Client",
"any",
"int",
"float",
"bool",
"dict",
"list",
"str",
"tuple",
"set",
"frozenset",
"bytes",
"bytearray",
"memoryview",
"range",
"slice",
"complex",
"Union",
"Optional",
"Any",
"Callable",
"Type",
"TypeVar",
"Generic",
"Literal",
"ClassVar",
"Final",
"Annotated",
"NotRequired",
"Required",
"None",
"NoneType",
"object",
"Unpack",
"BaseException",
"Exception",
]
# Documentation generation metadata
DOCUMENTATION_METADATA = {
"hogRef": "0.1",
"slugPrefix": "posthog-python",
"specUrl": "https://github.com/PostHog/posthog-python",
}
# Docstring parsing patterns for new format
DOCSTRING_PATTERNS = {
"examples_section": r"Examples:\s*\n(.*?)(?=\n\s*\n\s*Category:|\Z)",
"args_section": r"Args:\s*\n(.*?)(?=\n\s*\n\s*Examples:|\n\s*\n\s*Details:|\n\s*\n\s*Category:|\Z)",
"details_section": r"Details:\s*\n(.*?)(?=\n\s*\n\s*Examples:|\n\s*\n\s*Category:|\Z)",
"category_section": r"Category:\s*\n\s*(.+?)\s*(?:\n|$)",
"code_block": r"```(?:python)?\n(.*?)```",
"param_description": r"^\s*{param_name}:\s*(.+?)(?=\n\s*\w+:|\Z)",
"args_marker": r"\n\s*Args:\s*\n",
"examples_marker": r"\n\s*Examples:\s*\n",
"details_marker": r"\n\s*Details:\s*\n",
"category_marker": r"\n\s*Category:\s*\n",
}
# Output file configuration
OUTPUT_CONFIG: Dict[str, Union[str, int]] = {
"output_dir": ".",
"filename": "posthog-python-references.json",
"indent": 2,
}
# Documentation structure defaults
DOC_DEFAULTS = {
"showDocs": True,
"releaseTag": "public",
"return_type_void": "None",
"max_optional_params": 3,
}
+472
View File
@@ -0,0 +1,472 @@
#!/usr/bin/env python3
"""
Generate comprehensive SDK documentation JSON from PostHog Python SDK.
This script inspects the code and docstrings to create documentation in the specified format.
"""
import json
import inspect
import re
from dataclasses import is_dataclass, fields
from typing import get_origin, get_args, Union
from textwrap import dedent
from doc_constant import (
NO_DOCS_TYPES,
DOCUMENTATION_METADATA,
DOCSTRING_PATTERNS,
OUTPUT_CONFIG,
DOC_DEFAULTS,
)
import os
def extract_examples_from_docstring(docstring: str) -> list:
"""Extract code examples from docstring."""
if not docstring:
return []
examples = []
# Look for Examples section in the new format
examples_section_match = re.search(
DOCSTRING_PATTERNS["examples_section"], docstring, re.DOTALL
)
if examples_section_match:
examples_content = examples_section_match.group(1).strip()
# Extract code blocks from the Examples section
code_blocks = re.findall(
DOCSTRING_PATTERNS["code_block"], examples_content, re.DOTALL
)
for i, code_block in enumerate(code_blocks):
# Remove common leading whitespace while preserving relative indentation
code = dedent(code_block).strip()
# Extract name from first comment line if present
lines = code.split("\n")
name = f"Example {i + 1}" # Default fallback
if lines and lines[0].strip().startswith("#"):
# Extract name from first comment, keep the comment in the code
comment_text = lines[0].strip()[1:].strip()
if comment_text:
name = comment_text
examples.append({"id": f"example_{i + 1}", "name": name, "code": code})
return examples
def extract_details_from_docstring(docstring: str) -> str:
"""Extract details section from docstring."""
if not docstring:
return ""
# Look for Details section
details_match = re.search(
DOCSTRING_PATTERNS["details_section"], docstring, re.DOTALL
)
if details_match:
details_content = details_match.group(1).strip()
# Clean up formatting
return details_content.replace("\n", " ")
return ""
def parse_docstring_tags(docstring: str) -> dict:
"""Parse tags from docstring Category section."""
if not docstring:
return {}
tags = {}
# Extract Category section
category_match = re.search(DOCSTRING_PATTERNS["category_section"], docstring)
if category_match:
category_value = category_match.group(1).strip()
tags["category"] = category_value
return tags
def extract_description_from_docstring(docstring: str) -> str:
"""Extract main description from docstring."""
if not docstring:
return ""
# Clean up the docstring
cleaned = dedent(docstring).strip()
# Find the end of the description by looking for first section marker
# Check for Args:, Examples:, Details:, or Category: sections
section_patterns = [
DOCSTRING_PATTERNS["args_marker"],
DOCSTRING_PATTERNS["examples_marker"],
DOCSTRING_PATTERNS["details_marker"],
DOCSTRING_PATTERNS["category_marker"],
]
end_pos = len(cleaned)
for pattern in section_patterns:
match = re.search(pattern, cleaned)
if match:
end_pos = min(end_pos, match.start())
# Extract description up to the first section marker
description = cleaned[:end_pos].strip()
# Remove one level of \n since it will be rendered as markdown
# and \n will be padded in later steps
description = description.replace("\n", " ")
return description
def get_type_name(type_annotation) -> str:
"""Convert type annotation to string name."""
if type_annotation is None or type_annotation is type(None):
return "any"
# Handle typing constructs
origin = get_origin(type_annotation)
if origin is not None:
# Handle Union types (including Optional)
if origin is Union:
args = get_args(type_annotation)
if len(args) == 2 and type(None) in args:
# This is Optional[Type] - get the non-None type
non_none_type = next(arg for arg in args if arg is not type(None))
return f"Optional[{get_type_name(non_none_type)}]"
else:
# Regular Union - list all types
type_names = [get_type_name(arg) for arg in args]
return f"Union[{', '.join(type_names)}]"
# Handle other generic types (List, Dict, etc.)
origin_name = getattr(origin, "__name__", str(origin))
args = get_args(type_annotation)
if args:
arg_names = [get_type_name(arg) for arg in args]
return f"{origin_name}[{', '.join(arg_names)}]"
else:
return origin_name
# Handle regular types
elif hasattr(type_annotation, "__name__"):
return type_annotation.__name__
else:
return str(type_annotation)
def analyze_parameter(param: inspect.Parameter, docstring: str = "") -> dict:
"""Analyze a function parameter and return its documentation."""
# Determine if parameter is optional (has default value)
is_optional = param.default == inspect.Parameter.empty
# Get the type annotation
type_annotation = param.annotation
param_type = "any"
if type_annotation != inspect.Parameter.empty:
# Handle Union/Optional types first
origin = get_origin(type_annotation)
if origin is Union:
args = get_args(type_annotation)
if len(args) == 2 and type(None) in args:
# This is Optional[Type]
non_none_type = next(arg for arg in args if arg is not type(None))
param_type = get_type_name(non_none_type)
is_optional = True
else:
# Other Union types, use first type
param_type = get_type_name(args[0]) if args else "any"
else:
param_type = get_type_name(type_annotation)
elif param.default != inspect.Parameter.empty:
# No type annotation, but has default value - infer type from default
param_type = get_type_name(type(param.default))
# Extract parameter description from Args section
param_description = f"Parameter: {param.name}"
if docstring:
# Look for Args section and extract description for this parameter
args_section_match = re.search(
DOCSTRING_PATTERNS["args_section"], docstring, re.DOTALL
)
if args_section_match:
args_content = args_section_match.group(1)
# Look for the parameter description
param_pattern = DOCSTRING_PATTERNS["param_description"].format(
param_name=re.escape(param.name)
)
param_match = re.search(
param_pattern, args_content, re.MULTILINE | re.DOTALL
)
if param_match:
param_description = param_match.group(1).strip().replace("\n", " ")
param_info = {
"name": param.name,
"description": param_description,
"isOptional": is_optional,
"type": param_type,
}
return param_info
def analyze_function(func, name: str) -> dict:
"""Analyze a function and return its documentation."""
try:
sig = inspect.signature(func)
docstring = inspect.getdoc(func) or ""
# Skip functions with empty docstrings
if not docstring.strip():
return {}
# Extract parameters (excluding 'self')
params = []
for param_name, param in sig.parameters.items():
if param_name != "self":
params.append(analyze_parameter(param, docstring))
# Special handling for constructor
display_name = name
if name == "__init__":
display_name = func.__qualname__.split(".")[0]
# Parse tags from docstring
tags = parse_docstring_tags(docstring)
category = tags.get("category", None)
# Extract description
description = extract_description_from_docstring(docstring)
# Skip if no meaningful description
if not description.strip():
return {}
# Extract details section (only if it exists)
details = extract_details_from_docstring(docstring)
# Get examples from docstring, do not generate fallback examples
examples = extract_examples_from_docstring(docstring)
# If no examples, do not include the examples key or set to empty list
result = {
"id": name,
"title": display_name,
"description": description,
"details": details,
"category": category,
"params": params,
"showDocs": DOC_DEFAULTS["showDocs"],
"releaseTag": DOC_DEFAULTS["releaseTag"],
"returnType": {
"id": "return_type",
"name": get_type_name(sig.return_annotation)
if sig.return_annotation != inspect.Signature.empty
else DOC_DEFAULTS["return_type_void"],
},
}
if examples:
result["examples"] = examples
return result
except Exception as e:
print(f"Error analyzing function {name}: {e}")
return {}
def analyze_class(cls) -> dict:
"""Analyze a class and return its documentation."""
class_doc = inspect.getdoc(cls) or f"Class: {cls.__name__}"
# Get all public methods and constructor
functions = []
for method_name in dir(cls):
if method_name.startswith("_") and method_name != "__init__":
continue
method = getattr(cls, method_name)
if callable(method):
func_info = analyze_function(method, method_name)
if func_info: # Only add if not None (empty docstring check)
functions.append(func_info)
return {
"id": cls.__name__,
"title": cls.__name__,
"description": extract_description_from_docstring(class_doc),
"functions": functions,
}
def analyze_type(cls) -> dict:
"""Analyze a type/dataclass and return its documentation."""
type_info = {
"id": cls.__name__,
"name": cls.__name__,
"path": f"{cls.__module__}.{cls.__name__}",
"properties": [],
"example": "",
}
if is_dataclass(cls):
# Handle dataclass
for field in fields(cls):
prop = {
"name": field.name,
"type": get_type_name(field.type),
"description": f"Field: {field.name}",
}
type_info["properties"].append(prop)
elif hasattr(cls, "__annotations__"):
# Handle TypedDict or annotated class
for field_name, field_type in cls.__annotations__.items():
prop = {
"name": field_name,
"type": get_type_name(field_type),
"description": f"Field: {field_name}",
}
type_info["properties"].append(prop)
return type_info
def generate_sdk_documentation():
"""Generate complete SDK documentation in the requested format."""
# Import PostHog components
import posthog
from posthog.client import Client
import posthog.types as types_module
import posthog.args as args_module
from posthog.version import VERSION
# Main SDK info
sdk_info = {
"version": VERSION,
"id": "posthog-python",
"title": "PostHog Python SDK",
"description": "Integrate PostHog into any python application.",
"slugPrefix": DOCUMENTATION_METADATA["slugPrefix"],
"specUrl": DOCUMENTATION_METADATA["specUrl"],
}
# Collect types
types_list = []
# Types from posthog.types
for name in dir(types_module):
obj = getattr(types_module, name)
if inspect.isclass(obj) and not name.startswith("_"):
try:
type_info = analyze_type(obj)
types_list.append(type_info)
except Exception as e:
print(f"Error analyzing type {name}: {e}")
# Types from posthog.args
for name in dir(args_module):
obj = getattr(args_module, name)
if inspect.isclass(obj) and not name.startswith("_"):
try:
type_info = analyze_type(obj)
types_list.append(type_info)
except Exception as e:
print(f"Error analyzing type {name}: {e}")
# Collect classes
classes_list = []
# Main PostHog class (renamed from Client)
client_class = analyze_class(Client)
client_class["id"] = "PostHog"
client_class["title"] = "PostHog"
classes_list.append(client_class)
# Global module functions (functions callable as posthog.function_name)
global_functions = []
for func_name in dir(posthog):
# Skip private functions and non-callables
if func_name.startswith("_") or not callable(getattr(posthog, func_name)):
continue
func = getattr(posthog, func_name)
# Only include functions actually defined in the posthog module (not imported)
# and exclude class references
if (
func_name not in ["Client", "Posthog"]
and hasattr(func, "__module__")
and func.__module__ == "posthog"
):
try:
func_info = analyze_function(func, func_name)
if func_info: # Only add if not None (has proper docstring)
global_functions.append(func_info)
except Exception:
continue
# Add global functions as a "class"
if global_functions:
classes_list.append(
{
"id": "PostHogModule",
"title": "PostHog Module Functions",
"description": "Global functions available in the PostHog module",
"functions": global_functions,
}
)
# Create the final structure
result = {
"id": "posthog-python",
"hogRef": DOCUMENTATION_METADATA["hogRef"],
"info": sdk_info,
"noDocsTypes": NO_DOCS_TYPES,
"types": types_list,
"classes": classes_list,
}
return result
if __name__ == "__main__":
print("Generating PostHog Python SDK documentation...")
try:
documentation = generate_sdk_documentation()
# Write to file
output_file = os.path.join(
str(OUTPUT_CONFIG["output_dir"]), str(OUTPUT_CONFIG["filename"])
)
with open(output_file, "w") as f:
json.dump(documentation, f, indent=int(OUTPUT_CONFIG["indent"]))
print(f"✓ Generated {output_file}")
# Print summary
types_count = len(documentation["types"])
classes_count = len(documentation["classes"])
total_functions = sum(len(cls["functions"]) for cls in documentation["classes"])
print("📊 Documentation Summary:")
print(f"{types_count} types documented")
print(f"{classes_count} classes documented")
print(f"{total_functions} functions documented")
no_docs = documentation["noDocsTypes"]
if no_docs:
print(
f"{len(no_docs)} types without documentation: {', '.join(no_docs[:5])}{'...' if len(no_docs) > 5 else ''}"
)
except Exception as e:
print(f"❌ Error generating documentation: {e}")
import traceback
traceback.print_exc()
+14 -10
View File
@@ -41,9 +41,9 @@ print(
# Capture an event
posthog.capture(
"distinct_id",
"event",
{"property1": "value", "property2": "value"},
distinct_id="distinct_id",
properties={"property1": "value", "property2": "value"},
send_feature_flags=True,
)
@@ -65,31 +65,35 @@ exit()
posthog.alias("distinct_id", "new_distinct_id")
posthog.capture(
"new_distinct_id", "event2", {"property1": "value", "property2": "value"}
"event2",
distinct_id="new_distinct_id",
properties={"property1": "value", "property2": "value"},
)
posthog.capture(
"new_distinct_id",
"event-with-groups",
{"property1": "value", "property2": "value"},
distinct_id="new_distinct_id",
properties={"property1": "value", "property2": "value"},
groups={"company": "id:5"},
)
# # Add properties to the person
posthog.identify("new_distinct_id", {"email": "something@something.com"})
posthog.set(
distinct_id="new_distinct_id", properties={"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(distinct_id="new_distinct_id", properties={"self_serve_signup": True})
posthog.set_once(
"new_distinct_id", {"self_serve_signup": False}
distinct_id="new_distinct_id", properties={"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"})
posthog.set(distinct_id="new_distinct_id", properties={"current_browser": "Chrome"})
posthog.set(distinct_id="new_distinct_id", properties={"current_browser": "Firefox"})
# #############################################################################
+16 -7
View File
@@ -22,16 +22,25 @@ posthog/client.py:0: error: Library stubs not installed for "six" [import-untyp
posthog/client.py:0: note: Hint: "python3 -m pip install types-six"
posthog/client.py:0: error: Name "queue" already defined (by an import) [no-redef]
posthog/client.py:0: error: Need type annotation for "queue" [var-annotated]
posthog/client.py:0: error: Item "None" of "Any | None" has no attribute "get" [union-attr]
simulator.py:0: error: Unexpected keyword argument "anonymous_id" for "capture" [call-arg]
posthog/__init__.py:0: note: "capture" defined here
simulator.py:0: error: Unexpected keyword argument "anonymous_id" for "identify" [call-arg]
posthog/__init__.py:0: note: "identify" defined here
simulator.py:0: error: Unexpected keyword argument "traits" for "identify" [call-arg]
posthog/__init__.py:0: note: "identify" defined here
posthog/client.py:0: error: Incompatible types in assignment (expression has type "Any | list[Any]", variable has type "None") [assignment]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "dict[Any, Any]", variable has type "None") [assignment]
posthog/client.py:0: error: "None" has no attribute "__iter__" (not iterable) [attr-defined]
posthog/client.py:0: error: Statement is unreachable [unreachable]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "Any | dict[Any, Any]", variable has type "None") [assignment]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "Any | dict[Any, Any]", variable has type "None") [assignment]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "dict[Never, Never]", variable has type "None") [assignment]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "dict[Never, Never]", variable has type "None") [assignment]
posthog/client.py:0: error: Right operand of "and" is never evaluated [unreachable]
posthog/client.py:0: error: Incompatible types in assignment (expression has type "Poller", variable has type "None") [assignment]
posthog/client.py:0: error: "None" has no attribute "start" [attr-defined]
posthog/client.py:0: error: "None" has no attribute "get" [attr-defined]
posthog/client.py:0: error: Statement is unreachable [unreachable]
posthog/client.py:0: error: Statement is unreachable [unreachable]
example.py:0: error: Statement is unreachable [unreachable]
posthog/ai/utils.py:0: error: Need type annotation for "output" (hint: "output: list[<type>] = ...") [var-annotated]
posthog/ai/utils.py:0: error: Function "builtins.any" is not valid as a type [valid-type]
posthog/ai/utils.py:0: note: Perhaps you meant "typing.Any" instead of "any"?
posthog/ai/utils.py:0: error: Function "builtins.any" is not valid as a type [valid-type]
posthog/ai/utils.py:0: note: Perhaps you meant "typing.Any" instead of "any"?
posthog/client.py:0: error: Name "urlparse" already defined (possibly by an import) [no-redef]
posthog/client.py:0: error: Name "parse_qs" already defined (possibly by an import) [no-redef]
+397 -309
View File
@@ -1,17 +1,15 @@
import datetime # noqa: F401
import warnings
from typing import Callable, Dict, List, Optional, Tuple # noqa: F401
from typing import Callable, Dict, Optional, Any # noqa: F401
from typing_extensions import Unpack
from posthog.args import OptionalCaptureArgs, OptionalSetArgs, ExceptionArg
from posthog.client import Client
from posthog.exception_capture import Integrations # noqa: F401
from posthog.scopes import (
clear_tags,
get_tags,
new_context,
scoped,
tag,
set_context_session,
identify_context,
from posthog.contexts import (
new_context as inner_new_context,
scoped as inner_scoped,
tag as inner_tag,
set_context_session as inner_set_context_session,
identify_context as inner_identify_context,
)
from posthog.types import FeatureFlag, FlagsAndPayloads
from posthog.version import VERSION
@@ -19,13 +17,109 @@ from posthog.version import VERSION
__version__ = VERSION
"""Context management."""
new_context = new_context
tag = tag
get_tags = get_tags
clear_tags = clear_tags
scoped = scoped
identify_context = identify_context
set_context_session = set_context_session
def new_context(fresh=False, capture_exceptions=True):
"""
Create a new context scope that will be active for the duration of the with block.
Args:
fresh: Whether to start with a fresh context (default: False)
capture_exceptions: Whether to capture exceptions raised within the context (default: True)
Examples:
```python
from posthog import new_context, tag, capture
with new_context():
tag("request_id", "123")
capture("event_name", properties={"property": "value"})
```
Category:
Contexts
"""
return inner_new_context(fresh=fresh, capture_exceptions=capture_exceptions)
def scoped(fresh=False, capture_exceptions=True):
"""
Decorator that creates a new context for the function.
Args:
fresh: Whether to start with a fresh context (default: False)
capture_exceptions: Whether to capture and track exceptions with posthog error tracking (default: True)
Examples:
```python
from posthog import scoped, tag, capture
@scoped()
def process_payment(payment_id):
tag("payment_id", payment_id)
capture("payment_started")
```
Category:
Contexts
"""
return inner_scoped(fresh=fresh, capture_exceptions=capture_exceptions)
def set_context_session(session_id: str):
"""
Set the session ID for the current context.
Args:
session_id: The session ID to associate with the current context and its children
Examples:
```python
from posthog import set_context_session
set_context_session("session_123")
```
Category:
Contexts
"""
return inner_set_context_session(session_id)
def identify_context(distinct_id: str):
"""
Identify the current context with a distinct ID.
Args:
distinct_id: The distinct ID to associate with the current context and its children
Examples:
```python
from posthog import identify_context
identify_context("user_123")
```
Category:
Identification
"""
return inner_identify_context(distinct_id)
def tag(name: str, value: Any):
"""
Add a tag to the current context.
Args:
name: The tag key
value: The tag value
Examples:
```python
from posthog import tag
tag("user_id", "123")
```
Category:
Contexts
"""
return inner_tag(name, value)
"""Settings."""
@@ -44,238 +138,177 @@ feature_flags_request_timeout_seconds = 3 # type: int
super_properties = None # type: Optional[Dict]
# Currently alpha, use at your own risk
enable_exception_autocapture = False # type: bool
exception_autocapture_integrations = [] # type: List[Integrations]
log_captured_exceptions = False # type: bool
# Used to determine in app paths for exception autocapture. Defaults to the current working directory
project_root = None # type: Optional[str]
# Used for our AI observability feature to not capture any prompt or output just usage + metadata
privacy_mode = False # type: bool
# Whether to enable feature flag polling for local evaluation by default. Defaults to True.
# We recommend setting this to False if you are only using the personalApiKey for evaluating remote config payloads via `get_remote_config_payload` and not using local evaluation.
enable_local_evaluation = True # type: bool
default_client = None # type: Optional[Client]
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]
# NOTE - this and following functions take unpacked kwargs because we needed to make
# it impossible to write `posthog.capture(distinct-id, event-name)` - basically, to enforce
# the breaking change made between 5.3.0 and 6.0.0. This decision can be unrolled in later
# versions, without a breaking change, to get back the type information in function signatures
def capture(event: str, **kwargs: Unpack[OptionalCaptureArgs]) -> Optional[str]:
"""
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.
Capture anything a user does within your system.
A `capture` call requires
- `distinct id` which uniquely identifies your user
- `event name` to specify the event
- We recommend using [verb] [noun], like `movie played` or `movie updated` to easily identify what your events mean later on.
Args:
event: The event name to specify the event
**kwargs: Optional arguments including:
distinct_id: Unique identifier for the user
properties: Dict of event properties
timestamp: When the event occurred
groups: Dict of group types and IDs
disable_geoip: Whether to disable GeoIP lookup
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
Details:
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 an event name to specify the event. We recommend using [verb] [noun], like `movie played` or `movie updated` to easily identify what your events mean later on. Capture takes a number of optional arguments, which are defined by the `OptionalCaptureArgs` type.
For example:
```python
posthog.capture('distinct id', 'opened app')
posthog.capture('distinct id', 'movie played', {'movie_id': '123', 'category': 'romcom'})
Examples:
```python
# Context and capture usage
from posthog import new_context, identify_context, tag_context, capture
# Enter a new context (e.g. a request/response cycle, an instance of a background job, etc)
with new_context():
# Associate this context with some user, by distinct_id
identify_context('some user')
posthog.capture('distinct id', 'purchase', groups={'company': 'id:5'})
```
"""
# Capture an event, associated with the context-level distinct ID ('some user')
capture('movie started')
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
# Capture an event associated with some other user (overriding the context-level distinct ID)
capture('movie joined', distinct_id='some-other-user')
# Capture an event with some properties
capture('movie played', properties={'movie_id': '123', 'category': 'romcom'})
# Capture an event with some properties
capture('purchase', properties={'product_id': '123', 'category': 'romcom'})
# Capture an event with some associated group
capture('purchase', groups={'company': 'id:5'})
# Adding a tag to the current context will cause it to appear on all subsequent events
tag_context('some-tag', 'some-value')
capture('another-event') # Will be captured with `'some-tag': 'some-value'` in the properties dict
```
```python
# Set event properties
from posthog import capture
capture(
"user_signed_up",
distinct_id="distinct_id_of_the_user",
properties={
"login_type": "email",
"is_free_trial": "true"
}
)
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,
)
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]
"""
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
For example:
```python
posthog.identify('distinct id', {
'email': 'dwayne@gmail.com',
'name': 'Dwayne Johnson'
})
```
```
Category:
Events
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
return _proxy(
"identify",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
return _proxy("capture", event, **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]
def set(**kwargs: Unpack[OptionalSetArgs]) -> Optional[str]:
"""
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
Details:
This will overwrite previous people property values. Generally operates similar to `capture`, with distinct_id being an optional argument, defaulting to the current context's distinct ID. If there is no context-level distinct ID, and no override distinct_id is passed, this function will do nothing. Context tags are folded into $set properties, so tagging the current context and then calling `set` will cause those tags to be set on the user (unlike capture, which causes them to just be set on the event).
For example:
```python
posthog.set('distinct id', {
'current_browser': 'Chrome',
})
```
Examples:
```python
# Set person properties
from posthog import capture
capture(
'distinct_id',
event='event_name',
properties={
'$set': {'name': 'Max Hedgehog'},
'$set_once': {'initial_url': '/blog'}
}
)
```
Category:
Identification
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
return _proxy(
"set",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
return _proxy("set", **kwargs)
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]
def set_once(**kwargs: Unpack[OptionalSetArgs]) -> Optional[str]:
"""
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
Details:
This will not overwrite previous people property values, unlike `set`. Otherwise, operates in an identical manner to `set`.
For example:
```python
posthog.set_once('distinct id', {
'referred_by': 'friend',
})
```
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
Examples:
```python
# Set property once
from posthog import capture
capture(
'distinct_id',
event='event_name',
properties={
'$set': {'name': 'Max Hedgehog'},
'$set_once': {'initial_url': '/blog'}
}
)
return _proxy(
"set_once",
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
)
```
Category:
Identification
"""
return _proxy("set_once", **kwargs)
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]
# type: (...) -> Optional[str]
"""
Set properties on a group
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
Args:
group_type: Type of your group
group_key: Unique identifier of the group
properties: Properties to set on the group
timestamp: Optional timestamp for the event
uuid: Optional UUID for the event
disable_geoip: Whether to disable GeoIP lookup
For example:
```python
posthog.group_identify('company', 5, {
'employees': 11,
})
```
Examples:
```python
# Group identify
from posthog import group_identify
group_identify('company', 'company_id_in_your_db', {
'name': 'Awesome Inc.',
'employees': 11
})
```
Category:
Identification
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
return _proxy(
"group_identify",
group_type=group_type,
group_key=group_key,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
@@ -285,41 +318,38 @@ def group_identify(
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]
# type: (...) -> Optional[str]
"""
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?"
Associate user behaviour before and after they e.g. register, login, or perform some other identifying action.
In a purely back-end implementation, this means whenever an anonymous user does something, you'll want to send a session ID ([Django](https://stackoverflow.com/questions/526179/in-django-how-can-i-find-out-the-request-session-sessionid-and-use-it-as-a-vari), [Flask](https://stackoverflow.com/questions/15156132/flask-login-how-to-get-session-id)) with the capture call. Then, when that users signs up, you want to do an alias call with the session ID and the newly created user ID.
Args:
previous_id: The unique ID of the user before
distinct_id: The current unique id
timestamp: Optional timestamp for the event
uuid: Optional UUID for the event
disable_geoip: Whether to disable GeoIP lookup
The same concept applies for when a user logs in.
Details:
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?". Particularly useful for associating user behaviour before and after they e.g. register, login, or perform some other identifying action.
An `alias` call requires
- `previous distinct id` the unique ID of the user before
- `distinct id` the current unique id
For example:
```python
posthog.alias('anonymous session id', 'distinct id')
```
Examples:
```python
# Alias user
from posthog import alias
alias(previous_id='distinct_id', distinct_id='alias_id')
```
Category:
Identification
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
return _proxy(
"alias",
previous_id=previous_id,
distinct_id=distinct_id,
context=context,
timestamp=timestamp,
uuid=uuid,
disable_geoip=disable_geoip,
@@ -327,58 +357,32 @@ def alias(
def capture_exception(
exception=None, # type: Optional[BaseException]
distinct_id=None, # type: Optional[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]
**kwargs,
exception: Optional[ExceptionArg] = None,
**kwargs: Unpack[OptionalCaptureArgs],
):
# type: (...) -> Tuple[bool, dict]
"""
capture_exception allows you to capture exceptions that happen in your code. This is useful for debugging and understanding what errors your users are encountering.
This function never raises an exception, even if it fails to send the event.
Capture exceptions that happen in your code.
A `capture_exception` call does not require any fields, but we recommend sending:
- `distinct id` which uniquely identifies your user for which this exception happens
- `exception` to specify the exception to capture. If not provided, the current exception is captured via `sys.exc_info()`
Args:
exception: The exception to capture. If not provided, the current exception is captured via `sys.exc_info()`
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
- remaining `kwargs` will be logged if `log_captured_exceptions` is enabled
Details:
Capture exception is idempotent - if it is called twice with the same exception instance, only a occurrence will be tracked in posthog. This is because, generally, contexts will cause exceptions to be captured automatically. However, to ensure you track an exception, if you catch and do not re-raise it, capturing it manually is recommended, unless you are certain it will have crossed a context boundary (e.g. by existing a `with posthog.new_context():` block already). If the passed exception was raised and caught, the captured stack trace will consist of every frame between where the exception was raised and the point at which it is captured (the "traceback"). If the passed exception was never raised, e.g. if you call `posthog.capture_exception(ValueError("Some Error"))`, the stack trace captured will be the full stack trace at the moment the exception was captured. Note that heavy use of contexts will lead to truncated stack traces, as the exception will be captured by the context entered most recently, which may not be the point you catch the exception for the final time in your code. It's recommended to use contexts sparingly, for this reason. `capture_exception` takes the same set of optional arguments as `capture`.
For example:
```python
try:
1 / 0
except Exception as e:
posthog.capture_exception(e, 'my specific distinct id')
posthog.capture_exception(distinct_id='my specific distinct id')
```
Examples:
```python
# Capture exception
from posthog import capture_exception
try:
risky_operation()
except Exception as e:
capture_exception(e)
```
Category:
Events
"""
if context is not None:
warnings.warn(
"The 'context' parameter is deprecated and will be removed in a future version.",
DeprecationWarning,
stacklevel=2,
)
return _proxy(
"capture_exception",
exception=exception,
distinct_id=distinct_id,
properties=properties,
context=context,
timestamp=timestamp,
uuid=uuid,
groups=groups,
**kwargs,
)
return _proxy("capture_exception", exception=exception, **kwargs)
def feature_enabled(
@@ -395,15 +399,29 @@ def feature_enabled(
"""
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
```
Args:
key: The feature flag key
distinct_id: The user's distinct ID
groups: Groups mapping
person_properties: Person properties
group_properties: Group properties
only_evaluate_locally: Whether to evaluate only locally
send_feature_flag_events: Whether to send feature flag events
disable_geoip: Whether to disable GeoIP lookup
You can call `posthog.load_feature_flags()` before to make sure you're not doing unexpected requests.
Details:
You can call `posthog.load_feature_flags()` before to make sure you're not doing unexpected requests.
Examples:
```python
# Boolean feature flag
from posthog import feature_enabled, get_feature_flag_payload
is_my_flag_enabled = feature_enabled('flag-key', 'distinct_id_of_your_user')
if is_my_flag_enabled:
matched_flag_payload = get_feature_flag_payload('flag-key', 'distinct_id_of_your_user')
```
Category:
Feature flags
"""
return _proxy(
"feature_enabled",
@@ -430,25 +448,30 @@ def get_feature_flag(
) -> Optional[FeatureFlag]:
"""
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"}.
Args:
key: The feature flag key
distinct_id: The user's distinct ID
groups: Groups mapping from group type to group key
person_properties: Person properties
group_properties: Group properties in format { group_type_name: { group_properties } }
only_evaluate_locally: Whether to evaluate only locally
send_feature_flag_events: Whether to send feature flag events
disable_geoip: Whether to disable GeoIP lookup
`group_properties` take the format: { group_type_name: { group_properties } }
Details:
`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: group_properties={"organization": {"name": "PostHog", "employees": 11}}.
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}}
```
Examples:
```python
# Multivariate feature flag
from posthog import get_feature_flag, get_feature_flag_payload
enabled_variant = get_feature_flag('flag-key', 'distinct_id_of_your_user')
if enabled_variant == 'variant-key':
matched_flag_payload = get_feature_flag_payload('flag-key', 'distinct_id_of_your_user')
```
Category:
Feature flags
"""
return _proxy(
"get_feature_flag",
@@ -473,12 +496,26 @@ def get_all_flags(
) -> Optional[dict[str, FeatureFlag]]:
"""
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.
Args:
distinct_id: The user's distinct ID
groups: Groups mapping
person_properties: Person properties
group_properties: Group properties
only_evaluate_locally: Whether to evaluate only locally
disable_geoip: Whether to disable GeoIP lookup
Details:
Flags are key-value pairs where the key is the flag key and the value is the flag variant, or True, or False.
Examples:
```python
# All flags for user
from posthog import get_all_flags
get_all_flags('distinct_id_of_your_user')
```
Category:
Feature flags
"""
return _proxy(
"get_all_flags",
@@ -556,44 +593,94 @@ def get_all_flags_and_payloads(
def feature_flag_definitions():
"""Returns loaded feature flags, if any. Helpful for debugging what flag information you have loaded."""
"""
Returns loaded feature flags.
Details:
Returns loaded feature flags, if any. Helpful for debugging what flag information you have loaded.
Examples:
```python
from posthog import feature_flag_definitions
definitions = feature_flag_definitions()
```
Category:
Feature flags
"""
return _proxy("feature_flag_definitions")
def load_feature_flags():
"""Load feature flag definitions from PostHog."""
"""
Load feature flag definitions from PostHog.
Examples:
```python
from posthog import load_feature_flags
load_feature_flags()
```
Category:
Feature flags
"""
return _proxy("load_feature_flags")
def page(*args, **kwargs):
"""Send a page call."""
_proxy("page", *args, **kwargs)
def screen(*args, **kwargs):
"""Send a screen call."""
_proxy("screen", *args, **kwargs)
def flush():
"""Tell the client to flush."""
"""
Tell the client to flush all queued events.
Examples:
```python
from posthog import flush
flush()
```
Category:
Client management
"""
_proxy("flush")
def join():
"""Block program until the client clears the queue"""
"""
Block program until the client clears the queue. Used during program shutdown. You should use `shutdown()` directly in most cases.
Examples:
```python
from posthog import join
join()
```
Category:
Client management
"""
_proxy("join")
def shutdown():
"""Flush all messages and cleanly shutdown the client"""
"""
Flush all messages and cleanly shutdown the client.
Examples:
```python
from posthog import shutdown
shutdown()
```
Category:
Client management
"""
_proxy("flush")
_proxy("join")
def setup():
def setup() -> Client:
global default_client
if not default_client:
if not api_key:
raise ValueError("API key is required")
default_client = Client(
api_key,
host=host,
@@ -602,7 +689,6 @@ def setup():
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,
@@ -613,13 +699,15 @@ def setup():
# or deprecate this proxy option fully (it's already in the process of deprecation, no new clients should be using this method since like 5-6 months)
enable_exception_autocapture=enable_exception_autocapture,
log_captured_exceptions=log_captured_exceptions,
exception_autocapture_integrations=exception_autocapture_integrations,
enable_local_evaluation=enable_local_evaluation,
)
# always set incase user changes it
default_client.disabled = disabled
default_client.debug = debug
return default_client
def _proxy(method, *args, **kwargs):
"""Create an analytics client if one doesn't exist and send to it."""
+4 -3
View File
@@ -8,7 +8,7 @@ except ImportError:
import time
import uuid
from typing import Any, Dict, Optional
from typing import Any, Dict, Optional, cast
from posthog.ai.utils import (
call_llm_and_track_usage,
@@ -17,6 +17,7 @@ from posthog.ai.utils import (
with_privacy_mode,
)
from posthog.client import Client as PostHogClient
from posthog import setup
class Anthropic(anthropic.Anthropic):
@@ -26,14 +27,14 @@ class Anthropic(anthropic.Anthropic):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
posthog_client: PostHog client for tracking usage
**kwargs: Additional arguments passed to the Anthropic client
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = WrappedMessages(self)
+3 -2
View File
@@ -10,6 +10,7 @@ import time
import uuid
from typing import Any, Dict, Optional
from posthog import setup
from posthog.ai.utils import (
call_llm_and_track_usage_async,
get_model_params,
@@ -26,14 +27,14 @@ class AsyncAnthropic(anthropic.AsyncAnthropic):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
posthog_client: PostHog client for tracking usage
**kwargs: Additional arguments passed to the Anthropic client
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = AsyncWrappedMessages(self)
+11 -8
View File
@@ -5,9 +5,12 @@ except ImportError:
"Please install the Anthropic SDK to use this feature: 'pip install anthropic'"
)
from typing import Optional
from posthog.ai.anthropic.anthropic import WrappedMessages
from posthog.ai.anthropic.anthropic_async import AsyncWrappedMessages
from posthog.client import Client as PostHogClient
from posthog import setup
class AnthropicBedrock(anthropic.AnthropicBedrock):
@@ -17,9 +20,9 @@ class AnthropicBedrock(anthropic.AnthropicBedrock):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = WrappedMessages(self)
@@ -30,9 +33,9 @@ class AsyncAnthropicBedrock(anthropic.AsyncAnthropicBedrock):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = AsyncWrappedMessages(self)
@@ -43,9 +46,9 @@ class AnthropicVertex(anthropic.AnthropicVertex):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = WrappedMessages(self)
@@ -56,7 +59,7 @@ class AsyncAnthropicVertex(anthropic.AsyncAnthropicVertex):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
self.messages = AsyncWrappedMessages(self)
+10 -5
View File
@@ -10,6 +10,7 @@ except ImportError:
"Please install the Google Gemini SDK to use this feature: 'pip install google-genai'"
)
from posthog import setup
from posthog.ai.utils import (
call_llm_and_track_usage,
get_model_params,
@@ -36,6 +37,8 @@ class Client:
)
"""
_ph_client: PostHogClient
def __init__(
self,
api_key: Optional[str] = None,
@@ -56,12 +59,14 @@ class Client:
posthog_groups: Default groups for all calls (can be overridden per call)
**kwargs: Additional arguments (for future compatibility)
"""
if posthog_client is None:
self._ph_client = posthog_client or setup()
if self._ph_client is None:
raise ValueError("posthog_client is required for PostHog tracking")
self.models = Models(
api_key=api_key,
posthog_client=posthog_client,
posthog_client=self._ph_client,
posthog_distinct_id=posthog_distinct_id,
posthog_properties=posthog_properties,
posthog_privacy_mode=posthog_privacy_mode,
@@ -97,10 +102,10 @@ class Models:
posthog_groups: Default groups for all calls
**kwargs: Additional arguments (for future compatibility)
"""
if posthog_client is None:
raise ValueError("posthog_client is required for PostHog tracking")
self._ph_client = posthog_client or setup()
self._ph_client = posthog_client
if self._ph_client is None:
raise ValueError("posthog_client is required for PostHog tracking")
# Store default PostHog settings
self._default_distinct_id = posthog_distinct_id
+14 -15
View File
@@ -33,7 +33,7 @@ from langchain_core.messages import (
from langchain_core.outputs import ChatGeneration, LLMResult
from pydantic import BaseModel
from posthog import default_client
from posthog import setup
from posthog.ai.utils import get_model_params, with_privacy_mode
from posthog.client import Client
@@ -81,10 +81,10 @@ class CallbackHandler(BaseCallbackHandler):
The PostHog LLM observability callback handler for LangChain.
"""
_client: Client
_ph_client: Client
"""PostHog client instance."""
_distinct_id: Optional[Union[str, int, float, UUID]]
_distinct_id: Optional[Union[str, int, UUID]]
"""Distinct ID of the user to associate the trace with."""
_trace_id: Optional[Union[str, int, float, UUID]]
@@ -112,7 +112,7 @@ class CallbackHandler(BaseCallbackHandler):
self,
client: Optional[Client] = None,
*,
distinct_id: Optional[Union[str, int, float, UUID]] = None,
distinct_id: Optional[Union[str, int, UUID]] = None,
trace_id: Optional[Union[str, int, float, UUID]] = None,
properties: Optional[Dict[str, Any]] = None,
privacy_mode: bool = False,
@@ -127,10 +127,7 @@ class CallbackHandler(BaseCallbackHandler):
privacy_mode: Whether to redact the input and output of the trace.
groups: Optional additional PostHog groups to use for the trace.
"""
posthog_client = client or default_client
if posthog_client is None:
raise ValueError("PostHog client is required")
self._client = posthog_client
self._ph_client = client or setup()
self._distinct_id = distinct_id
self._trace_id = trace_id
self._properties = properties or {}
@@ -481,7 +478,7 @@ class CallbackHandler(BaseCallbackHandler):
event_properties = {
"$ai_trace_id": trace_id,
"$ai_input_state": with_privacy_mode(
self._client, self._privacy_mode, run.input
self._ph_client, self._privacy_mode, run.input
),
"$ai_latency": run.latency,
"$ai_span_name": run.name,
@@ -497,13 +494,13 @@ class CallbackHandler(BaseCallbackHandler):
event_properties["$ai_is_error"] = True
elif outputs is not None:
event_properties["$ai_output_state"] = with_privacy_mode(
self._client, self._privacy_mode, outputs
self._ph_client, self._privacy_mode, outputs
)
if self._distinct_id is None:
event_properties["$process_person_profile"] = False
self._client.capture(
self._ph_client.capture(
distinct_id=self._distinct_id or run_id,
event=event_name,
properties=event_properties,
@@ -550,14 +547,16 @@ class CallbackHandler(BaseCallbackHandler):
"$ai_provider": run.provider,
"$ai_model": run.model,
"$ai_model_parameters": run.model_params,
"$ai_input": with_privacy_mode(self._client, self._privacy_mode, run.input),
"$ai_input": with_privacy_mode(
self._ph_client, self._privacy_mode, run.input
),
"$ai_http_status": 200,
"$ai_latency": run.latency,
"$ai_base_url": run.base_url,
}
if run.tools:
event_properties["$ai_tools"] = with_privacy_mode(
self._client,
self._ph_client,
self._privacy_mode,
run.tools,
)
@@ -589,7 +588,7 @@ class CallbackHandler(BaseCallbackHandler):
_extract_raw_esponse(generation) for generation in generation_result
]
event_properties["$ai_output_choices"] = with_privacy_mode(
self._client, self._privacy_mode, completions
self._ph_client, self._privacy_mode, completions
)
if self._properties:
@@ -598,7 +597,7 @@ class CallbackHandler(BaseCallbackHandler):
if self._distinct_id is None:
event_properties["$process_person_profile"] = False
self._client.capture(
self._ph_client.capture(
distinct_id=self._distinct_id or trace_id,
event="$ai_generation",
properties=event_properties,
+4 -4
View File
@@ -15,6 +15,7 @@ from posthog.ai.utils import (
with_privacy_mode,
)
from posthog.client import Client as PostHogClient
from posthog import setup
class OpenAI(openai.OpenAI):
@@ -24,16 +25,15 @@ class OpenAI(openai.OpenAI):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
api_key: OpenAI API key.
posthog_client: If provided, events will be captured via this client instead
of the global posthog.
posthog_client: If provided, events will be captured via this client instead of the global `posthog`.
**openai_config: Any additional keyword args to set on openai (e.g. organization="xxx").
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
# Store original objects after parent initialization (only if they exist)
self._original_chat = getattr(self, "chat", None)
+4 -3
View File
@@ -1,6 +1,6 @@
import time
import uuid
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, cast
try:
import openai
@@ -9,6 +9,7 @@ except ImportError:
"Please install the OpenAI SDK to use this feature: 'pip install openai'"
)
from posthog import setup
from posthog.ai.utils import (
call_llm_and_track_usage_async,
get_model_params,
@@ -24,7 +25,7 @@ class AsyncOpenAI(openai.AsyncOpenAI):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
api_key: OpenAI API key.
@@ -33,7 +34,7 @@ class AsyncOpenAI(openai.AsyncOpenAI):
**openai_config: Any additional keyword args to set on openai (e.g. organization="xxx").
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
# Store original objects after parent initialization (only if they exist)
self._original_chat = getattr(self, "chat", None)
+7 -4
View File
@@ -15,7 +15,10 @@ from posthog.ai.openai.openai_async import WrappedBeta as AsyncWrappedBeta
from posthog.ai.openai.openai_async import WrappedChat as AsyncWrappedChat
from posthog.ai.openai.openai_async import WrappedEmbeddings as AsyncWrappedEmbeddings
from posthog.ai.openai.openai_async import WrappedResponses as AsyncWrappedResponses
from typing import Optional
from posthog.client import Client as PostHogClient
from posthog import setup
class AzureOpenAI(openai.AzureOpenAI):
@@ -25,7 +28,7 @@ class AzureOpenAI(openai.AzureOpenAI):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
api_key: Azure OpenAI API key.
@@ -34,7 +37,7 @@ class AzureOpenAI(openai.AzureOpenAI):
**openai_config: Any additional keyword args to set on Azure OpenAI (e.g. azure_endpoint="xxx").
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
# Store original objects after parent initialization (only if they exist)
self._original_chat = getattr(self, "chat", None)
@@ -63,7 +66,7 @@ class AsyncAzureOpenAI(openai.AsyncAzureOpenAI):
_ph_client: PostHogClient
def __init__(self, posthog_client: PostHogClient, **kwargs):
def __init__(self, posthog_client: Optional[PostHogClient] = None, **kwargs):
"""
Args:
api_key: Azure OpenAI API key.
@@ -72,7 +75,7 @@ class AsyncAzureOpenAI(openai.AsyncAzureOpenAI):
**openai_config: Any additional keyword args to set on Azure OpenAI (e.g. azure_endpoint="xxx").
"""
super().__init__(**kwargs)
self._ph_client = posthog_client
self._ph_client = posthog_client or setup()
# Store original objects after parent initialization (only if they exist)
self._original_chat = getattr(self, "chat", None)
+71
View File
@@ -0,0 +1,71 @@
from typing import TypedDict, Optional, Any, Dict, Union, Tuple, Type
from types import TracebackType
from typing_extensions import NotRequired # For Python < 3.11 compatibility
from datetime import datetime
import numbers
from uuid import UUID
from posthog.types import SendFeatureFlagsOptions
ID_TYPES = Union[numbers.Number, str, UUID, int]
class OptionalCaptureArgs(TypedDict):
"""Optional arguments for the capture method.
Args:
distinct_id: Unique identifier for the person associated with this event. If not set, the context
distinct_id is used, if available, otherwise a UUID is generated, and the event is marked
as personless. Setting context-level distinct_id's is recommended.
properties: Dictionary of properties to track with the event
timestamp: When the event occurred (defaults to current time)
uuid: Unique identifier for this specific event. If not provided, one is generated. The event
UUID is returned, so you can correlate it with actions in your app (like showing users an
error ID if you capture an exception).
groups: Group identifiers to associate with this event (format: {group_type: group_key})
send_feature_flags: Whether to include currently active feature flags in the event properties.
Can be a boolean (True/False) or a SendFeatureFlagsOptions object for advanced configuration.
Defaults to False.
disable_geoip: Whether to disable GeoIP lookup for this event. Defaults to False.
"""
distinct_id: NotRequired[Optional[ID_TYPES]]
properties: NotRequired[Optional[Dict[str, Any]]]
timestamp: NotRequired[Optional[Union[datetime, str]]]
uuid: NotRequired[Optional[str]]
groups: NotRequired[Optional[Dict[str, str]]]
send_feature_flags: NotRequired[
Optional[Union[bool, SendFeatureFlagsOptions]]
] # Updated to support both boolean and options object
disable_geoip: NotRequired[
Optional[bool]
] # As above, optional so we can tell if the user is intentionally overriding a client setting or not
class OptionalSetArgs(TypedDict):
"""Optional arguments for the set method.
Args:
distinct_id: Unique identifier for the user to set properties on. If not set, the context
distinct_id is used, if available, otherwise this function does nothing. Setting
context-level distinct_id's is recommended.
properties: Dictionary of properties to set on the person
timestamp: When the properties were set (defaults to current time)
uuid: Unique identifier for this operation. If not provided, one is generated. This
UUID is returned, so you can correlate it with actions in your app.
disable_geoip: Whether to disable GeoIP lookup for this operation. Defaults to False.
"""
distinct_id: NotRequired[Optional[ID_TYPES]]
properties: NotRequired[Optional[Dict[str, Any]]]
timestamp: NotRequired[Optional[Union[datetime, str]]]
uuid: NotRequired[Optional[str]]
disable_geoip: NotRequired[Optional[bool]]
ExcInfo = Union[
Tuple[Type[BaseException], BaseException, Optional[TracebackType]],
Tuple[None, None, None],
]
ExceptionArg = Union[BaseException, ExcInfo]
+785 -335
View File
File diff suppressed because it is too large Load Diff
+59 -20
View File
@@ -1,6 +1,10 @@
import contextvars
from contextlib import contextmanager
from typing import Optional, Any, Callable, Dict, TypeVar, cast
from typing import Optional, Any, Callable, Dict, TypeVar, cast, TYPE_CHECKING
if TYPE_CHECKING:
# To avoid circular imports
from posthog.client import Client
class ContextScope:
@@ -9,7 +13,9 @@ class ContextScope:
parent=None,
fresh: bool = False,
capture_exceptions: bool = True,
client: Optional["Client"] = None,
):
self.client: Optional[Client] = client
self.parent = parent
self.fresh = fresh
self.capture_exceptions = capture_exceptions
@@ -64,7 +70,11 @@ def _get_current_context() -> Optional[ContextScope]:
@contextmanager
def new_context(fresh=False, capture_exceptions=True):
def new_context(
fresh: bool = False,
capture_exceptions: bool = True,
client: Optional["Client"] = None,
):
"""
Create a new context scope that will be active for the duration of the with block.
Any tags set within this scope will be isolated to this context. Any exceptions raised
@@ -77,34 +87,49 @@ def new_context(fresh=False, capture_exceptions=True):
capture_exceptions: Whether to capture exceptions raised within the context (default: True).
If True, captures exceptions and tags them with the context tags before propagating them.
If False, exceptions will propagate without being tagged or captured.
client: Optional client instance to use for capturing exceptions (default: None).
If provided, the client will be used to capture exceptions within the context.
If not provided, the default (global) client will be used. Note that the passed
client is only used to capture exceptions within the context - other events captured
within the context via `Client.capture` or `posthog.capture` will still carry the context
state (tags, identity, session id), but will be captured by the client directly used (or
the global one, in the case of `posthog.capture`)
Examples:
```python
# Inherit parent context tags
with posthog.new_context():
posthog.tag("request_id", "123")
# Both this event and the exception will be tagged with the context tags
posthog.capture("event_name", {"property": "value"})
raise ValueError("Something went wrong")
```
```python
# Start with fresh context (no inherited tags)
with posthog.new_context(fresh=True):
posthog.tag("request_id", "123")
# Both this event and the exception will be tagged with the context tags
posthog.capture("event_name", {"property": "value"})
raise ValueError("Something went wrong")
```
Category:
Contexts
"""
from posthog import capture_exception
current_context = _get_current_context()
new_context = ContextScope(current_context, fresh, capture_exceptions)
new_context = ContextScope(current_context, fresh, capture_exceptions, client)
_context_stack.set(new_context)
try:
yield
except Exception as e:
if new_context.capture_exceptions:
capture_exception(e)
if new_context.client:
new_context.client.capture_exception(e)
else:
capture_exception(e)
raise
finally:
_context_stack.set(new_context.get_parent())
@@ -112,22 +137,26 @@ def new_context(fresh=False, capture_exceptions=True):
def tag(key: str, value: Any) -> None:
"""
Add a tag to the current context.
Add a tag to the current context. All tags are added as properties to any event, including exceptions, captured
within the context.
Args:
key: The tag key
value: The tag value
Example:
```python
posthog.tag("user_id", "123")
```
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
current_context.add_tag(key, value)
# NOTE: we should probably also remove this - there's no reason for the user to ever
# need to manually interact with the current tag set
def get_tags() -> Dict[str, Any]:
"""
Get all tags from the current context. Note, modifying
@@ -135,6 +164,9 @@ def get_tags() -> Dict[str, Any]:
Returns:
Dict of all tags in the current context
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
@@ -142,25 +174,20 @@ def get_tags() -> Dict[str, Any]:
return {}
# NOTE: We should probably remove this function - the way to clear scope context
# is by entering a new, fresh context, rather than by clearing the tags or other
# scope data directly.
def clear_tags() -> None:
"""Clear all tags in the current context. Does not clear parent tags"""
current_context = _get_current_context()
if current_context:
current_context.tags.clear()
def identify_context(distinct_id: str) -> None:
"""
Identify the current context with a distinct ID, associating all events captured in this or
child contexts with the given distinct ID (unless identify_context is called again). This is overridden by
distinct id's passed directly to posthog.capture and related methods (identify, set etc). Entering a
fresh context will clear the context-level distinct ID.
fresh context will clear the context-level distinct ID. The distinct-id passed should be uniquely associated
with one of your users. Events captured outside of a context, or in a context with no associated distinct
ID, will be assigned a random UUID, and captured as "personless".
Args:
distinct_id: The distinct ID to associate with the current context and its children.
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
@@ -175,6 +202,9 @@ def set_context_session(session_id: str) -> None:
Args:
session_id: The session ID to associate with the current context and its children. See https://posthog.com/docs/data/sessions
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
@@ -187,6 +217,9 @@ def get_context_session_id() -> Optional[str]:
Returns:
The session ID if set, None otherwise
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
@@ -200,6 +233,9 @@ def get_context_distinct_id() -> Optional[str]:
Returns:
The distinct ID if set, None otherwise
Category:
Contexts
"""
current_context = _get_current_context()
if current_context:
@@ -210,7 +246,7 @@ def get_context_distinct_id() -> Optional[str]:
F = TypeVar("F", bound=Callable[..., Any])
def scoped(fresh=False, capture_exceptions=True):
def scoped(fresh: bool = False, capture_exceptions: bool = True):
"""
Decorator that creates a new context for the function. Simply wraps
the function in a with posthog.new_context(): block.
@@ -230,6 +266,9 @@ def scoped(fresh=False, capture_exceptions=True):
# If this raises an exception, it will be captured with tags
# and then re-raised
some_risky_function()
Category:
Contexts
"""
def decorator(func: F) -> F:
+3 -25
View File
@@ -6,47 +6,25 @@
import logging
import sys
import threading
from enum import Enum
from typing import TYPE_CHECKING, List, Optional
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from posthog.client import Client
class Integrations(str, Enum):
Django = "django"
class ExceptionCapture:
# TODO: Add client side rate limiting to prevent spamming the server with exceptions
log = logging.getLogger("posthog")
def __init__(
self, client: "Client", integrations: Optional[List[Integrations]] = None
):
def __init__(self, client: "Client"):
self.client = client
self.original_excepthook = sys.excepthook
sys.excepthook = self.exception_handler
threading.excepthook = self.thread_exception_handler
self.enabled_integrations = []
for integration in integrations or []:
# TODO: Maybe find a better way of enabling integrations
# This is very annoying currently if we had to add any configuration per integration
if integration == Integrations.Django:
try:
from posthog.exception_integrations.django import DjangoIntegration
enabled_integration = DjangoIntegration(self.exception_receiver)
self.enabled_integrations.append(enabled_integration)
except Exception as e:
self.log.exception(f"Failed to enable Django integration: {e}")
def close(self):
sys.excepthook = self.original_excepthook
for integration in self.enabled_integrations:
integration.uninstall()
def exception_handler(self, exc_type, exc_value, exc_traceback):
# don't affect default behaviour.
@@ -66,6 +44,6 @@ class ExceptionCapture:
def capture_exception(self, exception, metadata=None):
try:
distinct_id = metadata.get("distinct_id") if metadata else None
self.client.capture_exception(exception, distinct_id)
self.client.capture_exception(exception, distinct_id=distinct_id)
except Exception as e:
self.log.exception(f"Failed to capture exception: {e}")
@@ -1,5 +0,0 @@
class IntegrationEnablingError(Exception):
"""
The integration could not be enabled due to a user error like
`django` not being installed for the `DjangoIntegration`.
"""
-117
View File
@@ -1,117 +0,0 @@
# Portions of this file are derived from getsentry/sentry-javascript by Software, Inc. dba Sentry
# Licensed under the MIT License
# 💖open source (under MIT License)
import re
import sys
from typing import TYPE_CHECKING
from posthog.exception_integrations import IntegrationEnablingError
try:
from django import VERSION as DJANGO_VERSION
from django.core import signals
except ImportError:
raise IntegrationEnablingError("Django not installed")
if TYPE_CHECKING:
from typing import Any, Dict # noqa: F401
from django.core.handlers.wsgi import WSGIRequest # noqa: F401
class DjangoIntegration:
# TODO: Abstract integrations one we have more and can see patterns
"""
Autocapture errors from a Django application.
"""
identifier = "django"
def __init__(self, capture_exception_fn=None):
if DJANGO_VERSION < (4, 2):
raise IntegrationEnablingError("Django 4.2 or newer is required.")
# TODO: Right now this seems too complicated / overkill for us, but seems like we can automatically plug in middlewares
# which is great for users (they don't need to do this) and everything should just work.
# We should consider this in the future, but for now we can just use the middleware and signals handlers.
# See: https://github.com/getsentry/sentry-python/blob/269d96d6e9821122fbff280e6a26956e5ed03c0b/sentry_sdk/integrations/django/__init__.py
self.capture_exception_fn = capture_exception_fn
def _got_request_exception(request=None, **kwargs):
# type: (WSGIRequest, **Any) -> None
extra_props = {}
if request is not None:
# get headers metadata
extra_props = DjangoRequestExtractor(request).extract_person_data()
self.capture_exception_fn(sys.exc_info(), extra_props)
signals.got_request_exception.connect(_got_request_exception)
def uninstall(self):
pass
class DjangoRequestExtractor:
def __init__(self, request):
# type: (Any) -> None
self.request = request
def extract_person_data(self):
headers = self.headers()
# Extract traceparent and tracestate headers
traceparent = headers.get("Traceparent")
tracestate = headers.get("Tracestate")
# Extract the distinct_id from tracestate
distinct_id = None
if tracestate:
# TODO: Align on the format of the distinct_id in tracestate
# We can't have comma or equals in header values here, so maybe we should base64 encode it?
match = re.search(r"posthog-distinct-id=([^,]+)", tracestate)
if match:
distinct_id = match.group(1)
return {
**self.user(),
"distinct_id": distinct_id,
"ip": headers.get("X-Forwarded-For"),
"user_agent": headers.get("User-Agent"),
"traceparent": traceparent,
"$request_path": self.request.path,
}
def user(self):
user_data: dict[str, str] = {}
user = getattr(self.request, "user", None)
if user is None or not user.is_authenticated:
return user_data
try:
user_id = str(user.pk)
if user_id:
user_data.setdefault("$user_id", user_id)
except Exception:
pass
try:
email = str(user.email)
if email:
user_data.setdefault("email", email)
except Exception:
pass
return user_data
def headers(self):
# type: () -> Dict[str, str]
return dict(self.request.headers)
+84 -167
View File
@@ -11,7 +11,24 @@ import re
import sys
import types
from datetime import datetime
from typing import TYPE_CHECKING
from types import FrameType, TracebackType # noqa: F401
from typing import ( # noqa: F401
Any,
Dict,
Iterator,
List,
Literal,
Optional,
Set,
Tuple,
TypedDict,
TypeVar,
Union,
cast,
TYPE_CHECKING,
)
from posthog.args import ExcInfo, ExceptionArg # noqa: F401
try:
# Python 3.11
@@ -23,85 +40,61 @@ except ImportError:
DEFAULT_MAX_VALUE_LENGTH = 1024
LogLevelStr = Literal["fatal", "critical", "error", "warning", "info", "debug"]
if TYPE_CHECKING:
from types import FrameType, TracebackType
from typing import ( # noqa: F401
Any,
Callable,
Dict,
Iterator,
List,
Literal,
Optional,
Set,
Tuple,
Type,
TypedDict,
TypeVar,
Union,
cast,
)
ExcInfo = Union[
Tuple[Type[BaseException], BaseException, Optional[TracebackType]],
Tuple[None, None, None],
]
LogLevelStr = Literal["fatal", "critical", "error", "warning", "info", "debug"]
Event = TypedDict(
"Event",
{
"breadcrumbs": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
"check_in_id": str,
"contexts": Dict[str, Dict[str, object]],
"dist": str,
"duration": Optional[float],
"environment": str,
"errors": List[Dict[str, Any]], # TODO: We can expand on this type
"event_id": str,
"exception": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
# "extra": MutableMapping[str, object],
# "fingerprint": List[str],
"level": LogLevelStr,
# "logentry": Mapping[str, object],
"logger": str,
# "measurements": Dict[str, MeasurementValue],
"message": str,
"modules": Dict[str, str],
# "monitor_config": Mapping[str, object],
"monitor_slug": Optional[str],
"platform": Literal["python"],
"profile": object,
"release": str,
"request": Dict[str, object],
# "sdk": Mapping[str, object],
"server_name": str,
"spans": List[Dict[str, object]],
"stacktrace": Dict[
str, object
], # We access this key in the code, but I am unsure whether we ever set it
"start_timestamp": datetime,
"status": Optional[str],
# "tags": MutableMapping[
# str, str
# ], # Tags must be less than 200 characters each
"threads": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
"timestamp": Optional[datetime], # Must be set before sending the event
"transaction": str,
# "transaction_info": Mapping[str, Any], # TODO: We can expand on this type
"type": Literal["check_in", "transaction"],
"user": Dict[str, object],
"_metrics_summary": Dict[str, object],
},
total=False,
)
Event = TypedDict(
"Event",
{
"breadcrumbs": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
"check_in_id": str,
"contexts": Dict[str, Dict[str, object]],
"dist": str,
"duration": Optional[float],
"environment": str,
"errors": List[Dict[str, Any]], # TODO: We can expand on this type
"event_id": str,
"exception": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
# "extra": MutableMapping[str, object],
# "fingerprint": List[str],
"level": LogLevelStr,
# "logentry": Mapping[str, object],
"logger": str,
# "measurements": Dict[str, MeasurementValue],
"message": str,
"modules": Dict[str, str],
# "monitor_config": Mapping[str, object],
"monitor_slug": Optional[str],
"platform": Literal["python"],
"profile": object,
"release": str,
"request": Dict[str, object],
# "sdk": Mapping[str, object],
"server_name": str,
"spans": List[Dict[str, object]],
"stacktrace": Dict[
str, object
], # We access this key in the code, but I am unsure whether we ever set it
"start_timestamp": datetime,
"status": Optional[str],
# "tags": MutableMapping[
# str, str
# ], # Tags must be less than 200 characters each
"threads": Dict[
Literal["values"], List[Dict[str, Any]]
], # TODO: We can expand on this type
"timestamp": Optional[datetime], # Must be set before sending the event
"transaction": str,
# "transaction_info": Mapping[str, Any], # TODO: We can expand on this type
"type": Literal["check_in", "transaction"],
"user": Dict[str, object],
"_metrics_summary": Dict[str, object],
},
total=False,
)
epoch = datetime(1970, 1, 1)
@@ -362,12 +355,9 @@ def filename_for_module(module, abs_path):
def serialize_frame(
frame,
tb_lineno=None,
include_local_variables=True,
include_source_context=True,
max_value_length=None,
custom_repr=None,
):
# type: (FrameType, Optional[int], bool, bool, Optional[int], Optional[Callable[..., Optional[str]]]) -> Dict[str, Any]
# type: (FrameType, Optional[int], Optional[int]) -> Dict[str, Any]
f_code = getattr(frame, "f_code", None)
if not f_code:
abs_path = None
@@ -392,45 +382,13 @@ def serialize_frame(
"lineno": tb_lineno,
} # type: Dict[str, Any]
if include_source_context:
rv["pre_context"], rv["context_line"], rv["post_context"] = get_source_context(
frame, tb_lineno, max_value_length
)
if include_local_variables:
# TODO - we don't support local variables, yet
pass
rv["pre_context"], rv["context_line"], rv["post_context"] = get_source_context(
frame, tb_lineno, max_value_length
)
return rv
def current_stacktrace(
include_local_variables=True, # type: bool
include_source_context=True, # type: bool
max_value_length=None, # type: Optional[int]
):
# type: (...) -> Dict[str, Any]
__tracebackhide__ = True
frames = []
f = sys._getframe() # type: Optional[FrameType]
while f is not None:
if not should_hide_frame(f):
frames.append(
serialize_frame(
f,
include_local_variables=include_local_variables,
include_source_context=include_source_context,
max_value_length=max_value_length,
)
)
f = f.f_back
frames.reverse()
return {"frames": frames, "type": "raw"}
def get_errno(exc_value):
# type: (BaseException) -> Optional[Any]
return getattr(exc_value, "errno", None)
@@ -451,7 +409,6 @@ def single_exception_from_error_tuple(
exc_type, # type: Optional[type]
exc_value, # type: Optional[BaseException]
tb, # type: Optional[TracebackType]
client_options=None, # type: Optional[Dict[str, Any]]
mechanism=None, # type: Optional[Dict[str, Any]]
exception_id=None, # type: Optional[int]
parent_id=None, # type: Optional[int]
@@ -499,25 +456,13 @@ def single_exception_from_error_tuple(
exception_value["type"] = get_type_name(exc_type)
exception_value["value"] = get_error_message(exc_value)
if client_options is None:
include_local_variables = True
include_source_context = True
max_value_length = DEFAULT_MAX_VALUE_LENGTH # fallback
custom_repr = None
else:
include_local_variables = client_options["include_local_variables"]
include_source_context = client_options["include_source_context"]
max_value_length = client_options["max_value_length"]
custom_repr = client_options.get("custom_repr")
max_value_length = DEFAULT_MAX_VALUE_LENGTH # fallback
frames = [
serialize_frame(
tb.tb_frame,
tb_lineno=tb.tb_lineno,
include_local_variables=include_local_variables,
include_source_context=include_source_context,
max_value_length=max_value_length,
custom_repr=custom_repr,
)
for tb in iter_stacks(tb)
]
@@ -573,7 +518,6 @@ def exceptions_from_error(
exc_type, # type: Optional[type]
exc_value, # type: Optional[BaseException]
tb, # type: Optional[TracebackType]
client_options=None, # type: Optional[Dict[str, Any]]
mechanism=None, # type: Optional[Dict[str, Any]]
exception_id=0, # type: int
parent_id=0, # type: int
@@ -589,7 +533,6 @@ def exceptions_from_error(
exc_type=exc_type,
exc_value=exc_value,
tb=tb,
client_options=client_options,
mechanism=mechanism,
exception_id=exception_id,
parent_id=parent_id,
@@ -617,7 +560,6 @@ def exceptions_from_error(
exc_type=type(cause),
exc_value=cause,
tb=getattr(cause, "__traceback__", None),
client_options=client_options,
mechanism=mechanism,
exception_id=exception_id,
source="__cause__",
@@ -638,7 +580,6 @@ def exceptions_from_error(
exc_type=type(context),
exc_value=context,
tb=getattr(context, "__traceback__", None),
client_options=client_options,
mechanism=mechanism,
exception_id=exception_id,
source="__context__",
@@ -653,7 +594,6 @@ def exceptions_from_error(
exc_type=type(e),
exc_value=e,
tb=getattr(e, "__traceback__", None),
client_options=client_options,
mechanism=mechanism,
exception_id=exception_id,
parent_id=parent_id,
@@ -666,7 +606,6 @@ def exceptions_from_error(
def exceptions_from_error_tuple(
exc_info, # type: ExcInfo
client_options=None, # type: Optional[Dict[str, Any]]
mechanism=None, # type: Optional[Dict[str, Any]]
):
# type: (...) -> List[Dict[str, Any]]
@@ -681,7 +620,6 @@ def exceptions_from_error_tuple(
exc_type=exc_type,
exc_value=exc_value,
tb=tb,
client_options=client_options,
mechanism=mechanism,
exception_id=0,
parent_id=0,
@@ -691,9 +629,7 @@ def exceptions_from_error_tuple(
exceptions = []
for exc_type, exc_value, tb in walk_exception_chain(exc_info):
exceptions.append(
single_exception_from_error_tuple(
exc_type, exc_value, tb, client_options, mechanism
)
single_exception_from_error_tuple(exc_type, exc_value, tb, mechanism)
)
exceptions.reverse()
@@ -783,7 +719,7 @@ def set_in_app_in_frames(frames, in_app_exclude, in_app_include, project_root=No
def exception_is_already_captured(error):
# type: (Union[BaseException, ExcInfo]) -> bool
# type: (ExceptionArg) -> bool
if isinstance(error, BaseException):
return hasattr(error, "__posthog_exception_captured")
# Autocaptured exceptions are passed as a tuple from our system hooks,
@@ -796,19 +732,21 @@ def exception_is_already_captured(error):
return False # type: ignore[unreachable]
def mark_exception_as_captured(error):
# type: (Union[BaseException, ExcInfo]) -> None
def mark_exception_as_captured(error, uuid):
# type: (ExceptionArg, str) -> None
if isinstance(error, BaseException):
setattr(error, "__posthog_exception_captured", True)
setattr(error, "__posthog_exception_uuid", uuid)
# Autocaptured exceptions are passed as a tuple from our system hooks,
# the second item is the exception value (the first is the exception type)
elif isinstance(error, tuple) and len(error) > 1:
if error[1] is not None:
setattr(error[1], "__posthog_exception_captured", True)
setattr(error[1], "__posthog_exception_uuid", uuid)
def exc_info_from_error(error):
# type: (Union[BaseException, ExcInfo]) -> ExcInfo
# type: (ExceptionArg) -> ExcInfo
if isinstance(error, tuple) and len(error) == 3:
exc_type, exc_value, tb = error
elif isinstance(error, BaseException):
@@ -865,27 +803,6 @@ def construct_artificial_traceback(e):
setattr(e, "__traceback__", tb)
def event_from_exception(
exc_info, # type: Union[BaseException, ExcInfo]
client_options=None, # type: Optional[Dict[str, Any]]
mechanism=None, # type: Optional[Dict[str, Any]]
):
# type: (...) -> Tuple[Event, Dict[str, Any]]
exc_info = exc_info_from_error(exc_info)
hint = event_hint_with_exc_info(exc_info)
return (
{
"level": "error",
"exception": {
"values": exceptions_from_error_tuple(
exc_info, client_options, mechanism
)
},
},
hint,
)
def _module_in_list(name, items):
# type: (str | None, Optional[List[str]]) -> bool
if name is None:
+15
View File
@@ -109,6 +109,14 @@ def is_condition_match(
property_type = prop.get("type")
if property_type == "cohort":
matches = match_cohort(prop, properties, cohort_properties)
elif property_type == "flag":
log.warning(
"Flag dependency filters are not supported in local evaluation. "
"Skipping condition for flag '%s' with dependency on flag '%s'",
feature_flag.get("key", "unknown"),
prop.get("key", "unknown"),
)
continue
else:
matches = match_property(prop, properties)
if not matches:
@@ -317,6 +325,13 @@ def match_property_group(property_group, property_values, cohort_properties) ->
try:
if prop.get("type") == "cohort":
matches = match_cohort(prop, property_values, cohort_properties)
elif prop.get("type") == "flag":
log.warning(
"Flag dependency filters are not supported in local evaluation. "
"Skipping condition with dependency on flag '%s'",
prop.get("key", "unknown"),
)
continue
else:
matches = match_property(prop, property_values)
+68 -8
View File
@@ -1,5 +1,6 @@
from typing import TYPE_CHECKING, cast
from posthog import scopes
from posthog import contexts, capture_exception
from posthog.client import Client
if TYPE_CHECKING:
from django.http import HttpRequest, HttpResponse # noqa: F401
@@ -16,7 +17,8 @@ class PosthogContextMiddleware:
- Request Method as $request_method
The context will also auto-capture exceptions and send them to PostHog, unless you disable it by setting
`POSTHOG_MW_CAPTURE_EXCEPTIONS` to `False` in your Django settings.
`POSTHOG_MW_CAPTURE_EXCEPTIONS` to `False` in your Django settings. The exceptions are captured using the
global client, unless the setting `POSTHOG_MW_CLIENT` is set to a custom client instance
The middleware behaviour is customisable through 3 additional functions:
- `POSTHOG_MW_EXTRA_TAGS`, which is a Callable[[HttpRequest], Dict[str, Any]] expected to return a dictionary of additional tags to be added to the context.
@@ -74,19 +76,32 @@ class PosthogContextMiddleware:
else:
self.capture_exceptions = True
if hasattr(settings, "POSTHOG_MW_CLIENT") and isinstance(
settings.POSTHOG_MW_CLIENT, Client
):
self.client = cast("Optional[Client]", settings.POSTHOG_MW_CLIENT)
else:
self.client = None
def extract_tags(self, request):
# type: (HttpRequest) -> Dict[str, Any]
tags = {}
(user_id, user_email) = self.extract_request_user(request)
# Extract session ID from X-POSTHOG-SESSION-ID header
session_id = request.headers.get("X-POSTHOG-SESSION-ID")
if session_id:
scopes.set_context_session(session_id)
contexts.set_context_session(session_id)
# Extract distinct ID from X-POSTHOG-DISTINCT-ID header
distinct_id = request.headers.get("X-POSTHOG-DISTINCT-ID")
# Extract distinct ID from X-POSTHOG-DISTINCT-ID header or request user id
distinct_id = request.headers.get("X-POSTHOG-DISTINCT-ID") or user_id
if distinct_id:
scopes.identify_context(distinct_id)
contexts.identify_context(distinct_id)
# Extract user email
if user_email:
tags["email"] = user_email
# Extract current URL
absolute_url = request.build_absolute_uri()
@@ -97,6 +112,20 @@ class PosthogContextMiddleware:
if request.method:
tags["$request_method"] = request.method
# Extract request path
if request.path:
tags["$request_path"] = request.path
# Extract IP address
ip_address = request.headers.get("X-Forwarded-For")
if ip_address:
tags["$ip_address"] = ip_address
# Extract user agent
user_agent = request.headers.get("User-Agent")
if user_agent:
tags["$user_agent"] = user_agent
# Apply extra tags if configured
if self.extra_tags:
extra = self.extra_tags(request)
@@ -109,13 +138,44 @@ class PosthogContextMiddleware:
return tags
def extract_request_user(self, request):
user_id = None
email = None
user = getattr(request, "user", None)
if user and getattr(user, "is_authenticated", False):
try:
user_id = str(user.pk)
except Exception:
pass
try:
email = str(user.email)
except Exception:
pass
return user_id, email
def __call__(self, request):
# type: (HttpRequest) -> HttpResponse
if self.request_filter and not self.request_filter(request):
return self.get_response(request)
with scopes.new_context(self.capture_exceptions):
with contexts.new_context(self.capture_exceptions, client=self.client):
for k, v in self.extract_tags(request).items():
scopes.tag(k, v)
contexts.tag(k, v)
return self.get_response(request)
def process_exception(self, request, exception):
if self.request_filter and not self.request_filter(request):
return
if not self.capture_exceptions:
return
if self.client:
self.client.capture_exception(exception)
else:
capture_exception(exception)
@@ -1727,3 +1727,29 @@ def test_openai_reasoning_tokens(mock_client):
assert call["properties"]["$ai_reasoning_tokens"] is not None
assert call["properties"]["$ai_input_tokens"] is not None
assert call["properties"]["$ai_output_tokens"] is not None
def test_callback_handler_without_client():
"""Test that CallbackHandler works properly when no PostHog client is passed."""
with patch("posthog.ai.langchain.callbacks.setup") as mock_setup:
mock_client = mock_setup.return_value
callbacks = CallbackHandler()
# Verify that setup() was called
mock_setup.assert_called_once()
# Verify that the client was set to the result of setup()
assert callbacks._ph_client == mock_client
# Test that the callback handler works with a simple chain
prompt = ChatPromptTemplate.from_messages([("user", "Foo")])
model = FakeMessagesListChatModel(responses=[AIMessage(content="Bar")])
chain = prompt | model
# This should work and call the mock client
result = chain.invoke({}, config={"callbacks": [callbacks]})
assert result.content == "Bar"
# Verify that the mock client was used for capturing events
assert mock_client.capture.call_count == 3
@@ -1,3 +0,0 @@
import pytest
pytest.importorskip("django")
@@ -1,121 +0,0 @@
from posthog.exception_integrations.django import DjangoRequestExtractor
from django.test import RequestFactory
from django.conf import settings
from django.core.management import call_command
import django
DEFAULT_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
# setup a test app
if not settings.configured:
settings.configure(
SECRET_KEY="test",
DEFAULT_CHARSET="utf-8",
INSTALLED_APPS=[
"django.contrib.auth",
"django.contrib.contenttypes",
],
DATABASES={
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": ":memory:",
}
},
)
django.setup()
call_command("migrate", verbosity=0, interactive=False)
def mock_request_factory(override_headers):
factory = RequestFactory(
headers={
"User-Agent": DEFAULT_USER_AGENT,
"Referrer": "http://example.com",
"X-Forwarded-For": "193.4.5.12",
**(override_headers or {}),
}
)
request = factory.get("/api/endpoint")
return request
def test_request_extractor_with_no_trace():
request = mock_request_factory(None)
extractor = DjangoRequestExtractor(request)
assert extractor.extract_person_data() == {
"ip": "193.4.5.12",
"user_agent": DEFAULT_USER_AGENT,
"traceparent": None,
"distinct_id": None,
"$request_path": "/api/endpoint",
}
def test_request_extractor_with_trace():
request = mock_request_factory(
{"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"}
)
extractor = DjangoRequestExtractor(request)
assert extractor.extract_person_data() == {
"ip": "193.4.5.12",
"user_agent": DEFAULT_USER_AGENT,
"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
"distinct_id": None,
"$request_path": "/api/endpoint",
}
def test_request_extractor_with_tracestate():
request = mock_request_factory(
{
"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
"tracestate": "posthog-distinct-id=1234",
}
)
extractor = DjangoRequestExtractor(request)
assert extractor.extract_person_data() == {
"ip": "193.4.5.12",
"user_agent": DEFAULT_USER_AGENT,
"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
"distinct_id": "1234",
"$request_path": "/api/endpoint",
}
def test_request_extractor_with_complicated_tracestate():
request = mock_request_factory(
{"tracestate": "posthog-distinct-id=alohaMountainsXUYZ,rojo=00f067aa0ba902b7"}
)
extractor = DjangoRequestExtractor(request)
assert extractor.extract_person_data() == {
"ip": "193.4.5.12",
"user_agent": DEFAULT_USER_AGENT,
"traceparent": None,
"distinct_id": "alohaMountainsXUYZ",
"$request_path": "/api/endpoint",
}
def test_request_extractor_with_request_user():
from django.contrib.auth.models import User
user = User.objects.create_user(
username="test", email="test@posthog.com", password="top_secret"
)
request = mock_request_factory(None)
request.user = user
extractor = DjangoRequestExtractor(request)
assert extractor.extract_person_data() == {
"ip": "193.4.5.12",
"user_agent": DEFAULT_USER_AGENT,
"traceparent": None,
"distinct_id": None,
"$request_path": "/api/endpoint",
"email": "test@posthog.com",
"$user_id": "1",
}
+5 -1
View File
@@ -1,4 +1,8 @@
from posthog.scopes import new_context, get_context_session_id, get_context_distinct_id
from posthog.contexts import (
new_context,
get_context_session_id,
get_context_distinct_id,
)
import unittest
from unittest.mock import Mock
+118 -71
View File
@@ -40,16 +40,30 @@ class TestClient(unittest.TestCase):
event["properties"]["processed_by_before_send"] = True
return event
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=my_before_send
)
success, msg = client.capture("user1", "test_event", {"original": "value"})
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=my_before_send,
sync_mode=True,
)
msg_uuid = client.capture(
"test_event", distinct_id="user1", properties={"original": "value"}
)
self.assertTrue(success)
self.assertEqual(msg["properties"]["processed_by_before_send"], True)
self.assertEqual(msg["properties"]["original"], "value")
self.assertEqual(len(processed_events), 1)
self.assertEqual(processed_events[0]["event"], "test_event")
self.assertIsNotNone(msg_uuid)
# Get the enqueued message from the mock
mock_post.assert_called_once()
batch_data = mock_post.call_args[1]["batch"]
enqueued_msg = batch_data[0]
self.assertEqual(
enqueued_msg["properties"]["processed_by_before_send"], True
)
self.assertEqual(enqueued_msg["properties"]["original"], "value")
self.assertEqual(len(processed_events), 1)
self.assertEqual(processed_events[0]["event"], "test_event")
def test_before_send_callback_drops_event(self):
"""Test that before_send callback can drop events by returning None."""
@@ -59,20 +73,27 @@ class TestClient(unittest.TestCase):
return None
return event
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=drop_test_events
)
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=drop_test_events,
sync_mode=True,
)
# Event should be dropped
success, msg = client.capture("user1", "test_drop_me")
self.assertTrue(success)
self.assertIsNone(msg)
# Event should be dropped
msg_uuid = client.capture("test_drop_me", distinct_id="user1")
self.assertIsNone(msg_uuid)
# Event should go through
success, msg = client.capture("user1", "keep_me")
self.assertTrue(success)
self.assertIsNotNone(msg)
self.assertEqual(msg["event"], "keep_me")
# Event should go through
msg_uuid = client.capture("keep_me", distinct_id="user1")
self.assertIsNotNone(msg_uuid)
# Check the enqueued message
mock_post.assert_called_once()
batch_data = mock_post.call_args[1]["batch"]
enqueued_msg = batch_data[0]
self.assertEqual(enqueued_msg["event"], "keep_me")
def test_before_send_callback_handles_exceptions(self):
"""Test that exceptions in before_send don't crash the client."""
@@ -80,18 +101,26 @@ class TestClient(unittest.TestCase):
def buggy_before_send(event):
raise ValueError("Oops!")
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=buggy_before_send
)
success, msg = client.capture("user1", "robust_event")
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=buggy_before_send,
sync_mode=True,
)
msg_uuid = client.capture("robust_event", distinct_id="user1")
# Event should still be sent despite the exception
self.assertTrue(success)
self.assertIsNotNone(msg)
self.assertEqual(msg["event"], "robust_event")
# Event should still be sent despite the exception
self.assertIsNotNone(msg_uuid)
# Check the enqueued message
mock_post.assert_called_once()
batch_data = mock_post.call_args[1]["batch"]
enqueued_msg = batch_data[0]
self.assertEqual(enqueued_msg["event"], "robust_event")
def test_before_send_callback_works_with_all_event_types(self):
"""Test that before_send works with capture, identify, set, etc."""
"""Test that before_send works with capture, set, etc."""
def add_marker(event):
if "properties" not in event:
@@ -99,38 +128,46 @@ class TestClient(unittest.TestCase):
event["properties"]["marked"] = True
return event
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=add_marker
)
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=add_marker,
sync_mode=True,
)
# Test capture
success, msg = client.capture("user1", "event")
self.assertTrue(success)
self.assertTrue(msg["properties"]["marked"])
# Test capture
msg_uuid = client.capture("event", distinct_id="user1")
self.assertIsNotNone(msg_uuid)
# Test identify
success, msg = client.identify("user1", {"trait": "value"})
self.assertTrue(success)
self.assertTrue(msg["properties"]["marked"])
# Test set
msg_uuid = client.set(distinct_id="user1", properties={"prop": "value"})
self.assertIsNotNone(msg_uuid)
# Test set
success, msg = client.set("user1", {"prop": "value"})
self.assertTrue(success)
self.assertTrue(msg["properties"]["marked"])
# Test page
success, msg = client.page("user1", "https://example.com")
self.assertTrue(success)
self.assertTrue(msg["properties"]["marked"])
# Check all events were marked
self.assertEqual(mock_post.call_count, 2)
for call in mock_post.call_args_list:
batch_data = call[1]["batch"]
enqueued_msg = batch_data[0]
self.assertTrue(enqueued_msg["properties"]["marked"])
def test_before_send_callback_disabled_when_none(self):
"""Test that client works normally when before_send is None."""
client = Client(FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=None)
success, msg = client.capture("user1", "normal_event")
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=None,
sync_mode=True,
)
msg_uuid = client.capture("normal_event", distinct_id="user1")
self.assertIsNotNone(msg_uuid)
self.assertTrue(success)
self.assertIsNotNone(msg)
self.assertEqual(msg["event"], "normal_event")
# Check the event was sent normally
mock_post.assert_called_once()
batch_data = mock_post.call_args[1]["batch"]
enqueued_msg = batch_data[0]
self.assertEqual(enqueued_msg["event"], "normal_event")
def test_before_send_callback_pii_scrubbing_example(self):
"""Test a realistic PII scrubbing use case."""
@@ -152,20 +189,30 @@ class TestClient(unittest.TestCase):
return event
client = Client(
FAKE_TEST_API_KEY, on_error=self.set_fail, before_send=scrub_pii
)
success, msg = client.capture(
"user1",
"form_submit",
{
"email": "user@example.com",
"credit_card": "1234-5678-9012-3456",
"form_name": "contact",
},
)
with mock.patch("posthog.client.batch_post") as mock_post:
client = Client(
FAKE_TEST_API_KEY,
on_error=self.set_fail,
before_send=scrub_pii,
sync_mode=True,
)
msg_uuid = client.capture(
"form_submit",
distinct_id="user1",
properties={
"email": "user@example.com",
"credit_card": "1234-5678-9012-3456",
"form_name": "contact",
},
)
self.assertTrue(success)
self.assertEqual(msg["properties"]["email"], "***@example.com")
self.assertNotIn("credit_card", msg["properties"])
self.assertEqual(msg["properties"]["form_name"], "contact")
self.assertIsNotNone(msg_uuid)
# Check the enqueued message was scrubbed
mock_post.assert_called_once()
batch_data = mock_post.call_args[1]["batch"]
enqueued_msg = batch_data[0]
self.assertEqual(enqueued_msg["properties"]["email"], "***@example.com")
self.assertNotIn("credit_card", enqueued_msg["properties"])
self.assertEqual(enqueued_msg["properties"]["form_name"], "contact")
+1240 -713
View File
File diff suppressed because it is too large Load Diff
@@ -1,8 +1,7 @@
import unittest
from unittest.mock import patch
from posthog.scopes import (
clear_tags,
from posthog.contexts import (
get_tags,
new_context,
scoped,
@@ -14,11 +13,7 @@ from posthog.scopes import (
)
class TestScopes(unittest.TestCase):
def setUp(self):
# Reset any context between tests
clear_tags()
class TestContexts(unittest.TestCase):
def test_tag_and_get_tags(self):
with new_context(fresh=True):
tag("key1", "value1")
@@ -28,14 +23,6 @@ class TestScopes(unittest.TestCase):
assert tags["key1"] == "value1"
assert tags["key2"] == 2
def test_clear_tags(self):
with new_context(fresh=True):
tag("key1", "value1")
assert get_tags()["key1"] == "value1"
clear_tags()
assert get_tags() == {}
def test_new_context_isolation(self):
with new_context(fresh=True):
# Set tag in outer context
-29
View File
@@ -32,32 +32,3 @@ def test_excepthook(tmpdir):
b'"$exception_list": [{"mechanism": {"type": "generic", "handled": true}, "module": null, "type": "ZeroDivisionError", "value": "division by zero", "stacktrace": {"frames": [{"platform": "python", "filename": "app.py", "abs_path"'
in output
)
def test_trying_to_use_django_integration(tmpdir):
app = tmpdir.join("app.py")
app.write(
dedent(
"""
from posthog import Posthog, Integrations
posthog = Posthog('phc_x', host='https://eu.i.posthog.com', enable_exception_autocapture=True, exception_autocapture_integrations=[Integrations.Django], debug=True, on_error=lambda e, batch: print('error handling batch: ', e, batch))
# frame_value = "LOL"
1/0
"""
)
)
with pytest.raises(subprocess.CalledProcessError) as excinfo:
subprocess.check_output([sys.executable, str(app)], stderr=subprocess.STDOUT)
output = excinfo.value.output
assert b"ZeroDivisionError" in output
assert b"LOL" in output
assert b"DEBUG:posthog:data uploaded successfully" in output
assert (
b'"$exception_list": [{"mechanism": {"type": "generic", "handled": true}, "module": null, "type": "ZeroDivisionError", "value": "division by zero", "stacktrace": {"frames": [{"platform": "python", "filename": "app.py", "abs_path"'
in output
)
+12 -12
View File
@@ -229,9 +229,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertEqual(flag_result.variant, None)
self.assertEqual(flag_result.payload, 300)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -283,9 +283,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertEqual(flag_result.payload, {"some": "value"})
patch_capture.assert_called_with(
"distinct_id",
"$feature_flag_called",
{
distinct_id="distinct_id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": "variant-1",
"locally_evaluated": True,
@@ -305,9 +305,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertIsNone(another_flag_result.payload)
patch_capture.assert_called_with(
"another-distinct-id",
"$feature_flag_called",
{
distinct_id="another-distinct-id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": "variant-2",
"locally_evaluated": True,
@@ -345,9 +345,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertEqual(flag_result.variant, None)
self.assertEqual(flag_result.payload, 300)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": True,
"locally_evaluated": False,
@@ -388,9 +388,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertEqual(flag_result.get_value(), "variant-1")
self.assertEqual(flag_result.payload, [1, 2, 3])
patch_capture.assert_called_with(
"distinct_id",
"$feature_flag_called",
{
distinct_id="distinct_id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": "variant-1",
"locally_evaluated": False,
@@ -431,9 +431,9 @@ class TestGetFeatureFlagResult(unittest.TestCase):
self.assertIsNone(flag_result)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "no-person-flag",
"$feature_flag_response": None,
"locally_evaluated": False,
+101 -22
View File
@@ -1355,6 +1355,77 @@ class TestLocalEvaluation(unittest.TestCase):
self.assertEqual(patch_flags.call_count, 0)
self.assertEqual(patch_get.call_count, 0)
@mock.patch("posthog.feature_flags.log")
@mock.patch("posthog.client.flags")
@mock.patch("posthog.client.get")
def test_feature_flags_with_flag_dependencies(
self, patch_get, patch_flags, mock_log
):
client = Client(FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
client.feature_flags = [
{
"id": 1,
"name": "Flag with Dependencies",
"key": "flag-with-dependencies",
"active": True,
"filters": {
"groups": [
{
"properties": [
{
"key": "beta-feature",
"operator": "exact",
"value": True,
"type": "flag",
},
{
"key": "email",
"operator": "icontains",
"value": "@example.com",
"type": "person",
},
],
"rollout_percentage": 100,
}
],
},
}
]
# Test that flag evaluation doesn't fail when encountering a flag dependency
# The flag should evaluate based on other conditions (email contains @example.com)
# Since flag dependencies aren't implemented, it should skip the flag condition
# and evaluate based on the email condition only
feature_flag_match = client.get_feature_flag(
"flag-with-dependencies",
"test-user",
person_properties={"email": "test@example.com"},
)
self.assertEqual(feature_flag_match, True)
self.assertEqual(patch_flags.call_count, 0)
self.assertEqual(patch_get.call_count, 0)
# Verify warning was logged for flag dependency
mock_log.warning.assert_called_with(
"Flag dependency filters are not supported in local evaluation. "
"Skipping condition for flag '%s' with dependency on flag '%s'",
"flag-with-dependencies",
"beta-feature",
)
# Test with email that doesn't match
feature_flag_match = client.get_feature_flag(
"flag-with-dependencies",
"test-user-2",
person_properties={"email": "test@other.com"},
)
self.assertEqual(feature_flag_match, False)
self.assertEqual(patch_flags.call_count, 0)
self.assertEqual(patch_get.call_count, 0)
# Verify warning was logged again for the second evaluation
self.assertEqual(mock_log.warning.call_count, 2)
@mock.patch("posthog.client.Poller")
@mock.patch("posthog.client.get")
def test_load_feature_flags(self, patch_get, patch_poll):
@@ -2695,9 +2766,9 @@ class TestCaptureCalls(unittest.TestCase):
)
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "complex-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -2729,9 +2800,9 @@ class TestCaptureCalls(unittest.TestCase):
)
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id2",
"$feature_flag_called",
{
distinct_id="some-distinct-id2",
properties={
"$feature_flag": "complex-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -2767,9 +2838,9 @@ class TestCaptureCalls(unittest.TestCase):
self.assertEqual(patch_flags.call_count, 1)
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id2",
"$feature_flag_called",
{
distinct_id="some-distinct-id2",
properties={
"$feature_flag": "decide-flag",
"$feature_flag_response": "decide-value",
"locally_evaluated": False,
@@ -2820,9 +2891,9 @@ class TestCaptureCalls(unittest.TestCase):
)
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "decide-flag",
"$feature_flag_response": "decide-variant",
"locally_evaluated": False,
@@ -2871,9 +2942,9 @@ class TestCaptureCalls(unittest.TestCase):
)
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "decide-flag-with-payload",
"$feature_flag_response": True,
"locally_evaluated": False,
@@ -2948,7 +3019,9 @@ class TestCaptureCalls(unittest.TestCase):
"featureFlags": {"person-flag": True},
"featureFlagPayloads": {"person-flag": 300},
}
client = Client(api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
client = Client(
project_api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY
)
client.feature_flags = [
{
@@ -2977,9 +3050,9 @@ class TestCaptureCalls(unittest.TestCase):
# Assert that capture was called once, with the correct parameters
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -3012,9 +3085,9 @@ class TestCaptureCalls(unittest.TestCase):
self.assertEqual(patch_capture.call_count, 1)
patch_capture.assert_called_with(
"some-distinct-id2",
"$feature_flag_called",
{
distinct_id="some-distinct-id2",
properties={
"$feature_flag": "person-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -3058,9 +3131,9 @@ class TestCaptureCalls(unittest.TestCase):
)
patch_capture.assert_called_with(
"some-distinct-id",
"$feature_flag_called",
{
distinct_id="some-distinct-id",
properties={
"$feature_flag": "complex-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -3102,9 +3175,9 @@ class TestCaptureCalls(unittest.TestCase):
person_properties={"region": "USA", "name": "Aloha"},
)
patch_capture.assert_called_with(
distinct_id,
"$feature_flag_called",
{
distinct_id=distinct_id,
properties={
"$feature_flag": "complex-flag",
"$feature_flag_response": True,
"locally_evaluated": True,
@@ -5229,7 +5302,9 @@ class TestConsistency(unittest.TestCase):
"featureFlags": {}
} # Ensure decide returns empty flags
client = Client(api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
client = Client(
project_api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY
)
client.feature_flags = [
{
"id": 1,
@@ -5253,7 +5328,9 @@ class TestConsistency(unittest.TestCase):
"featureFlagPayloads": {"Beta-Feature": {"some": "value"}},
}
client = Client(api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
client = Client(
project_api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY
)
client.feature_flags = [
{
"id": 1,
@@ -5282,7 +5359,9 @@ class TestConsistency(unittest.TestCase):
"featureFlagPayloads": {"Beta-Feature": {"some": "value"}},
}
client = Client(api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
client = Client(
project_api_key=FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY
)
client.feature_flags = [
{
"id": 1,
+2 -12
View File
@@ -7,8 +7,7 @@ class TestModule(unittest.TestCase):
posthog = None
def _assert_enqueue_result(self, result):
self.assertEqual(type(result[0]), bool)
self.assertEqual(type(result[1]), dict)
self.assertEqual(type(result[0]), str)
def failed(self):
self.failed = True
@@ -28,12 +27,7 @@ class TestModule(unittest.TestCase):
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"})
res = self.posthog.capture("python module event", distinct_id="distinct_id")
self._assert_enqueue_result(res)
self.posthog.flush()
@@ -42,9 +36,5 @@ class TestModule(unittest.TestCase):
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()
+123
View File
@@ -1,3 +1,4 @@
import time
import unittest
from dataclasses import dataclass
from datetime import date, datetime, timedelta
@@ -12,6 +13,7 @@ from pydantic import BaseModel
from pydantic.v1 import BaseModel as BaseModelV1
from posthog import utils
from posthog.types import FeatureFlagResult
TEST_API_KEY = "kOOlRy2QlMY9jHZQv0bKz0FZyazBUoY8Arj0lFVNjs4"
FAKE_TEST_API_KEY = "random_key"
@@ -173,3 +175,124 @@ class TestUtils(unittest.TestCase):
"inner_optional": None,
},
}
class TestFlagCache(unittest.TestCase):
def setUp(self):
self.cache = utils.FlagCache(max_size=3, default_ttl=1)
self.flag_result = FeatureFlagResult.from_value_and_payload(
"test-flag", True, None
)
def test_cache_basic_operations(self):
distinct_id = "user123"
flag_key = "test-flag"
flag_version = 1
# Test cache miss
result = self.cache.get_cached_flag(distinct_id, flag_key, flag_version)
assert result is None
# Test cache set and hit
self.cache.set_cached_flag(
distinct_id, flag_key, self.flag_result, flag_version
)
result = self.cache.get_cached_flag(distinct_id, flag_key, flag_version)
assert result is not None
assert result.get_value()
def test_cache_ttl_expiration(self):
distinct_id = "user123"
flag_key = "test-flag"
flag_version = 1
# Set flag in cache
self.cache.set_cached_flag(
distinct_id, flag_key, self.flag_result, flag_version
)
# Should be available immediately
result = self.cache.get_cached_flag(distinct_id, flag_key, flag_version)
assert result is not None
# Wait for TTL to expire (1 second + buffer)
time.sleep(1.1)
# Should be expired
result = self.cache.get_cached_flag(distinct_id, flag_key, flag_version)
assert result is None
def test_cache_version_invalidation(self):
distinct_id = "user123"
flag_key = "test-flag"
old_version = 1
new_version = 2
# Set flag with old version
self.cache.set_cached_flag(distinct_id, flag_key, self.flag_result, old_version)
# Should hit with old version
result = self.cache.get_cached_flag(distinct_id, flag_key, old_version)
assert result is not None
# Should miss with new version
result = self.cache.get_cached_flag(distinct_id, flag_key, new_version)
assert result is None
# Invalidate old version
self.cache.invalidate_version(old_version)
# Should miss even with old version after invalidation
result = self.cache.get_cached_flag(distinct_id, flag_key, old_version)
assert result is None
def test_stale_cache_functionality(self):
distinct_id = "user123"
flag_key = "test-flag"
flag_version = 1
# Set flag in cache
self.cache.set_cached_flag(
distinct_id, flag_key, self.flag_result, flag_version
)
# Wait for TTL to expire
time.sleep(1.1)
# Should not get fresh cache
result = self.cache.get_cached_flag(distinct_id, flag_key, flag_version)
assert result is None
# Should get stale cache (within 1 hour default)
stale_result = self.cache.get_stale_cached_flag(distinct_id, flag_key)
assert stale_result is not None
assert stale_result.get_value()
def test_lru_eviction(self):
# Cache has max_size=3, so adding 4 users should evict the LRU one
flag_version = 1
# Add 3 users
for i in range(3):
user_id = f"user{i}"
self.cache.set_cached_flag(
user_id, "test-flag", self.flag_result, flag_version
)
# Access user0 to make it recently used
self.cache.get_cached_flag("user0", "test-flag", flag_version)
# Add 4th user, should evict user1 (least recently used)
self.cache.set_cached_flag("user3", "test-flag", self.flag_result, flag_version)
# user0 should still be there (was recently accessed)
result = self.cache.get_cached_flag("user0", "test-flag", flag_version)
assert result is not None
# user2 should still be there (was recently added)
result = self.cache.get_cached_flag("user2", "test-flag", flag_version)
assert result is not None
# user3 should be there (just added)
result = self.cache.get_cached_flag("user3", "test-flag", flag_version)
assert result is not None
+18
View File
@@ -9,6 +9,24 @@ FlagValue = Union[bool, str]
BeforeSendCallback = Callable[[dict[str, Any]], Optional[dict[str, Any]]]
class SendFeatureFlagsOptions(TypedDict, total=False):
"""Options for sending feature flags with capture events.
Args:
only_evaluate_locally: Whether to only use local evaluation for feature flags.
If True, only flags that can be evaluated locally will be included.
If False, remote evaluation via /flags API will be used when needed.
person_properties: Properties to use for feature flag evaluation specific to this event.
These properties will be merged with any existing person properties.
group_properties: Group properties to use for feature flag evaluation specific to this event.
Format: { group_type_name: { group_properties } }
"""
only_evaluate_locally: Optional[bool]
person_properties: Optional[dict[str, Any]]
group_properties: Optional[dict[str, dict[str, Any]]]
@dataclass(frozen=True)
class FlagReason:
code: str
+319
View File
@@ -1,12 +1,17 @@
import json
import logging
import numbers
import re
import time
from collections import defaultdict
from dataclasses import asdict, is_dataclass
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Any, Optional
from uuid import UUID
import sys
import platform
import distro # For Linux OS detection
import six
from dateutil.tz import tzlocal, tzutc
@@ -154,6 +159,266 @@ class SizeLimitedDict(defaultdict):
super().__setitem__(key, value)
class FlagCacheEntry:
def __init__(self, flag_result, flag_definition_version, timestamp=None):
self.flag_result = flag_result
self.flag_definition_version = flag_definition_version
self.timestamp = timestamp or time.time()
def is_valid(self, current_time, ttl, current_flag_version):
time_valid = (current_time - self.timestamp) < ttl
version_valid = self.flag_definition_version == current_flag_version
return time_valid and version_valid
def is_stale_but_usable(self, current_time, max_stale_age=3600):
return (current_time - self.timestamp) < max_stale_age
class FlagCache:
def __init__(self, max_size=10000, default_ttl=300):
self.cache = {} # distinct_id -> {flag_key: FlagCacheEntry}
self.access_times = {} # distinct_id -> last_access_time
self.max_size = max_size
self.default_ttl = default_ttl
def get_cached_flag(self, distinct_id, flag_key, current_flag_version):
current_time = time.time()
if distinct_id not in self.cache:
return None
user_flags = self.cache[distinct_id]
if flag_key not in user_flags:
return None
entry = user_flags[flag_key]
if entry.is_valid(current_time, self.default_ttl, current_flag_version):
self.access_times[distinct_id] = current_time
return entry.flag_result
return None
def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=3600):
current_time = time.time()
if distinct_id not in self.cache:
return None
user_flags = self.cache[distinct_id]
if flag_key not in user_flags:
return None
entry = user_flags[flag_key]
if entry.is_stale_but_usable(current_time, max_stale_age):
return entry.flag_result
return None
def set_cached_flag(
self, distinct_id, flag_key, flag_result, flag_definition_version
):
current_time = time.time()
# Evict LRU users if we're at capacity
if distinct_id not in self.cache and len(self.cache) >= self.max_size:
self._evict_lru()
# Initialize user cache if needed
if distinct_id not in self.cache:
self.cache[distinct_id] = {}
# Store the flag result
self.cache[distinct_id][flag_key] = FlagCacheEntry(
flag_result, flag_definition_version, current_time
)
self.access_times[distinct_id] = current_time
def invalidate_version(self, old_version):
users_to_remove = []
for distinct_id, user_flags in self.cache.items():
flags_to_remove = []
for flag_key, entry in user_flags.items():
if entry.flag_definition_version == old_version:
flags_to_remove.append(flag_key)
# Remove invalidated flags
for flag_key in flags_to_remove:
del user_flags[flag_key]
# Remove user entirely if no flags remain
if not user_flags:
users_to_remove.append(distinct_id)
# Clean up empty users
for distinct_id in users_to_remove:
del self.cache[distinct_id]
if distinct_id in self.access_times:
del self.access_times[distinct_id]
def _evict_lru(self):
if not self.access_times:
return
# Remove 20% of least recently used entries
sorted_users = sorted(self.access_times.items(), key=lambda x: x[1])
to_remove = max(1, len(sorted_users) // 5)
for distinct_id, _ in sorted_users[:to_remove]:
if distinct_id in self.cache:
del self.cache[distinct_id]
if distinct_id in self.access_times:
del self.access_times[distinct_id]
def clear(self):
self.cache.clear()
self.access_times.clear()
class RedisFlagCache:
def __init__(
self, redis_client, default_ttl=300, stale_ttl=3600, key_prefix="posthog:flags:"
):
self.redis = redis_client
self.default_ttl = default_ttl
self.stale_ttl = stale_ttl
self.key_prefix = key_prefix
self.version_key = f"{key_prefix}version"
def _get_cache_key(self, distinct_id, flag_key):
return f"{self.key_prefix}{distinct_id}:{flag_key}"
def _serialize_entry(self, flag_result, flag_definition_version, timestamp=None):
if timestamp is None:
timestamp = time.time()
# Use clean to make flag_result JSON-serializable for cross-platform compatibility
serialized_result = clean(flag_result)
entry = {
"flag_result": serialized_result,
"flag_version": flag_definition_version,
"timestamp": timestamp,
}
return json.dumps(entry)
def _deserialize_entry(self, data):
try:
entry = json.loads(data)
flag_result = entry["flag_result"]
return FlagCacheEntry(
flag_result=flag_result,
flag_definition_version=entry["flag_version"],
timestamp=entry["timestamp"],
)
except (json.JSONDecodeError, KeyError, ValueError):
# If deserialization fails, treat as cache miss
return None
def get_cached_flag(self, distinct_id, flag_key, current_flag_version):
try:
cache_key = self._get_cache_key(distinct_id, flag_key)
data = self.redis.get(cache_key)
if data:
entry = self._deserialize_entry(data)
if entry and entry.is_valid(
time.time(), self.default_ttl, current_flag_version
):
return entry.flag_result
return None
except Exception:
# Redis error - return None to fall back to normal evaluation
return None
def get_stale_cached_flag(self, distinct_id, flag_key, max_stale_age=None):
try:
if max_stale_age is None:
max_stale_age = self.stale_ttl
cache_key = self._get_cache_key(distinct_id, flag_key)
data = self.redis.get(cache_key)
if data:
entry = self._deserialize_entry(data)
if entry and entry.is_stale_but_usable(time.time(), max_stale_age):
return entry.flag_result
return None
except Exception:
# Redis error - return None
return None
def set_cached_flag(
self, distinct_id, flag_key, flag_result, flag_definition_version
):
try:
cache_key = self._get_cache_key(distinct_id, flag_key)
serialized_entry = self._serialize_entry(
flag_result, flag_definition_version
)
# Set with TTL for automatic cleanup (use stale_ttl for total lifetime)
self.redis.setex(cache_key, self.stale_ttl, serialized_entry)
# Update the current version
self.redis.set(self.version_key, flag_definition_version)
except Exception:
# Redis error - silently fail, don't break flag evaluation
pass
def invalidate_version(self, old_version):
try:
# For Redis, we use a simple approach: scan for keys with old version
# and delete them. This could be expensive with many keys, but it's
# necessary for correctness.
cursor = 0
pattern = f"{self.key_prefix}*"
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
for key in keys:
if key.decode() == self.version_key:
continue
try:
data = self.redis.get(key)
if data:
entry_dict = json.loads(data)
if entry_dict.get("flag_version") == old_version:
self.redis.delete(key)
except (json.JSONDecodeError, KeyError):
# If we can't parse the entry, delete it to be safe
self.redis.delete(key)
if cursor == 0:
break
except Exception:
# Redis error - silently fail
pass
def clear(self):
try:
# Delete all keys matching our pattern
cursor = 0
pattern = f"{self.key_prefix}*"
while True:
cursor, keys = self.redis.scan(cursor, match=pattern, count=100)
if keys:
self.redis.delete(*keys)
if cursor == 0:
break
except Exception:
# Redis error - silently fail
pass
def convert_to_datetime_aware(date_obj):
if date_obj.tzinfo is None:
date_obj = date_obj.replace(tzinfo=timezone.utc)
@@ -198,3 +463,57 @@ def str_iequals(value, comparand):
False
"""
return str(value).casefold() == str(comparand).casefold()
def get_os_info():
"""
Returns standardized OS name and version information.
Similar to how user agent parsing works in JS.
"""
os_name = ""
os_version = ""
platform_name = sys.platform
if platform_name.startswith("win"):
os_name = "Windows"
if hasattr(platform, "win32_ver"):
win_version = platform.win32_ver()[0]
if win_version:
os_version = win_version
elif platform_name == "darwin":
os_name = "Mac OS X"
if hasattr(platform, "mac_ver"):
mac_version = platform.mac_ver()[0]
if mac_version:
os_version = mac_version
elif platform_name.startswith("linux"):
os_name = "Linux"
linux_info = distro.info()
if linux_info["version"]:
os_version = linux_info["version"]
elif platform_name.startswith("freebsd"):
os_name = "FreeBSD"
if hasattr(platform, "release"):
os_version = platform.release()
else:
os_name = platform_name
if hasattr(platform, "release"):
os_version = platform.release()
return os_name, os_version
def system_context() -> dict[str, Any]:
os_name, os_version = get_os_info()
return {
"$python_runtime": platform.python_implementation(),
"$python_version": "%s.%s.%s" % (sys.version_info[:3]),
"$os": os_name,
"$os_version": os_version,
}
+1 -1
View File
@@ -1,4 +1,4 @@
VERSION = "5.4.0"
VERSION = "6.3.0"
if __name__ == "__main__":
print(VERSION, end="") # noqa: T201
+1 -1
View File
@@ -29,6 +29,7 @@ dependencies = [
"python-dateutil>=2.2",
"backoff>=1.10.0",
"distro>=1.5.0",
"typing-extensions>=4.2.0",
]
[project.urls]
@@ -86,7 +87,6 @@ packages = [
"posthog.ai.anthropic",
"posthog.ai.gemini",
"posthog.test",
"posthog.exception_integrations",
"posthog.integrations",
]
-112
View File
@@ -1,112 +0,0 @@
import argparse
import json
import logging
import posthog
__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]
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("--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("--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("--traits", help="the identify/group traits to send (JSON-encoded)")
parser.add_argument("--groupId", help="the group id")
options = parser.parse_args()
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),
)
def page():
posthog.page(
options.distinct_id,
name=options.name,
anonymous_id=options.anonymousId,
properties=json_hash(options.properties),
)
def identify():
posthog.identify(
options.distinct_id,
anonymous_id=options.anonymousId,
traits=json_hash(options.traits),
)
def set_once():
posthog.set_once(
options.distinct_id,
properties=json_hash(options.traits),
)
def set():
posthog.set(
options.distinct_id,
properties=json_hash(options.traits),
)
def unknown():
print()
posthog.api_key = options.writeKey
posthog.on_error = failed
posthog.debug = True
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,
}
func = switcher.get(options.type)
if func:
func()
posthog.shutdown()
else:
print("Invalid Message Type " + options.type)
Generated
+2082 -2151
View File
File diff suppressed because it is too large Load Diff