import atexit

import logging

import os

import sys

import warnings

from datetime import datetime, timedelta

from typing import Any, Dict, Optional, Union

from uuid import uuid4

from dateutil.tz import tzutc

from six import string_types

from typing_extensions import Unpack

from posthog.args import ID_TYPES, ExceptionArg, OptionalCaptureArgs, OptionalSetArgs

from posthog.consumer import Consumer

from posthog.contexts import (

_get_current_context,

get_capture_exception_code_variables_context,

get_code_variables_ignore_patterns_context,

get_code_variables_mask_patterns_context,

get_context_device_id,

get_context_distinct_id,

get_context_session_id,

new_context,

)

from posthog.exception_capture import ExceptionCapture

from posthog.exception_utils import (

DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS,

DEFAULT_CODE_VARIABLES_MASK_PATTERNS,

exc_info_from_error,

exception_is_already_captured,

exceptions_from_error_tuple,

handle_in_app,

mark_exception_as_captured,

try_attach_code_variables_to_frames,

)

from posthog.feature_flags import (

InconclusiveMatchError,

RequiresServerEvaluation,

match_feature_flag_properties,

resolve_bucketing_value,

)

from posthog.flag_definition_cache import (

FlagDefinitionCacheData,

FlagDefinitionCacheProvider,

)

from posthog.poller import Poller

from posthog.request import (

DEFAULT_HOST,

APIError,

QuotaLimitError,

RequestsConnectionError,

RequestsTimeout,

batch_post,

determine_server_host,

flags,

get,

remote_config,

)

from posthog.types import (

FeatureFlag,

FeatureFlagError,

FeatureFlagResult,

FlagMetadata,

FlagsAndPayloads,

FlagsResponse,

FlagValue,

SendFeatureFlagsOptions,

normalize_flags_response,

to_flags_and_payloads,

to_payloads,

to_values,

)

from posthog.utils import (

FlagCache,

RedisFlagCache,

SizeLimitedDict,

clean,

guess_timezone,

system_context,

)

from posthog.version import VERSION

try:

import queue

except ImportError:

import Queue as queue

MAX_DICT_SIZE = 50_000

def get_identity_state(passed) -> tuple[str, bool]:

"""Returns the distinct id to use, and whether this is a personless event or not"""

stringified = stringify_id(passed)

if stringified and len(stringified):

return (stringified, False)

context_id = get_context_distinct_id()

if context_id:

return (context_id, False)

return (str(uuid4()), True)

def add_context_tags(properties):

properties = properties or {}

current_context = _get_current_context()

if current_context:

context_tags = current_context.collect_tags()

properties["$context_tags"] = set(context_tags.keys())

# We want explicitly passed properties to override context tags

context_tags.update(properties)

properties = context_tags

if "$session_id" not in properties and get_context_session_id():

properties["$session_id"] = get_context_session_id()

return properties

def no_throw(default_return=None):

"""

Decorator to prevent raising exceptions from public API methods.

Note that this doesn't prevent errors from propagating via `on_error`.

Exceptions will still be raised if the debug flag is enabled.

Args:

default_return: Value to return on exception (default: None)

"""

def decorator(func):

from functools import wraps

@wraps(func)

def wrapper(self, *args, **kwargs):

try:

return func(self, *args, **kwargs)

except Exception as e:

if self.debug:

raise e

self.log.exception(f"Error in {func.__name__}: {e}")

return default_return

return wrapper

return decorator

class Client(object):

"""

This is the SDK reference for the PostHog Python SDK.

You can learn more about example usage in the [Python SDK documentation](/docs/libraries/python).

You can also follow [Flask](/docs/libraries/flask) and [Django](/docs/libraries/django)

guides to integrate PostHog into your project.

Examples:

```python

from posthog import Posthog

posthog = Posthog('<ph_project_api_key>', host='<ph_client_api_host>')

posthog.debug = True

if settings.TEST:

posthog.disabled = True

```

"""

log = logging.getLogger("posthog")

def __init__(

self,

project_api_key: str,

host=None,

debug=False,

max_queue_size=10000,

send=True,

on_error=None,

flush_at=100,

flush_interval=0.5,

gzip=False,

max_retries=3,

sync_mode=False,

timeout=15,

thread=1,

poll_interval=30,

personal_api_key=None,

disabled=False,

disable_geoip=True,

historical_migration=False,

feature_flags_request_timeout_seconds=3,

super_properties=None,

enable_exception_autocapture=False,

log_captured_exceptions=False,

project_root=None,

privacy_mode=False,

before_send=None,

flag_fallback_cache_url=None,

enable_local_evaluation=True,

flag_definition_cache_provider: Optional[FlagDefinitionCacheProvider] = None,

capture_exception_code_variables=False,

code_variables_mask_patterns=None,

code_variables_ignore_patterns=None,

in_app_modules: list[str] | None = None,

):

"""

Initialize a new PostHog client instance.

Args:

project_api_key: The project API key.

host: The host to use for the client.

debug: Whether to enable debug mode.

Examples:

```python

from posthog import Posthog

posthog = Posthog('<ph_project_api_key>', host='<ph_app_host>')

```

Category:

Initialization

"""

self.queue = queue.Queue(max_queue_size)

# api_key: This should be the Team API Key (token), public

self.api_key = project_api_key

self.on_error = on_error

self.debug = debug

self.send = send

self.sync_mode = sync_mode

# Used for session replay URL generation - we don't want the server host here.

self.raw_host = host or DEFAULT_HOST

self.host = determine_server_host(host)

self.gzip = gzip

self.timeout = timeout

self._feature_flags = None # private variable to store flags

self.feature_flags_by_key = None

self.group_type_mapping: Optional[dict[str, str]] = None

self.cohorts: Optional[dict[str, Any]] = None

self.poll_interval = poll_interval

self.feature_flags_request_timeout_seconds = (

feature_flags_request_timeout_seconds

)

self.poller = None

self.distinct_ids_feature_flags_reported = SizeLimitedDict(MAX_DICT_SIZE, set)

self.flag_cache = self._initialize_flag_cache(flag_fallback_cache_url)

self.flag_definition_version = 0

self._flags_etag: Optional[str] = None

self._flag_definition_cache_provider = flag_definition_cache_provider

self.disabled = disabled

self.disable_geoip = disable_geoip

self.historical_migration = historical_migration

self.super_properties = super_properties

self.enable_exception_autocapture = enable_exception_autocapture

self.log_captured_exceptions = log_captured_exceptions

self.exception_capture = None

self.privacy_mode = privacy_mode

self.enable_local_evaluation = enable_local_evaluation

self.capture_exception_code_variables = capture_exception_code_variables

self.code_variables_mask_patterns = (

code_variables_mask_patterns

if code_variables_mask_patterns is not None

else DEFAULT_CODE_VARIABLES_MASK_PATTERNS

)

self.code_variables_ignore_patterns = (

code_variables_ignore_patterns

if code_variables_ignore_patterns is not None

else DEFAULT_CODE_VARIABLES_IGNORE_PATTERNS

)

self.in_app_modules = in_app_modules

if project_root is None:

try:

project_root = os.getcwd()

except Exception:

project_root = None

self.project_root = project_root

# personal_api_key: This should be a generated Personal API Key, private

self.personal_api_key = personal_api_key

if debug:

# Ensures that debug level messages are logged when debug mode is on.

# Otherwise, defaults to WARNING level. See https://docs.python.org/3/howto/logging.html#what-happens-if-no-configuration-is-provided

logging.basicConfig()

self.log.setLevel(logging.DEBUG)

else:

self.log.setLevel(logging.WARNING)

if before_send is not None:

if callable(before_send):

self.before_send = before_send

else:

self.log.warning("before_send is not callable, it will be ignored")

self.before_send = None

else:

self.before_send = None

if self.enable_exception_autocapture:

self.exception_capture = ExceptionCapture(self)

if sync_mode:

self.consumers = None

else:

# On program exit, allow the consumer thread to exit cleanly.

# This prevents exceptions and a messy shutdown when the

# interpreter is destroyed before the daemon thread finishes

# execution. However, it is *not* the same as flushing the queue!

# To guarantee all messages have been delivered, you'll still need

# to call flush().

if send:

atexit.register(self.join)

self.consumers = []

for _ in range(thread):

consumer = Consumer(

self.queue,

self.api_key,

host=self.host,

on_error=on_error,

flush_at=flush_at,

flush_interval=flush_interval,

gzip=gzip,

retries=max_retries,

timeout=timeout,

historical_migration=historical_migration,

)

self.consumers.append(consumer)

# if we've disabled sending, just don't start the consumer

if send:

consumer.start()

def new_context(self, fresh=False, capture_exceptions=True):

"""

Create a new context for managing shared state. Learn more about [contexts](/docs/libraries/python#contexts).

Args:

fresh: Whether to create a fresh context that doesn't inherit from parent.

capture_exceptions: Whether to automatically capture exceptions in this context.

Examples:

```python

with posthog.new_context():

identify_context('<distinct_id>')

posthog.capture('event_name')

```

Category:

Contexts

"""

return new_context(

fresh=fresh, capture_exceptions=capture_exceptions, client=self

)

@property

def feature_flags(self):

"""

Get the local evaluation feature flags.

"""

return self._feature_flags

@feature_flags.setter

def feature_flags(self, flags):

"""

Set the local evaluation feature flags.

"""

self._feature_flags = flags or []

self.feature_flags_by_key = {

flag["key"]: flag

for flag in self._feature_flags

if flag.get("key") is not None

}

assert self.feature_flags_by_key is not None, (

"feature_flags_by_key should be initialized when feature_flags is set"

)

def get_feature_variants(

self,

distinct_id,

groups=None,

person_properties=None,

group_properties=None,

disable_geoip=None,

flag_keys_to_evaluate: Optional[list[str]] = None,

device_id: Optional[str] = None,

) -> dict[str, Union[bool, str]]:

"""

Get feature flag variants for a user by calling decide.

Args:

distinct_id: The distinct ID of the user.

groups: A dictionary of group information.

person_properties: A dictionary of person properties.

group_properties: A dictionary of group properties.

disable_geoip: Whether to disable GeoIP for this request.

flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,

only these flags will be evaluated, improving performance.

device_id: The device ID for this request.

Category:

Feature flags

"""

resp_data = self.get_flags_decision(

distinct_id,

groups,

person_properties,

group_properties,

disable_geoip,

flag_keys_to_evaluate,

device_id=device_id,

)

return to_values(resp_data) or {}

def get_feature_payloads(

self,

distinct_id,

groups=None,

person_properties=None,

group_properties=None,

disable_geoip=None,

flag_keys_to_evaluate: Optional[list[str]] = None,

device_id: Optional[str] = None,

) -> dict[str, str]:

"""

Get feature flag payloads for a user by calling decide.

Args:

distinct_id: The distinct ID of the user.

groups: A dictionary of group information.

person_properties: A dictionary of person properties.

group_properties: A dictionary of group properties.

disable_geoip: Whether to disable GeoIP for this request.

flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,

only these flags will be evaluated, improving performance.

device_id: The device ID for this request.

Examples:

```python

payloads = posthog.get_feature_payloads('<distinct_id>')

```

Category:

Feature flags

"""

resp_data = self.get_flags_decision(

distinct_id,

groups,

person_properties,

group_properties,

disable_geoip,

flag_keys_to_evaluate,

device_id=device_id,

)

return to_payloads(resp_data) or {}

def get_feature_flags_and_payloads(

self,

distinct_id,

groups=None,

person_properties=None,

group_properties=None,

disable_geoip=None,

flag_keys_to_evaluate: Optional[list[str]] = None,

device_id: Optional[str] = None,

) -> FlagsAndPayloads:

"""

Get feature flags and payloads for a user by calling decide.

Args:

distinct_id: The distinct ID of the user.

groups: A dictionary of group information.

person_properties: A dictionary of person properties.

group_properties: A dictionary of group properties.

disable_geoip: Whether to disable GeoIP for this request.

flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,

only these flags will be evaluated, improving performance.

device_id: The device ID for this request.

Examples:

```python

result = posthog.get_feature_flags_and_payloads('<distinct_id>')

```

Category:

Feature flags

"""

resp = self.get_flags_decision(

distinct_id,

groups,

person_properties,

group_properties,

disable_geoip,

flag_keys_to_evaluate,

device_id=device_id,

)

return to_flags_and_payloads(resp)

def get_flags_decision(

self,

distinct_id: Optional[ID_TYPES] = None,

groups: Optional[dict] = None,

person_properties=None,

group_properties=None,

disable_geoip=None,

flag_keys_to_evaluate: Optional[list[str]] = None,

device_id: Optional[str] = None,

) -> FlagsResponse:

"""

Get feature flags decision.

Args:

distinct_id: The distinct ID of the user.

groups: A dictionary of group information.

person_properties: A dictionary of person properties.

group_properties: A dictionary of group properties.

disable_geoip: Whether to disable GeoIP for this request.

flag_keys_to_evaluate: A list of specific flag keys to evaluate. If provided,

only these flags will be evaluated, improving performance.

device_id: The device ID for this request.

Examples:

```python

decision = posthog.get_flags_decision('user123')

```

Category:

Feature flags

"""

groups = groups or {}

person_properties = person_properties or {}

group_properties = group_properties or {}

if distinct_id is None:

distinct_id = get_context_distinct_id()

if device_id is None:

device_id = get_context_device_id()

if disable_geoip is None:

disable_geoip = self.disable_geoip

if not groups:

groups = {}

request_data = {

"distinct_id": distinct_id,

"groups": groups,

"person_properties": person_properties,

"group_properties": group_properties,

"geoip_disable": disable_geoip,

"device_id": device_id,

}

if flag_keys_to_evaluate:

request_data["flag_keys_to_evaluate"] = flag_keys_to_evaluate

resp_data = flags(

self.api_key,

self.host,

timeout=self.feature_flags_request_timeout_seconds,

**request_data,

)

return normalize_flags_response(resp_data)

@no_throw()

def capture(

self, event: str, **kwargs: Unpack[OptionalCaptureArgs]

) -> Optional[str]:

"""

Captures an event manually. [Learn about capture best practices](https://posthog.com/docs/product-analytics/capture-events)

Args:

event: The event name to capture.

distinct_id: The distinct ID of the user.

properties: A dictionary of properties to include with the event.

timestamp: The timestamp of the event.

uuid: A unique identifier for the event.

groups: A dictionary of group information.

send_feature_flags: Whether to send feature flags with the event.

disable_geoip: Whether to disable GeoIP for this event.

Examples:

```python

# Anonymous event

posthog.capture('some-anon-event')

```

```python

# Context usage

from posthog import identify_context, new_context

with new_context():

identify_context('distinct_id_of_the_user')

posthog.capture('user_signed_up')

posthog.capture('user_logged_in')

posthog.capture('some-custom-action', distinct_id='distinct_id_of_the_user')

```

```python

# Set event properties

posthog.capture(

"user_signed_up",

distinct_id="distinct_id_of_the_user",

properties={

"login_type": "email",

"is_free_trial": "true"

}

)

```

```python

# Page view event

posthog.capture('$pageview', distinct_id="distinct_id_of_the_user", properties={'$current_url': 'https://example.com'})

```

Category:

Capture

"""

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)

send_feature_flags = kwargs.get("send_feature_flags", False)

disable_geoip = kwargs.get("disable_geoip", None)

properties = {**(properties or {}), **system_context()}

properties = add_context_tags(properties)

assert properties is not None # Type hint for mypy

(distinct_id, personless) = get_identity_state(distinct_id)

if personless and "$process_person_profile" not in properties:

properties["$process_person_profile"] = False

msg = {

"properties": properties,

"timestamp": timestamp,

"distinct_id": distinct_id,

"event": event,

"uuid": uuid,

}

if groups:

properties["$groups"] = groups

extra_properties: dict[str, Any] = {}

feature_variants: Optional[dict[str, Union[bool, str]]] = {}

# Parse and normalize send_feature_flags parameter

flag_options = self._parse_send_feature_flags(send_feature_flags)

if flag_options["should_send"]:

try:

if flag_options["only_evaluate_locally"] is True:

# Local evaluation explicitly requested

feature_variants = self.get_all_flags(

distinct_id,

groups=(groups or {}),

person_properties=flag_options["person_properties"],

group_properties=flag_options["group_properties"],

disable_geoip=disable_geoip,

only_evaluate_locally=True,

flag_keys_to_evaluate=flag_options["flag_keys_filter"],

)

elif flag_options["only_evaluate_locally"] is False:

# Remote evaluation explicitly requested

feature_variants = self.get_feature_variants(

distinct_id,

groups,

person_properties=flag_options["person_properties"],

group_properties=flag_options["group_properties"],

disable_geoip=disable_geoip,

flag_keys_to_evaluate=flag_options["flag_keys_filter"],

)

elif self.feature_flags:

# Local flags available, prefer local evaluation

feature_variants = self.get_all_flags(

distinct_id,

groups=(groups or {}),

person_properties=flag_options["person_properties"],

group_properties=flag_options["group_properties"],

disable_geoip=disable_geoip,

only_evaluate_locally=True,

flag_keys_to_evaluate=flag_options["flag_keys_filter"],

)

else:

# Fall back to remote evaluation

feature_variants = self.get_feature_variants(

distinct_id,

groups,

person_properties=flag_options["person_properties"],

group_properties=flag_options["group_properties"],

disable_geoip=disable_geoip,

flag_keys_to_evaluate=flag_options["flag_keys_filter"],

)

except Exception as e:

self.log.exception(

f"[FEATURE FLAGS] Unable to get feature variants: {e}"

)

for feature, variant in (feature_variants or {}).items():

extra_properties[f"$feature/{feature}"] = variant

active_feature_flags = [

key

for (key, value) in (feature_variants or {}).items()

if value is not False

]

if active_feature_flags:

extra_properties["$active_feature_flags"] = active_feature_flags

if extra_properties:

properties = {**extra_properties, **properties}

msg["properties"] = properties

return self._enqueue(msg, disable_geoip)

def _parse_send_feature_flags(self, send_feature_flags) -> SendFeatureFlagsOptions:

"""

Parse and normalize send_feature_flags parameter into a standard format.

Args:

send_feature_flags: Either bool or SendFeatureFlagsOptions dict

Returns:

SendFeatureFlagsOptions: Normalized options with keys: should_send, only_evaluate_locally,

person_properties, group_properties, flag_keys_filter

Raises:

TypeError: If send_feature_flags is not bool or dict

"""

if isinstance(send_feature_flags, dict):

return {

"should_send": True,

"only_evaluate_locally": send_feature_flags.get(

"only_evaluate_locally"

),

"person_properties": send_feature_flags.get("person_properties"),

"group_properties": send_feature_flags.get("group_properties"),

"flag_keys_filter": send_feature_flags.get("flag_keys_filter"),

}

elif isinstance(send_feature_flags, bool):

return {

"should_send": send_feature_flags,

"only_evaluate_locally": None,

"person_properties": None,

"group_properties": None,

"flag_keys_filter": None,

}

else:

raise TypeError(

f"Invalid type for send_feature_flags: {type(send_feature_flags)}. "

f"Expected bool or dict."

)

@no_throw()

def set(self, **kwargs: Unpack[OptionalSetArgs]) -> Optional[str]:

"""

Set properties on a person profile.

Args:

distinct_id: The distinct ID of the user.

properties: A dictionary of properties to set.

timestamp: The timestamp of the event.

uuid: A unique identifier for the event.

disable_geoip: Whether to disable GeoIP for this event.

Examples:

```python

# Set with distinct id

posthog.set(distinct_id='user123', properties={'name': 'Max Hedgehog'})

```

Category:

Identification

Note: This method will not raise exceptions. Errors are logged.

"""

distinct_id = kwargs.get("distinct_id", None)

properties = kwargs.get("properties", None)

timestamp = kwargs.get("timestamp", None)

uuid = kwargs.get("uuid", None)

disable_geoip = kwargs.get("disable_geoip", None)

properties = properties or {}

properties = add_context_tags(properties)

(distinct_id, personless) = get_identity_state(distinct_id)

if personless or not properties:

return None # Personless set() does nothing

msg = {

"timestamp": timestamp,

"distinct_id": distinct_id,

"$set": properties,

"event": "$set",

"uuid": uuid,

}

return self._enqueue(msg, disable_geoip)

@no_throw()

def set_once(self, **kwargs: Unpack[OptionalSetArgs]) -> Optional[str]:

"""

Set properties on a person profile only if they haven't been set before.

Args:

distinct_id: The distinct ID of the user.

properties: A dictionary of properties to set once.

timestamp: The timestamp of the event.

uuid: A unique identifier for the event.

disable_geoip: Whether to disable GeoIP for this event.

Examples:

```python

posthog.set_once(distinct_id='user123', properties={'initial_signup_date': '2024-01-01'})

```

Category:

Identification

Note: This method will not raise exceptions. Errors are logged.

"""

distinct_id = kwargs.get("distinct_id", None)

properties = kwargs.get("properties", None)

timestamp = kwargs.get("timestamp", None)

uuid = kwargs.get("uuid", None)

disable_geoip = kwargs.get("disable_geoip", None)

properties = properties or {}

properties = add_context_tags(properties)

(distinct_id, personless) = get_identity_state(distinct_id)

if personless or not properties:

return None # Personless set_once() does nothing

msg = {

"timestamp": timestamp,

"distinct_id": distinct_id,

"$set_once": properties,

"event": "$set_once",

"uuid": uuid,

}

return self._enqueue(msg, disable_geoip)

@no_throw()

def group_identify(

self,

group_type: str,

group_key: str,

properties: Optional[Dict[str, Any]] = None,

timestamp: Optional[Union[datetime, str]] = None,

uuid: Optional[str] = None,

disable_geoip: Optional[bool] = None,

distinct_id: Optional[ID_TYPES] = None,

) -> Optional[str]:

"""

Identify a group and set its properties.

Args:

group_type: The type of group (e.g., 'company', 'team').

group_key: The unique identifier for the group.

properties: A dictionary of properties to set on the group.

timestamp: The timestamp of the event.

uuid: A unique identifier for the event.

disable_geoip: Whether to disable GeoIP for this event.

distinct_id: The distinct ID of the user performing the action.

Examples:

```python

posthog.group_identify('company', 'company_id_in_your_db', {

'name': 'Awesome Inc.',

'employees': 11

})

```

Category:

Identification

Note: This method will not raise exceptions. Errors are logged.

"""

properties = properties or {}

# group_identify is purposefully always personful

distinct_id = get_identity_state(distinct_id)[0]

msg: Dict[str, Any] = {

"event": "$groupidentify",

"properties": {

"$group_type": group_type,

"$group_key": group_key,

"$group_set": properties,

},

"distinct_id": distinct_id,

"timestamp": timestamp,

"uuid": uuid,

}

# NOTE - group_identify doesn't generally use context properties - should it?

if get_context_session_id():

msg["properties"]["$session_id"] = str(get_context_session_id())

return self._enqueue(msg, disable_geoip)

@no_throw()

def alias(

self,

previous_id: str,

distinct_id: Optional[str],

timestamp=None,

uuid=None,

disable_geoip=None,

):

"""

Create an alias between two distinct IDs.

Args:

previous_id: The previous distinct ID.

distinct_id: The new distinct ID to alias to.

timestamp: The timestamp of the event.

uuid: A unique identifier for the event.

disable_geoip: Whether to disable GeoIP for this event.

Examples:

```python

posthog.alias(previous_id='distinct_id', distinct_id='alias_id')

```

Category:

Identification

Note: This method will not raise exceptions. Errors are logged.

"""

(distinct_id, personless) = get_identity_state(distinct_id)

if personless:

return None # Personless alias() does nothing - should this throw?

msg = {

"properties": {

"distinct_id": previous_id,

"alias": distinct_id,

},

"timestamp": timestamp,

"event": "$create_alias",

"distinct_id": previous_id,

"uuid": uuid,

}

if get_context_session_id():

msg["properties"]["$session_id"] = str(get_context_session_id())

return self._enqueue(msg, disable_geoip)

def capture_exception(

self,

exception: Optional[ExceptionArg],

**kwargs: Unpack[OptionalCaptureArgs],

):

"""

Capture an exception for error tracking.

Args:

exception: The exception to capture.

distinct_id: The distinct ID of the user.

properties: A dictionary of additional properties.

send_feature_flags: Whether to send feature flags with the exception.

disable_geoip: Whether to disable GeoIP for this event.

Examples:

```python

try:

# Some code that might fail

pass

except Exception as e:

posthog.capture_exception(e, 'user_distinct_id', properties=additional_properties)

```

Category:

Error Tracking

"""

distinct_id = kwargs.get("distinct_id", None)

properties = kwargs.get("properties", None)

send_feature_flags = kwargs.get("send_feature_flags", False)

disable_geoip = kwargs.get("disable_geoip", None)

# this function shouldn't ever throw an error, so it logs exceptions instead of raising them.

# this is important to ensure we don't unexpectedly re-raise exceptions in the user's code.

try:

properties = properties or {}

# Check if this exception has already been captured

if exception is not None and exception_is_already_captured(exception):

self.log.debug("Exception already captured, skipping")

return None