Compare commits

...
Author SHA1 Message Date
Radu Raicea 2f1ac45f08 WIP 2025-10-31 15:12:27 -04:00
Julian BezandGitHub 50b0c7170a fix(django): restore process_exception to capture view exceptions (#350)
Restores the process_exception method that was removed in v6.7.5 (PR #328),
which broke exception capture from Django views and downstream middleware.

Django converts view exceptions into responses before they propagate through
the middleware stack's __call__ method, so the context manager's exception
handler never sees them. Django provides these exceptions via the
process_exception hook instead.

Changes:
- Add process_exception method to capture exceptions from views and downstream
  middleware with proper request context and tags
- Add tests verifying process_exception behavior and settings (capture_exceptions,
  request_filter)
2025-10-29 10:40:08 +00:00
github-actions[bot] f719c3dadf Update generated references 2025-10-28 13:06:45 +00:00
Andrew MaguireandGitHub 105090a6ba chore: bump version to 6.7.11 for AI framework feature (#354)
Update version and changelog for PR #347
2025-10-28 13:05:50 +00:00
edfadcc6a8 feat(ai): Add $ai_framework property for framework integrations (#347)
* Add $ai_lib_metadata to AI integrations

Adds framework identification metadata to all AI events for easier filtering
and analytics. Each integration now includes a $ai_lib_metadata property with
schema version and framework name.

- LangChain: Hardcoded to "langchain"
- Native wrappers (Anthropic, OpenAI, Gemini): Uses provider name
- Ready for future frameworks (pydantic-ai, crewai, llamaindex)

This enables PostHog queries to easily distinguish between:
- Direct SDK wrapper usage
- Framework-mediated usage (LangChain, etc.)
- Different framework types

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Add \$ai_lib_metadata to sync/async paths and tests

- Added \$ai_lib_metadata to call_llm_and_track_usage (sync)
- Added \$ai_lib_metadata to call_llm_and_track_usage_async (async)
- Added test assertion in test_basic_completion
- Placed metadata at end of properties for consistency

All tests pass successfully.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Refactor: use unified utility function for $ai_lib_metadata

Creates a single `get_ai_lib_metadata(framework)` utility function to generate
the $ai_lib_metadata object, replacing inline implementations across the
codebase.

Changes:
- Add get_ai_lib_metadata() utility to utils.py
- Update LangChain callbacks to use utility function
- Update call_llm_and_track_usage() to use utility function
- Update call_llm_and_track_usage_async() to use utility function
- Update capture_streaming_event() to use utility function

Benefits:
- Consistency across all integrations
- Single source of truth for metadata structure
- Easier to extend with version detection later

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Add $ai_lib_metadata assertions to provider tests

Add missing $ai_lib_metadata assertions to Anthropic, Gemini, and LangChain tests to match the validation already present in OpenAI tests. Each test now verifies the metadata field contains the correct schema version and framework name.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Simplify to $ai_framework property, only for actual frameworks

Changes:
- Replace complex $ai_lib_metadata object with simple $ai_framework string
- Only include $ai_framework when using actual framework (LangChain)
- Remove $ai_framework from direct provider calls (OpenAI, Anthropic, Gemini)
- Update all tests to reflect new behavior

Before: {"schema": "v1", "frameworks": [{"name": "langchain"}]}
After: "langchain" (only when using LangChain framework)

This eliminates wasteful redundancy where framework=provider for direct calls.

---------

Co-authored-by: Claude <noreply@anthropic.com>
2025-10-28 09:59:20 +00:00
Julian BezandGitHub 13184e2e16 chore: standardize workflow extensions to .yml (#349)
Rename workflow files from .yaml to .yml for consistency with existing
workflows (ci.yml, call-flags-project-board.yml).

This resolves naming confusion and standardizes all GitHub Actions
workflow files to use the .yml extension.
2025-10-24 14:17:27 +00:00
webjunkieandgithub-actions[bot] 1b8642331f Update generated references 2025-10-24 14:16:57 +00:00
Julian BezandGitHub 6af129f414 fix(django): make middleware truly hybrid-compatible with sync and async Django stacks (#348)
Address code review feedback and critical issues from PR #328.

Changes:
- Keep __call__ as sync method that conditionally routes to __acall__ for async paths
- Use markcoroutinefunction() to properly mark instances when async is detected
- Detect async/sync at init time via iscoroutinefunction(get_response)
- Remove process_exception method - it was non-functional (Django doesn't call it on new-style middleware without MiddlewareMixin)
- Fix markcoroutinefunction fallback to be a simple no-op instead of accessing private API
- Exception capture works correctly via contexts.new_context() which has built-in exception handling
- Add comprehensive test coverage for sync, async, and hybrid middleware behavior
- Add async exception capture tests
- Refactor tests to use proper middleware initialization

This implementation follows Django's recommended hybrid middleware pattern where
both sync_capable and async_capable are True, allowing Django to pass requests
without conversion while the middleware adapts based on the detected mode.

The sync path behavior is identical to version 6.7.4 (pre-async), ensuring perfect
backward compatibility for WSGI deployments.

Addresses #329
Related to #328
2025-10-24 15:50:21 +02:00
Phil HaackandGitHub 02e82a6050 Bump version to 6.7.9 (#345) 2025-10-22 20:53:07 +00:00
Phil HaackandGitHub 9a05db8b20 fix(flags): multi-condition flags with static cohorts returning wrong variants (#343)
* Fix multi-condition flags with static cohorts returning wrong variants

When a feature flag has multiple conditions and one contains a static
cohort, the SDK now correctly falls back to the API instead of
evaluating subsequent conditions locally and returning incorrect variants.

Introduce RequiresServerEvaluation exception to distinguish between:
- Missing server-side data (static cohorts) → immediate API fallback
- Evaluation errors (bad regex, missing properties) → try next condition

Changes:
- Add RequiresServerEvaluation exception class
- Update match_cohort() to throw RequiresServerEvaluation for static cohorts
- Update match_property_group() to propagate RequiresServerEvaluation
- Update match_feature_flag_properties() to handle both exception types
- Update client.py to catch both exceptions for API fallback
- Export RequiresServerEvaluation in __init__.py
- Add test for multi-condition static cohort scenario

All 84 feature flag tests pass.

* Add unit test for payloads

* ruff format
2025-10-21 13:40:36 -07:00
Radu RaiceaandGitHub e06830e068 fix(llma): missing await in OpenAI's streaming implementation (#342)
* fix(llma): missing async for OpenAI async

* chore(llma): bump version

* chore(llma): bump version
2025-10-16 14:46:26 +00:00
gewenyu99andgithub-actions[bot] 465baea6f8 Update generated references 2025-10-16 00:02:49 +00:00
Vincent (Wen Yu) GeandGitHub 2bd6e9eaf1 fix: Check for references directory and generate 6.7.7 specs (#341)
* Fix check for directory and generate 6.7.7 specs

* Delete references
2025-10-15 20:02:16 -04:00
Vincent (Wen Yu) GeandGitHub e6fe39a0dd Run SDK generation after release job (#340)
* run after release job

* use bot pat

* Fix token placement

* Run this with a gh cli command
2025-10-15 19:45:33 -04:00
21 changed files with 11033 additions and 166 deletions
@@ -1,15 +1,11 @@
name: "Generate References"
on:
release:
types: [published]
workflow_dispatch:
jobs:
docs-generation:
name: Generate references
permissions:
contents: write
runs-on: ubuntu-latest
steps:
- name: Checkout the repository
@@ -45,7 +41,7 @@ jobs:
echo "No new references generated in references directory"
fi
- uses: stefanzweifel/git-auto-commit-action@b3e3f72439fc3af08948f989a19a825463598a
- uses: stefanzweifel/git-auto-commit-action@778341af668090896ca464160c2def5d1d1a3eb0
if: steps.changes.outputs.changed == 'true'
with:
commit_message: "Update generated references"
@@ -49,3 +49,9 @@ jobs:
with:
tag_name: v${{ env.REPO_VERSION }}
release_name: ${{ env.REPO_VERSION }}
- name: Dispatch generate-references for posthog-python
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
gh workflow run generate-references.yml --ref master
+20
View File
@@ -1,3 +1,23 @@
# Unreleased
- fix(django): Restore process_exception method to capture view and downstream middleware exceptions (fixes #329)
# 6.7.11 - 2025-10-28
- feat(ai): Add `$ai_framework` property for framework integrations (e.g. LangChain)
# 6.7.10 - 2025-10-24
- fix(django): Make middleware truly hybrid - compatible with both sync (WSGI) and async (ASGI) Django stacks without breaking sync-only deployments
# 6.7.9 - 2025-10-22
- fix(flags): multi-condition flags with static cohorts returning wrong variants
# 6.7.8 - 2025-10-16
- fix(llma): missing async for OpenAI's streaming implementation
# 6.7.7 - 2025-10-14
- fix: remove deprecated attribute $exception_personURL from exception events
@@ -460,6 +460,10 @@ if __name__ == "__main__":
try:
documentation = generate_sdk_documentation()
# Ensure output directory exists
output_dir = str(OUTPUT_CONFIG["output_dir"])
os.makedirs(output_dir, exist_ok=True)
output_file = os.path.join(
str(OUTPUT_CONFIG["output_dir"]), str(OUTPUT_CONFIG["filename"])
)
+91 -2
View File
@@ -2,7 +2,13 @@ import datetime # 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.args import (
OptionalCaptureArgs,
OptionalSetArgs,
OptionalCaptureAIArgs,
AI_EVENT_TYPE,
ExceptionArg,
)
from posthog.client import Client
from posthog.contexts import (
new_context as inner_new_context,
@@ -11,7 +17,15 @@ from posthog.contexts import (
set_context_session as inner_set_context_session,
identify_context as inner_identify_context,
)
from posthog.types import FeatureFlag, FlagsAndPayloads, FeatureFlagResult
from posthog.feature_flags import (
InconclusiveMatchError as InconclusiveMatchError,
RequiresServerEvaluation as RequiresServerEvaluation,
)
from posthog.types import (
FeatureFlag,
FlagsAndPayloads,
FeatureFlagResult as FeatureFlagResult,
)
from posthog.version import VERSION
__version__ = VERSION
@@ -385,6 +399,81 @@ def capture_exception(
return _proxy("capture_exception", exception=exception, **kwargs)
def capture_ai(
event: AI_EVENT_TYPE, **kwargs: Unpack[OptionalCaptureAIArgs]
) -> Optional[str]:
"""
Capture an AI event to the dedicated AI endpoint with support for large payloads.
This method sends AI events (like $ai_generation, $ai_trace, etc.) to PostHog's
specialized AI endpoint (/i/v0/ai) which supports large payloads through multipart/form-data
and blob storage in S3.
Args:
event: The AI event type. Must be one of: "$ai_generation", "$ai_trace", "$ai_span",
"$ai_embedding", "$ai_metric", "$ai_feedback"
distinct_id: The distinct ID of the user.
properties: A dictionary of AI event properties. Must include required properties based on event type:
- All events: "$ai_model" (required)
- $ai_generation: "$ai_provider", "$ai_trace_id" (required)
- $ai_trace: "$ai_trace_id" (required)
- $ai_span: "$ai_trace_id", "$ai_span_id" (required)
- $ai_embedding: "$ai_provider", "$ai_trace_id" (required)
blob_properties: List of property names to send as blobs (large data stored in S3).
Common blob properties: "$ai_input", "$ai_output_choices", "$ai_input_state", "$ai_output_state"
If not provided, defaults to common blob properties based on event type.
timestamp: The timestamp of the event.
uuid: A unique identifier for the event.
groups: A dictionary of group information.
disable_geoip: Whether to disable GeoIP for this event.
Examples:
```python
# $ai_generation event with blobs
from posthog import capture_ai
capture_ai(
"$ai_generation",
distinct_id="user_123",
properties={
"$ai_model": "gpt-4",
"$ai_provider": "openai",
"$ai_trace_id": "trace_abc123",
"$ai_input": {
"messages": [
{"role": "user", "content": "Hello!"}
]
},
"$ai_output_choices": {
"choices": [{"message": {"role": "assistant", "content": "Hi there!"}}]
},
"$ai_completion_tokens": 150,
"$ai_prompt_tokens": 50
},
blob_properties=["$ai_input", "$ai_output_choices"]
)
```
```python
# $ai_trace event
from posthog import capture_ai
capture_ai(
"$ai_trace",
distinct_id="user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": "trace_abc123"
}
)
```
Category:
AI Events
Note: This method sends events synchronously to the AI endpoint, bypassing the queue system.
"""
return _proxy("capture_ai", event, **kwargs)
def feature_enabled(
key, # type: str
distinct_id, # type: str
+2
View File
@@ -486,6 +486,7 @@ class CallbackHandler(BaseCallbackHandler):
"$ai_latency": run.latency,
"$ai_span_name": run.name,
"$ai_span_id": run_id,
"$ai_framework": "langchain",
}
if parent_run_id is not None:
event_properties["$ai_parent_id"] = parent_run_id
@@ -556,6 +557,7 @@ class CallbackHandler(BaseCallbackHandler):
"$ai_http_status": 200,
"$ai_latency": run.latency,
"$ai_base_url": run.base_url,
"$ai_framework": "langchain",
}
if run.tools:
+3 -3
View File
@@ -128,7 +128,7 @@ class WrappedResponses:
start_time = time.time()
usage_stats: TokenUsage = TokenUsage()
final_content = []
response = self._original.create(**kwargs)
response = await self._original.create(**kwargs)
async def async_generator():
nonlocal usage_stats
@@ -345,7 +345,7 @@ class WrappedCompletions:
if "stream_options" not in kwargs:
kwargs["stream_options"] = {}
kwargs["stream_options"]["include_usage"] = True
response = self._original.create(**kwargs)
response = await self._original.create(**kwargs)
async def async_generator():
nonlocal usage_stats
@@ -499,7 +499,7 @@ class WrappedEmbeddings:
posthog_trace_id = str(uuid.uuid4())
start_time = time.time()
response = self._original.create(**kwargs)
response = await self._original.create(**kwargs)
end_time = time.time()
# Extract usage statistics if available
+31 -1
View File
@@ -1,4 +1,4 @@
from typing import TypedDict, Optional, Any, Dict, Union, Tuple, Type
from typing import TypedDict, Optional, Any, Dict, List, Union, Tuple, Type
from types import TracebackType
from typing_extensions import NotRequired # For Python < 3.11 compatibility
from datetime import datetime
@@ -69,3 +69,33 @@ ExcInfo = Union[
]
ExceptionArg = Union[BaseException, ExcInfo]
# AI Event Types (literal strings to enforce valid event types)
AI_EVENT_TYPE = Union[
str, # Allow str for flexibility but document the expected values
] # "$ai_generation", "$ai_trace", "$ai_span", "$ai_embedding", "$ai_metric", "$ai_feedback"
class OptionalCaptureAIArgs(TypedDict):
"""Optional arguments for the capture_ai method.
Args:
distinct_id: Unique identifier for the person associated with this AI event. If not set, the context
distinct_id is used, if available, otherwise a UUID is generated.
properties: Dictionary of AI event properties to track. Must include required properties for the event type.
blob_properties: List of property names that should be sent as blobs (e.g., '$ai_input', '$ai_output_choices').
These properties will be extracted from `properties` and sent as multipart blobs.
timestamp: When the event occurred (defaults to current time)
uuid: Unique identifier for this specific event. If not provided, one is generated.
groups: Group identifiers to associate with this event (format: {group_type: group_key})
disable_geoip: Whether to disable GeoIP lookup for this event.
"""
distinct_id: NotRequired[Optional[ID_TYPES]]
properties: NotRequired[Optional[Dict[str, Any]]]
blob_properties: NotRequired[Optional[List[str]]]
timestamp: NotRequired[Optional[Union[datetime, str]]]
uuid: NotRequired[Optional[str]]
groups: NotRequired[Optional[Dict[str, str]]]
disable_geoip: NotRequired[Optional[bool]]
+307 -4
View File
@@ -1,16 +1,25 @@
import atexit
import json
import logging
import os
import sys
from datetime import datetime, timedelta
from typing import Any, Callable, Dict, Optional, Union
from io import BytesIO
from typing import Any, Dict, List, Optional, Tuple, Union
from typing_extensions import Unpack
from uuid import uuid4
from dateutil.tz import tzutc
from six import string_types
from posthog.args import OptionalCaptureArgs, OptionalSetArgs, ID_TYPES, ExceptionArg
from posthog.args import (
OptionalCaptureArgs,
OptionalSetArgs,
OptionalCaptureAIArgs,
AI_EVENT_TYPE,
ID_TYPES,
ExceptionArg,
)
from posthog.consumer import Consumer
from posthog.exception_capture import ExceptionCapture
from posthog.exception_utils import (
@@ -20,7 +29,11 @@ from posthog.exception_utils import (
exception_is_already_captured,
mark_exception_as_captured,
)
from posthog.feature_flags import InconclusiveMatchError, match_feature_flag_properties
from posthog.feature_flags import (
InconclusiveMatchError,
RequiresServerEvaluation,
match_feature_flag_properties,
)
from posthog.poller import Poller
from posthog.request import (
DEFAULT_HOST,
@@ -99,6 +112,106 @@ def add_context_tags(properties):
return properties
def _generate_multipart_boundary() -> str:
"""Generate a random boundary string for multipart requests."""
return f"----WebKitFormBoundary{uuid4().hex[:16]}"
def _encode_multipart_part(
boundary: str,
name: str,
content: bytes,
content_type: str,
filename: Optional[str] = None,
) -> bytes:
"""Encode a single part of a multipart/form-data request."""
part = f"--{boundary}\r\n".encode("utf-8")
if filename:
part += f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode(
"utf-8"
)
else:
part += f'Content-Disposition: form-data; name="{name}"\r\n'.encode("utf-8")
part += f"Content-Type: {content_type}\r\n\r\n".encode("utf-8")
part += content
part += b"\r\n"
return part
def _build_multipart_body(
event_data: Dict[str, Any],
properties: Optional[Dict[str, Any]],
blob_properties: Optional[List[str]],
) -> Tuple[bytes, str]:
"""
Build a multipart/form-data body for AI capture endpoint.
Args:
event_data: The event data (uuid, event, distinct_id, timestamp)
properties: Event properties (small properties)
blob_properties: List of property names to send as blobs
Returns:
Tuple of (body_bytes, content_type_header)
"""
boundary = _generate_multipart_boundary()
body = BytesIO()
# Part 1: Event (required, must be first)
event_json = json.dumps(event_data).encode("utf-8")
body.write(
_encode_multipart_part(boundary, "event", event_json, "application/json")
)
# Separate properties into small properties and blob properties
small_properties = {}
blob_data = {}
if properties:
blob_property_names = set(blob_properties or [])
for key, value in properties.items():
if key in blob_property_names:
blob_data[key] = value
else:
small_properties[key] = value
# Part 2: Event properties (if there are any small properties)
if small_properties:
properties_json = json.dumps(small_properties).encode("utf-8")
body.write(
_encode_multipart_part(
boundary, "event.properties", properties_json, "application/json"
)
)
# Part 3+: Blob parts
for property_name, property_value in blob_data.items():
# Serialize the blob data as JSON
blob_json = json.dumps(property_value).encode("utf-8")
blob_filename = f"blob_{uuid4().hex[:8]}"
body.write(
_encode_multipart_part(
boundary,
f"event.properties.{property_name}",
blob_json,
"application/json",
filename=blob_filename,
)
)
# Final boundary
body.write(f"--{boundary}--\r\n".encode("utf-8"))
body_bytes = body.getvalue()
content_type = f"multipart/form-data; boundary={boundary}"
return body_bytes, content_type
def no_throw(default_return=None):
"""
Decorator to prevent raising exceptions from public API methods.
@@ -1001,6 +1114,196 @@ class Client(object):
except Exception as e:
self.log.exception(f"Failed to capture exception: {e}")
@no_throw()
def capture_ai(
self, event: AI_EVENT_TYPE, **kwargs: Unpack[OptionalCaptureAIArgs]
) -> Optional[str]:
"""
Capture an AI event to the dedicated AI endpoint with support for large payloads.
This method sends AI events (like $ai_generation, $ai_trace, etc.) to PostHog's
specialized AI endpoint (/i/v0/ai) which supports large payloads through multipart/form-data
and blob storage in S3.
Args:
event: The AI event type. Must be one of: "$ai_generation", "$ai_trace", "$ai_span",
"$ai_embedding", "$ai_metric", "$ai_feedback"
distinct_id: The distinct ID of the user.
properties: A dictionary of AI event properties. Must include required properties based on event type:
- All events: "$ai_model" (required)
- $ai_generation: "$ai_provider", "$ai_trace_id" (required)
- $ai_trace: "$ai_trace_id" (required)
- $ai_span: "$ai_trace_id", "$ai_span_id" (required)
- $ai_embedding: "$ai_provider", "$ai_trace_id" (required)
blob_properties: List of property names to send as blobs (large data stored in S3).
Common blob properties: "$ai_input", "$ai_output_choices", "$ai_input_state", "$ai_output_state"
If not provided, defaults to common blob properties based on event type.
timestamp: The timestamp of the event.
uuid: A unique identifier for the event.
groups: A dictionary of group information.
disable_geoip: Whether to disable GeoIP for this event.
Examples:
```python
# $ai_generation event with blobs
posthog.capture_ai(
"$ai_generation",
distinct_id="user_123",
properties={
"$ai_model": "gpt-4",
"$ai_provider": "openai",
"$ai_trace_id": "trace_abc123",
"$ai_input": {
"messages": [
{"role": "user", "content": "Hello!"}
]
},
"$ai_output_choices": {
"choices": [{"message": {"role": "assistant", "content": "Hi there!"}}]
},
"$ai_completion_tokens": 150,
"$ai_prompt_tokens": 50
},
blob_properties=["$ai_input", "$ai_output_choices"]
)
```
```python
# $ai_trace event
posthog.capture_ai(
"$ai_trace",
distinct_id="user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": "trace_abc123"
}
)
```
```python
# $ai_metric event
posthog.capture_ai(
"$ai_metric",
distinct_id="user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": "trace_abc123",
"$ai_metric_name": "accuracy",
"$ai_metric_value": "0.95"
}
)
```
Category:
AI Events
Note: This method sends events synchronously to the AI endpoint, bypassing the queue system.
"""
import requests
if self.disabled:
return None
distinct_id = kwargs.get("distinct_id", None)
properties = kwargs.get("properties", None)
timestamp = kwargs.get("timestamp", None)
uuid = kwargs.get("uuid", None)
groups = kwargs.get("groups", None)
blob_properties = kwargs.get("blob_properties", None)
# Default blob properties based on event type if not provided
if blob_properties is None:
default_blob_properties = {
"$ai_generation": ["$ai_input", "$ai_output_choices"],
"$ai_trace": ["$ai_input_state", "$ai_output_state"],
"$ai_span": ["$ai_input_state", "$ai_output_state"],
"$ai_embedding": ["$ai_input"],
}
blob_properties = default_blob_properties.get(event, [])
properties = properties or {}
# Get distinct_id
(distinct_id, personless) = get_identity_state(distinct_id)
if personless and "$process_person_profile" not in properties:
properties["$process_person_profile"] = False
# Prepare timestamp
if timestamp is None:
timestamp = datetime.now(tz=tzutc())
timestamp = guess_timezone(timestamp)
timestamp_str = timestamp.isoformat()
# Generate UUID if not provided
if uuid:
uuid_str = stringify_id(uuid)
else:
uuid_str = stringify_id(uuid4())
# Add groups to properties
if groups:
properties["$groups"] = groups
# Prepare event data (the main event part)
event_data = {
"uuid": uuid_str,
"event": event,
"distinct_id": stringify_id(distinct_id),
"timestamp": timestamp_str,
}
# Build multipart body
body_bytes, content_type = _build_multipart_body(
event_data, properties, blob_properties
)
# Send request to AI endpoint
url = remove_trailing_slash(self.host) + "/i/v0/ai"
headers = {
"Content-Type": content_type,
"Authorization": f"Bearer {self.api_key}",
"User-Agent": f"posthog-python/{VERSION}",
}
self.log.debug(f"Sending AI event to {url}: {event} (uuid: {uuid_str})")
try:
response = requests.post(
url,
data=body_bytes,
headers=headers,
timeout=self.timeout,
)
if response.status_code == 200:
self.log.debug(f"AI event captured successfully: {event}")
return uuid_str
else:
error_message = f"AI capture failed with status {response.status_code}"
try:
error_detail = response.json()
error_message = f"{error_message}: {error_detail}"
except Exception:
error_message = f"{error_message}: {response.text}"
self.log.error(error_message)
if self.debug:
raise Exception(error_message)
return None
except Exception as e:
self.log.exception(f"Error sending AI event: {e}")
if self.debug:
raise e
return None
def _enqueue(self, msg, disable_geoip):
# type: (...) -> Optional[str]
"""Push a new `msg` onto the queue, return `(success, msg)`"""
@@ -1583,7 +1886,7 @@ class Client(object):
self.log.debug(
f"Successfully computed flag locally: {key} -> {response}"
)
except InconclusiveMatchError as e:
except (RequiresServerEvaluation, InconclusiveMatchError) as e:
self.log.debug(f"Failed to compute flag {key} locally: {e}")
except Exception as e:
self.log.exception(
+25 -2
View File
@@ -22,6 +22,18 @@ class InconclusiveMatchError(Exception):
pass
class RequiresServerEvaluation(Exception):
"""
Raised when feature flag evaluation requires server-side data that is not
available locally (e.g., static cohorts, experience continuity).
This error should propagate immediately to trigger API fallback, unlike
InconclusiveMatchError which allows trying other conditions.
"""
pass
# This function takes a distinct_id and a feature flag key and returns a float between 0 and 1.
# Given the same distinct_id and key, it'll always return the same float. These floats are
# uniformly distributed between 0 and 1, so if we want to show this feature to 20% of traffic
@@ -239,7 +251,12 @@ def match_feature_flag_properties(
else:
variant = get_matching_variant(flag, distinct_id)
return variant or True
except RequiresServerEvaluation:
# Static cohort or other missing server-side data - must fallback to API
raise
except InconclusiveMatchError:
# Evaluation error (bad regex, invalid date, missing property, etc.)
# Track that we had an inconclusive match, but try other conditions
is_inconclusive = True
if is_inconclusive:
@@ -449,8 +466,8 @@ def match_cohort(
# }
cohort_id = str(property.get("value"))
if cohort_id not in cohort_properties:
raise InconclusiveMatchError(
"can't match cohort without a given cohort property value"
raise RequiresServerEvaluation(
f"cohort {cohort_id} not found in local cohorts - likely a static cohort that requires server evaluation"
)
property_group = cohort_properties[cohort_id]
@@ -503,6 +520,9 @@ def match_property_group(
# OR group
if matches:
return True
except RequiresServerEvaluation:
# Immediately propagate - this condition requires server-side data
raise
except InconclusiveMatchError as e:
log.debug(f"Failed to compute property {prop} locally: {e}")
error_matching_locally = True
@@ -552,6 +572,9 @@ def match_property_group(
return True
if not matches and negation:
return True
except RequiresServerEvaluation:
# Immediately propagate - this condition requires server-side data
raise
except InconclusiveMatchError as e:
log.debug(f"Failed to compute property {prop} locally: {e}")
error_matching_locally = True
+69 -38
View File
@@ -3,13 +3,19 @@ from posthog import contexts
from posthog.client import Client
try:
from asgiref.sync import iscoroutinefunction
from asgiref.sync import iscoroutinefunction, markcoroutinefunction
except ImportError:
# Fallback for older Django versions
# Fallback for older Django versions without asgiref
import asyncio
iscoroutinefunction = asyncio.iscoroutinefunction
# No-op fallback for markcoroutinefunction
# Older Django versions without asgiref typically don't support async middleware anyway
def markcoroutinefunction(func):
return func
if TYPE_CHECKING:
from django.http import HttpRequest, HttpResponse # noqa: F401
from typing import Callable, Dict, Any, Optional, Union, Awaitable # noqa: F401
@@ -39,26 +45,24 @@ class PosthogContextMiddleware:
See the context documentation for more information. The extracted distinct ID and session ID, if found, are used to
associate all events captured in the middleware context with the same distinct ID and session as currently active on the
frontend. See the documentation for `set_context_session` and `identify_context` for more details.
This middleware is hybrid-capable: it supports both WSGI (sync) and ASGI (async) Django applications. The middleware
detects at initialization whether the next middleware in the chain is async or sync, and adapts its behavior accordingly.
This ensures compatibility with both pure sync and pure async middleware chains, as well as mixed chains in ASGI mode.
"""
# Django middleware capability flags
sync_capable = True
async_capable = True
def __init__(self, get_response):
# type: (Union[Callable[[HttpRequest], HttpResponse], Callable[[HttpRequest], Awaitable[HttpResponse]]]) -> None
self.get_response = get_response
self._is_coroutine = iscoroutinefunction(get_response)
self._async_get_response = None # type: Optional[Callable[[HttpRequest], Awaitable[HttpResponse]]]
self._sync_get_response = None # type: Optional[Callable[[HttpRequest], HttpResponse]]
# Mark this instance as a coroutine function if get_response is async
# This is required for Django to correctly detect async middleware
if self._is_coroutine:
self._async_get_response = cast(
"Callable[[HttpRequest], Awaitable[HttpResponse]]", get_response
)
else:
self._sync_get_response = cast(
"Callable[[HttpRequest], HttpResponse]", get_response
)
markcoroutinefunction(self)
from django.conf import settings
@@ -181,40 +185,67 @@ class PosthogContextMiddleware:
return user_id, email
def __call__(self, request):
# type: (HttpRequest) -> HttpResponse
# Purely defensive around django's internal sync/async handling - this should be unreachable, but if it's reached, we may
# as well return something semi-meaningful
# type: (HttpRequest) -> Union[HttpResponse, Awaitable[HttpResponse]]
"""
Unified entry point for both sync and async request handling.
When sync_capable and async_capable are both True, Django passes requests
without conversion. This method detects the mode and routes accordingly.
"""
if self._is_coroutine:
raise RuntimeError(
"PosthogContextMiddleware received sync call but get_response is async"
)
return self.__acall__(request)
else:
# Synchronous path
if self.request_filter and not self.request_filter(request):
return self.get_response(request)
if self.request_filter and not self.request_filter(request):
assert self._sync_get_response is not None
return self._sync_get_response(request)
with contexts.new_context(self.capture_exceptions, client=self.client):
for k, v in self.extract_tags(request).items():
contexts.tag(k, v)
with contexts.new_context(self.capture_exceptions, client=self.client):
for k, v in self.extract_tags(request).items():
contexts.tag(k, v)
assert self._sync_get_response is not None
return self._sync_get_response(request)
return self.get_response(request)
async def __acall__(self, request):
# type: (HttpRequest) -> HttpResponse
# type: (HttpRequest) -> Awaitable[HttpResponse]
"""
Asynchronous entry point for async request handling.
This method is called when the middleware chain is async.
"""
if self.request_filter and not self.request_filter(request):
if self._async_get_response is not None:
return await self._async_get_response(request)
else:
assert self._sync_get_response is not None
return self._sync_get_response(request)
return await self.get_response(request)
with contexts.new_context(self.capture_exceptions, client=self.client):
for k, v in self.extract_tags(request).items():
contexts.tag(k, v)
if self._async_get_response is not None:
return await self._async_get_response(request)
else:
assert self._sync_get_response is not None
return self._sync_get_response(request)
return await self.get_response(request)
def process_exception(self, request, exception):
# type: (HttpRequest, Exception) -> None
"""
Process exceptions from views and downstream middleware.
Django calls this WHILE still inside the context created by __call__,
so request tags have already been extracted and set. This method just
needs to capture the exception directly.
Django converts view exceptions into responses before they propagate through
the middleware stack, so the context manager in __call__/__acall__ never sees them.
Note: Django's process_exception is always synchronous, even for async views.
"""
if self.request_filter and not self.request_filter(request):
return
if not self.capture_exceptions:
return
# Context and tags already set by __call__ or __acall__
# Just capture the exception
if self.client:
self.client.capture_exception(exception)
else:
from posthog import capture_exception
capture_exception(exception)
@@ -204,6 +204,7 @@ def test_basic_chat_chain(mock_client, stream):
# Generation is second
assert generation_args["event"] == "$ai_generation"
assert "distinct_id" in generation_args
assert generation_props["$ai_framework"] == "langchain"
assert "$ai_model" in generation_props
assert "$ai_provider" in generation_props
assert generation_props["$ai_input"] == [
+268 -102
View File
@@ -1,5 +1,5 @@
import time
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
@@ -34,6 +34,7 @@ try:
)
from posthog.ai.openai import OpenAI
from posthog.ai.openai.openai_async import AsyncOpenAI
OPENAI_AVAILABLE = True
except ImportError:
@@ -218,6 +219,106 @@ def mock_openai_response_with_cached_tokens():
)
@pytest.fixture
def streaming_tool_call_chunks():
return [
ChatCompletionChunk(
id="chunk1",
model="gpt-4",
object="chat.completion.chunk",
created=1234567890,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
role="assistant",
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
name="get_weather",
arguments='{"location": "',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk2",
model="gpt-4",
object="chat.completion.chunk",
created=1234567891,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
arguments='San Francisco"',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk3",
model="gpt-4",
object="chat.completion.chunk",
created=1234567892,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
arguments=', "unit": "celsius"}',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk4",
model="gpt-4",
object="chat.completion.chunk",
created=1234567893,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
content="The weather in San Francisco is 15°C.",
),
finish_reason=None,
)
],
usage=CompletionUsage(
prompt_tokens=20,
completion_tokens=15,
total_tokens=35,
),
),
]
@pytest.fixture
def mock_openai_response_with_tool_calls():
return ChatCompletion(
@@ -734,109 +835,11 @@ def test_responses_api_tool_calls(mock_client, mock_responses_api_with_tool_call
assert props["$ai_http_status"] == 200
def test_streaming_with_tool_calls(mock_client):
# Create mock tool call chunks that will be returned in sequence
tool_call_chunks = [
ChatCompletionChunk(
id="chunk1",
model="gpt-4",
object="chat.completion.chunk",
created=1234567890,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
role="assistant",
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
name="get_weather",
arguments='{"location": "',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk2",
model="gpt-4",
object="chat.completion.chunk",
created=1234567891,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
arguments='San Francisco"',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk3",
model="gpt-4",
object="chat.completion.chunk",
created=1234567892,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
tool_calls=[
ChoiceDeltaToolCall(
index=0,
id="call_abc123",
type="function",
function=ChoiceDeltaToolCallFunction(
arguments=', "unit": "celsius"}',
),
)
],
),
finish_reason=None,
)
],
),
ChatCompletionChunk(
id="chunk4",
model="gpt-4",
object="chat.completion.chunk",
created=1234567893,
choices=[
ChoiceChunk(
index=0,
delta=ChoiceDelta(
content="The weather in San Francisco is 15°C.",
),
finish_reason=None,
)
],
usage=CompletionUsage(
prompt_tokens=20,
completion_tokens=15,
total_tokens=35,
),
),
]
def test_streaming_with_tool_calls(mock_client, streaming_tool_call_chunks):
# Mock the create method to return our chunks
with patch("openai.resources.chat.completions.Completions.create") as mock_create:
# Set up the mock to return our chunks when iterated
mock_create.return_value = tool_call_chunks
mock_create.return_value = streaming_tool_call_chunks
client = OpenAI(api_key="test-key", posthog_client=mock_client)
@@ -865,7 +868,7 @@ def test_streaming_with_tool_calls(mock_client):
# Verify the chunks were returned correctly
assert len(chunks) == 4
assert chunks == tool_call_chunks
assert chunks == streaming_tool_call_chunks
# Verify the capture was called with the right arguments
assert mock_client.capture.call_count == 1
@@ -1112,6 +1115,169 @@ def test_responses_api_streaming_with_tokens(mock_client):
assert isinstance(props["$ai_latency"], float)
@pytest.mark.asyncio
async def test_async_chat_streaming_with_tool_calls(
mock_client, streaming_tool_call_chunks
):
captured_kwargs = {}
async def mock_create(self, **kwargs):
captured_kwargs["kwargs"] = kwargs
async def chunk_iterable():
for chunk in streaming_tool_call_chunks:
yield chunk
return chunk_iterable()
with patch(
"openai.resources.chat.completions.AsyncCompletions.create", new=mock_create
):
client = AsyncOpenAI(api_key="test-key", posthog_client=mock_client)
response_stream = await client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "user", "content": "What's the weather in San Francisco?"}
],
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather",
"parameters": {},
},
}
],
stream=True,
posthog_distinct_id="test-id",
)
chunks = []
async for chunk in response_stream:
chunks.append(chunk)
kwargs = captured_kwargs["kwargs"]
assert kwargs["stream_options"]["include_usage"] is True
assert len(chunks) == len(streaming_tool_call_chunks)
assert chunks == streaming_tool_call_chunks
assert mock_client.capture.call_count == 1
call_args = mock_client.capture.call_args[1]
props = call_args["properties"]
assert call_args["distinct_id"] == "test-id"
assert call_args["event"] == "$ai_generation"
assert props["$ai_provider"] == "openai"
assert props["$ai_model"] == "gpt-4"
assert props["$ai_output_tokens"] == 15
assert props["$ai_input_tokens"] == 20
assert isinstance(props["$ai_latency"], float)
@pytest.mark.asyncio
async def test_async_responses_streaming_with_tokens(mock_client):
from openai.types.responses import ResponseUsage
from unittest.mock import MagicMock
chunks = []
chunk1 = MagicMock()
chunk1.type = "response.text.delta"
chunk1.text = "Test "
chunks.append(chunk1)
chunk2 = MagicMock()
chunk2.type = "response.text.delta"
chunk2.text = "response"
chunks.append(chunk2)
chunk3 = MagicMock()
chunk3.type = "response.completed"
chunk3.response = MagicMock()
chunk3.response.usage = ResponseUsage(
input_tokens=25,
output_tokens=30,
total_tokens=55,
input_tokens_details={"prompt_tokens": 25, "cached_tokens": 0},
output_tokens_details={"reasoning_tokens": 0},
)
chunk3.response.output = ["Test response"]
chunks.append(chunk3)
captured_kwargs = {}
async def mock_create(self, **kwargs):
captured_kwargs["kwargs"] = kwargs
async def chunk_iterable():
for chunk in chunks:
yield chunk
return chunk_iterable()
with patch("openai.resources.responses.AsyncResponses.create", new=mock_create):
client = AsyncOpenAI(api_key="test-key", posthog_client=mock_client)
response_stream = await client.responses.create(
model="gpt-4o-mini",
input=[{"role": "user", "content": "Test message"}],
stream=True,
posthog_distinct_id="test-id",
posthog_properties={"test": "streaming"},
)
async for _ in response_stream:
pass
kwargs = captured_kwargs["kwargs"]
assert "stream_options" not in kwargs
assert mock_client.capture.call_count == 1
call_args = mock_client.capture.call_args[1]
props = call_args["properties"]
assert call_args["distinct_id"] == "test-id"
assert call_args["event"] == "$ai_generation"
assert props["$ai_provider"] == "openai"
assert props["$ai_model"] == "gpt-4o-mini"
assert props["$ai_input_tokens"] == 25
assert props["$ai_output_tokens"] == 30
assert props["test"] == "streaming"
assert isinstance(props["$ai_latency"], float)
@pytest.mark.asyncio
async def test_async_embeddings_create(mock_client, mock_embedding_response):
mock_create = AsyncMock(return_value=mock_embedding_response)
with patch("openai.resources.embeddings.AsyncEmbeddings.create", new=mock_create):
client = AsyncOpenAI(api_key="test-key", posthog_client=mock_client)
response = await client.embeddings.create(
model="text-embedding-3-small",
input="Hello world",
posthog_distinct_id="test-id",
posthog_properties={"foo": "bar"},
)
assert response == mock_embedding_response
assert mock_create.await_count == 1
assert mock_client.capture.call_count == 1
call_args = mock_client.capture.call_args[1]
props = call_args["properties"]
assert call_args["distinct_id"] == "test-id"
assert call_args["event"] == "$ai_embedding"
assert props["$ai_provider"] == "openai"
assert props["$ai_model"] == "text-embedding-3-small"
assert props["foo"] == "bar"
assert isinstance(props["$ai_latency"], float)
def test_tool_definition(mock_client, mock_openai_response):
"""Test that tools defined in the create function are captured in $ai_tools property"""
with patch(
+383 -8
View File
@@ -4,7 +4,21 @@ from posthog.contexts import (
get_context_distinct_id,
)
import unittest
from unittest.mock import Mock
from unittest.mock import Mock, patch
import asyncio
# Configure Django settings before importing middleware
import django
from django.conf import settings
if not settings.configured:
settings.configure(
DEBUG=True,
SECRET_KEY="test-secret-key",
INSTALLED_APPS=[],
MIDDLEWARE=[],
)
django.setup()
from posthog.integrations.django import PosthogContextMiddleware
@@ -38,14 +52,33 @@ class TestPosthogContextMiddleware(unittest.TestCase):
request_filter=None,
tag_map=None,
capture_exceptions=True,
get_response=None,
):
"""Helper to create middleware instance without calling __init__"""
middleware = PosthogContextMiddleware.__new__(PosthogContextMiddleware)
middleware.get_response = Mock()
middleware.extra_tags = extra_tags
middleware.request_filter = request_filter
middleware.tag_map = tag_map
middleware.capture_exceptions = capture_exceptions
"""Helper to create middleware instance with mock Django settings"""
if get_response is None:
get_response = Mock()
with patch("django.conf.settings") as mock_settings:
# Configure mock settings
mock_settings.POSTHOG_MW_EXTRA_TAGS = extra_tags
mock_settings.POSTHOG_MW_REQUEST_FILTER = request_filter
mock_settings.POSTHOG_MW_TAG_MAP = tag_map
mock_settings.POSTHOG_MW_CAPTURE_EXCEPTIONS = capture_exceptions
mock_settings.POSTHOG_MW_CLIENT = None
# Make hasattr work correctly
def mock_hasattr(obj, name):
return name in [
"POSTHOG_MW_EXTRA_TAGS",
"POSTHOG_MW_REQUEST_FILTER",
"POSTHOG_MW_TAG_MAP",
"POSTHOG_MW_CAPTURE_EXCEPTIONS",
"POSTHOG_MW_CLIENT",
]
with patch("builtins.hasattr", side_effect=mock_hasattr):
middleware = PosthogContextMiddleware(get_response)
return middleware
def test_extract_tags_basic(self):
@@ -168,6 +201,348 @@ class TestPosthogContextMiddleware(unittest.TestCase):
self.assertEqual(tags["$request_method"], "PATCH")
def test_process_exception_called_during_view_exception(self):
"""
Unit test verifying process_exception captures exceptions per Django's contract.
Since this is a library test (no Django runtime), we simulate how Django
would invoke our middleware in production:
1. Middleware.__call__ creates context with request tags
2. View raises exception inside get_response
3. Django's BaseHandler catches it, calls process_exception, returns error response
4. Exception never propagates to middleware's context manager
We manually call process_exception to simulate Django's behavior - this is
the only way to test the hook without a full Django integration test.
"""
mock_client = Mock()
view_exception = ValueError("View raised this error")
error_response = Mock(status_code=500)
def mock_get_response(request):
# Simulate Django's exception handling: catches view exception,
# calls process_exception hook if it exists, returns error response
if hasattr(middleware, "process_exception"):
middleware.process_exception(request, view_exception)
return error_response
middleware = self.create_middleware(get_response=mock_get_response)
middleware.client = mock_client
request = MockRequest(
headers={"X-POSTHOG-DISTINCT-ID": "test-user"},
method="POST",
path="/api/endpoint",
)
response = middleware(request)
self.assertEqual(response.status_code, 500)
mock_client.capture_exception.assert_called_once_with(view_exception)
def test_process_exception_respects_capture_exceptions_false(self):
"""Verify process_exception respects capture_exceptions=False setting"""
mock_client = Mock()
view_exception = ValueError("Should not be captured")
def mock_get_response(request):
if hasattr(middleware, "process_exception"):
middleware.process_exception(request, view_exception)
return Mock(status_code=500)
middleware = self.create_middleware(
capture_exceptions=False, get_response=mock_get_response
)
middleware.client = mock_client
request = MockRequest()
middleware(request)
mock_client.capture_exception.assert_not_called()
def test_process_exception_respects_request_filter(self):
"""Verify process_exception respects request_filter setting"""
mock_client = Mock()
view_exception = ValueError("Should be filtered")
def mock_get_response(request):
if hasattr(middleware, "process_exception"):
middleware.process_exception(request, view_exception)
return Mock(status_code=500)
middleware = self.create_middleware(
request_filter=lambda req: False,
capture_exceptions=True,
get_response=mock_get_response,
)
middleware.client = mock_client
request = MockRequest()
middleware(request)
mock_client.capture_exception.assert_not_called()
class TestPosthogContextMiddlewareSync(unittest.TestCase):
"""Test synchronous middleware behavior"""
def test_sync_middleware_call(self):
"""Test that sync middleware correctly processes requests"""
mock_response = Mock()
get_response = Mock(return_value=mock_response)
# Create middleware with sync get_response
middleware = PosthogContextMiddleware(get_response)
# Verify sync mode detected
self.assertFalse(middleware._is_coroutine)
request = MockRequest(
headers={"X-POSTHOG-SESSION-ID": "test-session"},
method="GET",
path="/test",
)
with new_context():
response = middleware(request)
# Verify response returned
self.assertEqual(response, mock_response)
get_response.assert_called_once_with(request)
def test_sync_middleware_with_filter(self):
"""Test sync middleware respects request filter"""
mock_response = Mock()
get_response = Mock(return_value=mock_response)
# Create middleware with request filter that filters all requests
request_filter = lambda req: False
middleware = PosthogContextMiddleware.__new__(PosthogContextMiddleware)
middleware.get_response = get_response
middleware._is_coroutine = False
middleware.request_filter = request_filter
middleware.capture_exceptions = True
middleware.client = None
request = MockRequest()
# Should skip context creation and return response directly
response = middleware(request)
self.assertEqual(response, mock_response)
get_response.assert_called_once_with(request)
def test_view_exceptions_only_captured_via_process_exception(self):
"""
Demonstrates that process_exception is required to capture view exceptions.
In production Django, view exceptions don't propagate to middleware's context
manager because Django's BaseHandler catches them first and converts them to
error responses. Django provides the exception via process_exception hook instead.
This unit test proves:
1. Context manager in __call__ never sees view exceptions (Django intercepts)
2. Only process_exception can capture them
3. Without process_exception, exceptions are silently lost (v6.7.5 regression)
We manually call process_exception to verify the hook works - in production,
Django's BaseHandler would call it when a view raises.
"""
mock_client = Mock()
get_response = Mock(return_value=Mock(status_code=500))
middleware = PosthogContextMiddleware(get_response)
middleware.client = mock_client
def get_response_simulating_django(request):
# Simulates Django behavior: view exception converted to error response,
# never propagates to middleware's context manager
return Mock(status_code=500)
middleware._sync_get_response = get_response_simulating_django
request = MockRequest()
response = middleware(request)
self.assertEqual(response.status_code, 500)
# Context manager didn't capture anything - exception was intercepted by Django
mock_client.capture_exception.assert_not_called()
# Verify process_exception hook exists and captures exceptions when called
if hasattr(middleware, "process_exception"):
exception = ValueError("View error")
middleware.process_exception(request, exception)
mock_client.capture_exception.assert_called_once_with(exception)
else:
self.fail(
"process_exception missing - view exceptions will not be captured!"
)
class TestPosthogContextMiddlewareAsync(unittest.TestCase):
"""Test asynchronous middleware behavior"""
def test_async_middleware_detection(self):
"""Test that async get_response is correctly detected"""
async def async_get_response(request):
return Mock()
middleware = PosthogContextMiddleware(async_get_response)
# Verify async mode detected
self.assertTrue(middleware._is_coroutine)
def test_async_middleware_call(self):
"""Test that async middleware correctly processes requests"""
async def run_test():
mock_response = Mock()
async def async_get_response(request):
return mock_response
middleware = PosthogContextMiddleware(async_get_response)
request = MockRequest(
headers={"X-POSTHOG-SESSION-ID": "async-session"},
method="POST",
path="/async-test",
)
with new_context():
# Call should return the coroutine from __acall__
result = middleware(request)
# Verify it's a coroutine
self.assertTrue(asyncio.iscoroutine(result))
# Await the result
response = await result
self.assertEqual(response, mock_response)
asyncio.run(run_test())
def test_async_middleware_with_filter(self):
"""Test async middleware respects request filter"""
async def run_test():
mock_response = Mock()
async def async_get_response(request):
return mock_response
# Properly initialize middleware
middleware = PosthogContextMiddleware(async_get_response)
# Override request filter after initialization
middleware.request_filter = lambda req: False
request = MockRequest()
# Should skip context creation and return response directly
result = middleware(request)
response = await result
self.assertEqual(response, mock_response)
asyncio.run(run_test())
def test_async_middleware_context_propagation(self):
"""Test that async middleware properly propagates context"""
async def run_test():
mock_response = Mock()
async def async_get_response(request):
# Verify context is available during async processing
session_id = get_context_session_id()
self.assertEqual(session_id, "async-session-123")
return mock_response
middleware = PosthogContextMiddleware(async_get_response)
request = MockRequest(
headers={"X-POSTHOG-SESSION-ID": "async-session-123"},
method="GET",
)
with new_context():
result = middleware(request)
await result
asyncio.run(run_test())
def test_async_middleware_exception_capture(self):
"""Test that async middleware captures exceptions during request processing"""
async def run_test():
mock_client = Mock()
# Make async_get_response raise an exception
async def raise_exception(request):
raise ValueError("Async test exception")
# Properly initialize middleware
middleware = PosthogContextMiddleware(raise_exception)
middleware.client = mock_client # Override with mock client
request = MockRequest()
# Should capture exception and re-raise
with self.assertRaises(ValueError):
result = middleware(request)
await result
# Verify exception was captured by middleware
mock_client.capture_exception.assert_called_once()
captured_exception = mock_client.capture_exception.call_args[0][0]
self.assertIsInstance(captured_exception, ValueError)
self.assertEqual(str(captured_exception), "Async test exception")
asyncio.run(run_test())
class TestPosthogContextMiddlewareHybrid(unittest.TestCase):
"""Test hybrid middleware behavior with mixed sync/async chains"""
def test_hybrid_flags_set(self):
"""Test that both capability flags are set"""
self.assertTrue(PosthogContextMiddleware.sync_capable)
self.assertTrue(PosthogContextMiddleware.async_capable)
def test_sync_to_async_routing(self):
"""Test that __call__ routes to __acall__ when async"""
async def run_test():
async def async_get_response(request):
return Mock()
middleware = PosthogContextMiddleware(async_get_response)
# Verify routing happens
request = MockRequest()
result = middleware(request)
# Should be a coroutine from __acall__
self.assertTrue(asyncio.iscoroutine(result))
await result # Clean up
asyncio.run(run_test())
def test_sync_path_direct_return(self):
"""Test that sync path returns directly without coroutine"""
mock_response = Mock()
def sync_get_response(request):
return mock_response
middleware = PosthogContextMiddleware(sync_get_response)
request = MockRequest()
result = middleware(request)
# Should NOT be a coroutine
self.assertFalse(asyncio.iscoroutine(result))
self.assertEqual(result, mock_response)
if __name__ == "__main__":
unittest.main()
+123
View File
@@ -3013,6 +3013,75 @@ class TestLocalEvaluation(unittest.TestCase):
)
self.assertEqual(patch_flags.call_count, 0)
@mock.patch("posthog.client.flags")
@mock.patch("posthog.client.get")
def test_fallback_to_api_when_flag_has_static_cohort_in_multi_condition(
self, patch_get, patch_flags
):
"""
When a flag has multiple conditions and one contains a static cohort,
the SDK should fallback to API for the entire flag, not just skip that
condition and evaluate the next one locally.
This prevents returning wrong variants when later conditions could match
locally but the user is actually in the static cohort.
"""
client = Client(FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
# Mock the local flags response - cohort 999 is NOT in cohorts map (static cohort)
client.feature_flags = [
{
"id": 1,
"key": "multi-condition-flag",
"active": True,
"filters": {
"groups": [
{
"properties": [
{"key": "id", "value": 999, "type": "cohort"}
],
"rollout_percentage": 100,
"variant": "set-1",
},
{
"properties": [
{
"key": "$geoip_country_code",
"operator": "exact",
"value": ["DE"],
"type": "person",
}
],
"rollout_percentage": 100,
"variant": "set-8",
},
],
"multivariate": {
"variants": [
{"key": "set-1", "rollout_percentage": 50},
{"key": "set-8", "rollout_percentage": 50},
]
},
},
}
]
client.cohorts = {} # Note: cohort 999 is NOT here - it's a static cohort
# Mock the API response - user is in the static cohort
patch_flags.return_value = {"featureFlags": {"multi-condition-flag": "set-1"}}
result = client.get_feature_flag(
"multi-condition-flag",
"test-distinct-id",
person_properties={"$geoip_country_code": "DE"},
)
# Should return the API result (set-1), not local evaluation (set-8)
self.assertEqual(result, "set-1")
# Verify API was called (fallback occurred)
self.assertEqual(patch_flags.call_count, 1)
class TestMatchProperties(unittest.TestCase):
def property(self, key, value, operator=None):
@@ -4006,6 +4075,60 @@ class TestCaptureCalls(unittest.TestCase):
patch_capture.reset_mock()
@mock.patch("posthog.client.flags")
def test_fallback_to_api_in_get_feature_flag_payload_when_flag_has_static_cohort(
self, patch_flags
):
"""
Test that get_feature_flag_payload falls back to API when evaluating
a flag with static cohorts, similar to get_feature_flag behavior.
"""
client = Client(FAKE_TEST_API_KEY, personal_api_key=FAKE_TEST_API_KEY)
# Mock the local flags response - cohort 999 is NOT in cohorts map (static cohort)
client.feature_flags = [
{
"id": 1,
"name": "Multi-condition Flag",
"key": "multi-condition-flag",
"active": True,
"filters": {
"groups": [
{
"properties": [
{"key": "id", "value": 999, "type": "cohort"}
],
"rollout_percentage": 100,
"variant": "variant-1",
}
],
"multivariate": {
"variants": [{"key": "variant-1", "rollout_percentage": 100}]
},
"payloads": {"variant-1": '{"message": "local-payload"}'},
},
}
]
client.cohorts = {} # Note: cohort 999 is NOT here - it's a static cohort
# Mock the API response - user is in the static cohort
patch_flags.return_value = {
"featureFlags": {"multi-condition-flag": "variant-1"},
"featureFlagPayloads": {"multi-condition-flag": '{"message": "from-api"}'},
}
# Call get_feature_flag_payload without match_value to trigger evaluation
result = client.get_feature_flag_payload(
"multi-condition-flag",
"test-distinct-id",
)
# Should return the API payload, not local payload
self.assertEqual(result, {"message": "from-api"})
# Verify API was called (fallback occurred)
self.assertEqual(patch_flags.call_count, 1)
@mock.patch.object(Client, "capture")
@mock.patch("posthog.client.flags")
def test_disable_geoip_get_flag_capture_call(self, patch_flags, patch_capture):
+1 -1
View File
@@ -1,4 +1,4 @@
VERSION = "6.7.7"
VERSION = "6.7.11"
if __name__ == "__main__":
print(VERSION, end="") # noqa: T201
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
+326
View File
@@ -0,0 +1,326 @@
#!/usr/bin/env python3
"""
Test script to send capture_ai events to localhost:8010.
This script tests the actual network request to a local PostHog instance.
"""
from posthog import Posthog
from uuid import uuid4
# Create a client pointing to localhost:8010
posthog = Posthog(
"test-api-key", # Use your actual project API key if needed
host="http://localhost:8010",
debug=True, # Enable debug mode to see detailed logs
)
print("Testing capture_ai with localhost:8010")
print("=" * 60)
# Test 1: $ai_generation event with blobs
print("\n1. Testing $ai_generation event with blobs...")
print("-" * 60)
trace_id = f"trace_{uuid4().hex[:8]}"
try:
event_uuid = posthog.capture_ai(
"$ai_generation",
distinct_id="test_user_123",
properties={
"$ai_model": "gpt-4",
"$ai_provider": "openai",
"$ai_trace_id": trace_id,
"$ai_input": {
"messages": [
{
"role": "system",
"content": "You are a helpful assistant that answers questions about Python.",
},
{
"role": "user",
"content": "What is the difference between a list and a tuple?",
},
],
"temperature": 0.7,
"max_tokens": 500,
},
"$ai_output_choices": {
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "A list is mutable (can be changed) while a tuple is immutable (cannot be changed after creation). Lists use square brackets [] and tuples use parentheses ().",
},
"finish_reason": "stop",
}
],
"model": "gpt-4",
"usage": {
"prompt_tokens": 45,
"completion_tokens": 32,
"total_tokens": 77,
},
},
"$ai_completion_tokens": 32,
"$ai_prompt_tokens": 45,
"$ai_total_tokens": 77,
"$ai_latency": 1.234,
},
blob_properties=["$ai_input", "$ai_output_choices"],
)
if event_uuid:
print("✓ SUCCESS: $ai_generation event sent")
print(f" UUID: {event_uuid}")
print(f" Trace ID: {trace_id}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 2: $ai_trace event
print("\n2. Testing $ai_trace event...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_trace",
distinct_id="test_user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": trace_id,
"$ai_trace_name": "python_qa_session",
"$ai_input_state": {
"session_id": "session_123",
"user_context": "learning Python",
},
"$ai_output_state": {"questions_answered": 1, "satisfaction_score": 5},
},
blob_properties=["$ai_input_state", "$ai_output_state"],
)
if event_uuid:
print("✓ SUCCESS: $ai_trace event sent")
print(f" UUID: {event_uuid}")
print(f" Trace ID: {trace_id}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 3: $ai_span event
print("\n3. Testing $ai_span event...")
print("-" * 60)
span_id = f"span_{uuid4().hex[:8]}"
try:
event_uuid = posthog.capture_ai(
"$ai_span",
distinct_id="test_user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": trace_id,
"$ai_span_id": span_id,
"$ai_span_name": "answer_generation",
"$ai_parent_id": trace_id,
"$ai_span_kind": "llm",
"$ai_latency": 0.8,
},
)
if event_uuid:
print("✓ SUCCESS: $ai_span event sent")
print(f" UUID: {event_uuid}")
print(f" Span ID: {span_id}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 4: $ai_embedding event
print("\n4. Testing $ai_embedding event...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_embedding",
distinct_id="test_user_123",
properties={
"$ai_model": "text-embedding-ada-002",
"$ai_provider": "openai",
"$ai_trace_id": trace_id,
"$ai_input": {
"text": "What is the difference between a list and a tuple in Python?"
},
"$ai_embedding_dimension": 1536,
"$ai_latency": 0.123,
},
blob_properties=["$ai_input"],
)
if event_uuid:
print("✓ SUCCESS: $ai_embedding event sent")
print(f" UUID: {event_uuid}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 5: $ai_metric event
print("\n5. Testing $ai_metric event...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_metric",
distinct_id="test_user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": trace_id,
"$ai_metric_name": "response_quality",
"$ai_metric_value": "0.95",
},
)
if event_uuid:
print("✓ SUCCESS: $ai_metric event sent")
print(f" UUID: {event_uuid}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 6: $ai_feedback event
print("\n6. Testing $ai_feedback event...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_feedback",
distinct_id="test_user_123",
properties={
"$ai_model": "gpt-4",
"$ai_trace_id": trace_id,
"$ai_feedback_text": "Great explanation! Very clear and helpful.",
"$ai_feedback_rating": 5,
},
)
if event_uuid:
print("✓ SUCCESS: $ai_feedback event sent")
print(f" UUID: {event_uuid}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 7: Test with custom blob properties
print("\n7. Testing with custom blob properties...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_generation",
distinct_id="test_user_123",
properties={
"$ai_model": "claude-3-opus",
"$ai_provider": "anthropic",
"$ai_trace_id": trace_id,
"$ai_input": {
"messages": [{"role": "user", "content": "Write a haiku about Python"}]
},
"$ai_output_choices": {
"choices": [
{
"message": {
"role": "assistant",
"content": "Snake glides through code\nSimple syntax, powerful tools\nDevelopers smile",
}
}
]
},
"$ai_custom_data": {
"large_context": "This is some large custom data that should be sent as a blob"
},
"$ai_completion_tokens": 20,
"$ai_prompt_tokens": 10,
},
# Custom blob properties - including the default ones plus a custom one
blob_properties=["$ai_input", "$ai_output_choices", "$ai_custom_data"],
)
if event_uuid:
print("✓ SUCCESS: Event with custom blob properties sent")
print(f" UUID: {event_uuid}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
# Test 8: Test with groups
print("\n8. Testing with groups...")
print("-" * 60)
try:
event_uuid = posthog.capture_ai(
"$ai_generation",
distinct_id="test_user_123",
groups={"company": "posthog_inc", "team": "engineering"},
properties={
"$ai_model": "gpt-4",
"$ai_provider": "openai",
"$ai_trace_id": trace_id,
"$ai_input": {"messages": [{"role": "user", "content": "test"}]},
"$ai_output_choices": {
"choices": [{"message": {"role": "assistant", "content": "response"}}]
},
},
)
if event_uuid:
print("✓ SUCCESS: Event with groups sent")
print(f" UUID: {event_uuid}")
else:
print("✗ FAILED: No UUID returned")
except Exception as e:
print(f"✗ ERROR: {e}")
import traceback
traceback.print_exc()
print("\n" + "=" * 60)
print("All tests completed!")
print("\nMake sure your local PostHog instance is running on http://localhost:8010")
print("and that the /i/v0/ai endpoint is available.")