Analytics

`zenml.analytics` `special`

The 'analytics' module of ZenML.

`alias(user_id, previous_id)`

Alias user IDs.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`previous_id`	`UUID`	Previous ID for the alias.	required

Returns:

Type	Description
`bool`	True if event is sent successfully, False is not.

Source code in zenml/analytics/__init__.py

def alias(user_id: UUID, previous_id: UUID) -> bool:  # type: ignore[return]
    """Alias user IDs.

    Args:
        user_id: The user ID.
        previous_id: Previous ID for the alias.

    Returns:
        True if event is sent successfully, False is not.
    """
    from zenml.analytics.context import AnalyticsContext

    with AnalyticsContext() as analytics:
        return analytics.alias(user_id=user_id, previous_id=previous_id)

`group(group_id, group_metadata=None)`

Attach metadata to a segment group.

Parameters:

Name	Type	Description	Default
`group_id`	`UUID`	ID of the group.	required
`group_metadata`	`Optional[Dict[str, Any]]`	Metadata to attach to the group.	`None`

Returns:

Type	Description
`bool`	True if event is sent successfully, False if not.

Source code in zenml/analytics/__init__.py

def group(  # type: ignore[return]
    group_id: UUID,
    group_metadata: Optional[Dict[str, Any]] = None,
) -> bool:
    """Attach metadata to a segment group.

    Args:
        group_id: ID of the group.
        group_metadata: Metadata to attach to the group.

    Returns:
        True if event is sent successfully, False if not.
    """
    from zenml.analytics.context import AnalyticsContext

    with AnalyticsContext() as analytics:
        return analytics.group(group_id=group_id, traits=group_metadata)

`identify(metadata=None)`

Attach metadata to user directly.

Parameters:

Name	Type	Description	Default
`metadata`	`Optional[Dict[str, Any]]`	Dict of metadata to attach to the user.	`None`

Returns:

Type	Description
`bool`	True if event is sent successfully, False is not.

Source code in zenml/analytics/__init__.py

def identify(  # type: ignore[return]
    metadata: Optional[Dict[str, Any]] = None
) -> bool:
    """Attach metadata to user directly.

    Args:
        metadata: Dict of metadata to attach to the user.

    Returns:
        True if event is sent successfully, False is not.
    """
    from zenml.analytics.context import AnalyticsContext

    if metadata is None:
        return False

    with AnalyticsContext() as analytics:
        return analytics.identify(traits=metadata)

`track(event, metadata=None)`

Track segment event if user opted-in.

Parameters:

Name	Type	Description	Default
`event`	`AnalyticsEvent`	Name of event to track in segment.	required
`metadata`	`Optional[Dict[str, Any]]`	Dict of metadata to track.	`None`

Returns:

Type	Description
`bool`	True if event is sent successfully, False if not.

Source code in zenml/analytics/__init__.py

def track(  # type: ignore[return]
    event: "AnalyticsEvent",
    metadata: Optional[Dict[str, Any]] = None,
) -> bool:
    """Track segment event if user opted-in.

    Args:
        event: Name of event to track in segment.
        metadata: Dict of metadata to track.

    Returns:
        True if event is sent successfully, False if not.
    """
    from zenml.analytics.context import AnalyticsContext

    if metadata is None:
        metadata = {}

    metadata.setdefault("event_success", True)

    with AnalyticsContext() as analytics:
        return analytics.track(event=event, properties=metadata)

`client`

The analytics client of ZenML.

`Client`

The client class for ZenML analytics.

Source code in zenml/analytics/client.py

class Client(object):
    """The client class for ZenML analytics."""

    def __init__(self, send: bool = True, timeout: int = 15) -> None:
        """Initialization of the client.

        Args:
            send: Flag to determine whether to send the message.
            timeout: Timeout in seconds.
        """
        self.send = send
        self.timeout = timeout

    def identify(
        self, user_id: UUID, traits: Optional[Dict[Any, Any]]
    ) -> Tuple[bool, str]:
        """Method to identify a user with given traits.

        Args:
            user_id: The user ID.
            traits: The traits for the identification process.

        Returns:
            Tuple (success flag, the original message).
        """
        msg = {
            "user_id": user_id,
            "traits": traits or {},
            "type": "identify",
            "debug": IS_DEBUG_ENV,
        }
        return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

    def alias(self, user_id: UUID, previous_id: UUID) -> Tuple[bool, str]:
        """Method to alias user IDs.

        Args:
            user_id: The user ID.
            previous_id: Previous ID for the alias.

        Returns:
            Tuple (success flag, the original message).
        """
        msg = {
            "user_id": user_id,
            "previous_id": previous_id,
            "type": "alias",
            "debug": IS_DEBUG_ENV,
        }
        return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

    def track(
        self,
        user_id: UUID,
        event: "AnalyticsEvent",
        properties: Optional[Dict[Any, Any]],
    ) -> Tuple[bool, str]:
        """Method to track events.

        Args:
            user_id: The user ID.
            event: The type of the event.
            properties: Dict of additional properties for the event.

        Returns:
            Tuple (success flag, the original message).
        """
        msg = {
            "user_id": user_id,
            "event": event,
            "properties": properties or {},
            "type": "track",
            "debug": IS_DEBUG_ENV,
        }
        return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

    def group(
        self, user_id: UUID, group_id: UUID, traits: Optional[Dict[Any, Any]]
    ) -> Tuple[bool, str]:
        """Method to group users.

        Args:
            user_id: The user ID.
            group_id: The group ID.
            traits: Traits to assign to the group.

        Returns:
            Tuple (success flag, the original message).
        """
        msg = {
            "user_id": user_id,
            "group_id": group_id,
            "traits": traits or {},
            "type": "group",
            "debug": IS_DEBUG_ENV,
        }
        return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

    def _enqueue(self, msg: str) -> Tuple[bool, str]:
        """Method to queue messages to be sent.

        Args:
            msg: The message to queue.

        Returns:
            Tuple (success flag, the original message).
        """
        # if send is False, return msg as if it was successfully queued
        if not self.send:
            return True, msg

        post(timeout=self.timeout, batch=[msg])
        return True, msg

`init(self, send=True, timeout=15)` `special`

Initialization of the client.

Parameters:

Name	Type	Description	Default
`send`	`bool`	Flag to determine whether to send the message.	`True`
`timeout`	`int`	Timeout in seconds.	`15`

Source code in zenml/analytics/client.py

def __init__(self, send: bool = True, timeout: int = 15) -> None:
    """Initialization of the client.

    Args:
        send: Flag to determine whether to send the message.
        timeout: Timeout in seconds.
    """
    self.send = send
    self.timeout = timeout

`alias(self, user_id, previous_id)`

Method to alias user IDs.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`previous_id`	`UUID`	Previous ID for the alias.	required

Returns:

Type	Description
`Tuple[bool, str]`	Tuple (success flag, the original message).

Source code in zenml/analytics/client.py

def alias(self, user_id: UUID, previous_id: UUID) -> Tuple[bool, str]:
    """Method to alias user IDs.

    Args:
        user_id: The user ID.
        previous_id: Previous ID for the alias.

    Returns:
        Tuple (success flag, the original message).
    """
    msg = {
        "user_id": user_id,
        "previous_id": previous_id,
        "type": "alias",
        "debug": IS_DEBUG_ENV,
    }
    return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

`group(self, user_id, group_id, traits)`

Method to group users.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`group_id`	`UUID`	The group ID.	required
`traits`	`Optional[Dict[Any, Any]]`	Traits to assign to the group.	required

Returns:

Type	Description
`Tuple[bool, str]`	Tuple (success flag, the original message).

Source code in zenml/analytics/client.py

def group(
    self, user_id: UUID, group_id: UUID, traits: Optional[Dict[Any, Any]]
) -> Tuple[bool, str]:
    """Method to group users.

    Args:
        user_id: The user ID.
        group_id: The group ID.
        traits: Traits to assign to the group.

    Returns:
        Tuple (success flag, the original message).
    """
    msg = {
        "user_id": user_id,
        "group_id": group_id,
        "traits": traits or {},
        "type": "group",
        "debug": IS_DEBUG_ENV,
    }
    return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

`identify(self, user_id, traits)`

Method to identify a user with given traits.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`traits`	`Optional[Dict[Any, Any]]`	The traits for the identification process.	required

Returns:

Type	Description
`Tuple[bool, str]`	Tuple (success flag, the original message).

Source code in zenml/analytics/client.py

def identify(
    self, user_id: UUID, traits: Optional[Dict[Any, Any]]
) -> Tuple[bool, str]:
    """Method to identify a user with given traits.

    Args:
        user_id: The user ID.
        traits: The traits for the identification process.

    Returns:
        Tuple (success flag, the original message).
    """
    msg = {
        "user_id": user_id,
        "traits": traits or {},
        "type": "identify",
        "debug": IS_DEBUG_ENV,
    }
    return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

`track(self, user_id, event, properties)`

Method to track events.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`event`	`AnalyticsEvent`	The type of the event.	required
`properties`	`Optional[Dict[Any, Any]]`	Dict of additional properties for the event.	required

Returns:

Type	Description
`Tuple[bool, str]`	Tuple (success flag, the original message).

Source code in zenml/analytics/client.py

def track(
    self,
    user_id: UUID,
    event: "AnalyticsEvent",
    properties: Optional[Dict[Any, Any]],
) -> Tuple[bool, str]:
    """Method to track events.

    Args:
        user_id: The user ID.
        event: The type of the event.
        properties: Dict of additional properties for the event.

    Returns:
        Tuple (success flag, the original message).
    """
    msg = {
        "user_id": user_id,
        "event": event,
        "properties": properties or {},
        "type": "track",
        "debug": IS_DEBUG_ENV,
    }
    return self._enqueue(json.dumps(msg, cls=AnalyticsEncoder))

`context`

The analytics module of ZenML.

This module is based on the 'analytics-python' package created by Segment. The base functionalities are adapted to work with the ZenML analytics server.

`AnalyticsContext`

Client class for ZenML Analytics v2.

Source code in zenml/analytics/context.py

class AnalyticsContext:
    """Client class for ZenML Analytics v2."""

    def __init__(self) -> None:
        """Initialization.

        Use this as a context manager to ensure that analytics are initialized
        properly, only tracked when configured to do so and that any errors
        are handled gracefully.
        """
        self.analytics_opt_in: bool = False

        self.user_id: Optional[UUID] = None
        self.external_user_id: Optional[UUID] = None
        self.executed_by_service_account: Optional[bool] = None
        self.client_id: Optional[UUID] = None
        self.server_id: Optional[UUID] = None
        self.external_server_id: Optional[UUID] = None
        self.server_metadata: Optional[Dict[str, str]] = None

        self.database_type: Optional["ServerDatabaseType"] = None
        self.deployment_type: Optional["ServerDeploymentType"] = None

    def __enter__(self) -> "AnalyticsContext":
        """Enter analytics context manager.

        Returns:
            The analytics context.
        """
        # Fetch the analytics opt-in setting
        from zenml.config.global_config import GlobalConfiguration

        try:
            gc = GlobalConfiguration()
            store_info = gc.zen_store.get_store_info()

            if self.in_server:
                self.analytics_opt_in = store_info.analytics_enabled
            else:
                self.analytics_opt_in = gc.analytics_opt_in

            if not self.analytics_opt_in:
                return self

            # Fetch the `user_id`
            if self.in_server:
                from zenml.zen_server.auth import get_auth_context
                from zenml.zen_server.utils import server_config

                # If the code is running on the server, use the auth context.
                auth_context = get_auth_context()
                if auth_context is not None:
                    self.user_id = auth_context.user.id
                    self.executed_by_service_account = (
                        auth_context.user.is_service_account
                    )
                    self.external_user_id = auth_context.user.external_user_id

                self.external_server_id = server_config().external_server_id
            else:
                # If the code is running on the client, use the default user.
                active_user = gc.zen_store.get_user()
                self.user_id = active_user.id
                self.executed_by_service_account = (
                    active_user.is_service_account
                )
                self.external_user_id = active_user.external_user_id

            # Fetch the `client_id`
            if self.in_server:
                # If the code is running on the server, there is no client id.
                self.client_id = None
            else:
                # If the code is running on the client, attach the client id.
                self.client_id = gc.user_id

            self.server_id = store_info.id
            self.deployment_type = store_info.deployment_type
            self.database_type = store_info.database_type
            self.server_metadata = store_info.metadata
        except Exception as e:
            self.analytics_opt_in = False
            logger.debug(f"Analytics initialization failed: {e}")

        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> bool:
        """Exit context manager.

        Args:
            exc_type: Exception type.
            exc_val: Exception value.
            exc_tb: Exception traceback.

        Returns:
            True.
        """
        if exc_val is not None:
            logger.debug(f"Sending telemetry data failed: {exc_val}")

        return True

    @property
    def in_server(self) -> bool:
        """Flag to check whether the code is running in a ZenML server.

        Returns:
            True if running in a server, False otherwise.
        """
        return handle_bool_env_var(ENV_ZENML_SERVER)

    def identify(self, traits: Optional[Dict[str, Any]] = None) -> bool:
        """Identify the user through segment.

        Args:
            traits: Traits of the user.

        Returns:
            True if tracking information was sent, False otherwise.
        """
        success = False
        if self.analytics_opt_in and self.user_id is not None:
            success, _ = default_client.identify(
                user_id=self.user_id,
                traits=traits,
            )

        return success

    def alias(self, user_id: UUID, previous_id: UUID) -> bool:
        """Alias user IDs.

        Args:
            user_id: The user ID.
            previous_id: Previous ID for the alias.

        Returns:
            True if alias information was sent, False otherwise.
        """
        success = False
        if self.analytics_opt_in:
            success, _ = default_client.alias(
                user_id=user_id,
                previous_id=previous_id,
            )

        return success

    def group(
        self,
        group_id: UUID,
        traits: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """Group the user.

        Args:
            group_id: Group ID.
            traits: Traits of the group.

        Returns:
            True if tracking information was sent, False otherwise.
        """
        success = False
        if self.analytics_opt_in and self.user_id is not None:
            success, _ = default_client.group(
                user_id=self.user_id,
                group_id=group_id,
                traits=traits or {},
            )

        return success

    def track(
        self,
        event: "AnalyticsEvent",
        properties: Optional[Dict[str, Any]] = None,
    ) -> bool:
        """Track an event.

        Args:
            event: Event to track.
            properties: Event properties.

        Returns:
            True if tracking information was sent, False otherwise.
        """
        from zenml.analytics.enums import AnalyticsEvent

        if properties is None:
            properties = {}

        if (
            not self.analytics_opt_in
            and event.value
            not in {
                AnalyticsEvent.OPT_OUT_ANALYTICS,
                AnalyticsEvent.OPT_IN_ANALYTICS,
            }
            or self.user_id is None
        ):
            return False

        # add basics
        properties.update(Environment.get_system_info())
        properties.update(
            {
                "environment": get_environment(),
                "python_version": Environment.python_version(),
                "version": __version__,
                "client_id": str(self.client_id),
                "user_id": str(self.user_id),
                "server_id": str(self.server_id),
                "deployment_type": str(self.deployment_type),
                "database_type": str(self.database_type),
                "executed_by_service_account": self.executed_by_service_account,
            }
        )

        try:
            # Timezone as tzdata
            tz = (
                datetime.datetime.now(datetime.timezone.utc)
                .astimezone()
                .tzname()
            )
            if tz is not None:
                properties.update({"timezone": tz})

            # Language code such as "en_DE"
            language_code, encoding = locale.getlocale()
            if language_code is not None:
                properties.update({"locale": language_code})
        except Exception:
            pass

        if self.external_user_id:
            properties["external_user_id"] = self.external_user_id

        if self.external_server_id:
            properties["external_server_id"] = self.external_server_id

        if self.server_metadata:
            properties.update(self.server_metadata)

        for k, v in properties.items():
            if isinstance(v, UUID):
                properties[k] = str(v)

        success, _ = default_client.track(
            user_id=self.user_id,
            event=event,
            properties=properties,
        )

        logger.debug(
            f"Sending analytics: User: {self.user_id}, Event: {event}, "
            f"Metadata: {properties}"
        )

        return success

`in_server: bool` `property` `readonly`

Flag to check whether the code is running in a ZenML server.

Returns:

Type	Description
`bool`	True if running in a server, False otherwise.

`enter(self)` `special`

Enter analytics context manager.

Returns:

Type	Description
`AnalyticsContext`	The analytics context.

Source code in zenml/analytics/context.py

def __enter__(self) -> "AnalyticsContext":
    """Enter analytics context manager.

    Returns:
        The analytics context.
    """
    # Fetch the analytics opt-in setting
    from zenml.config.global_config import GlobalConfiguration

    try:
        gc = GlobalConfiguration()
        store_info = gc.zen_store.get_store_info()

        if self.in_server:
            self.analytics_opt_in = store_info.analytics_enabled
        else:
            self.analytics_opt_in = gc.analytics_opt_in

        if not self.analytics_opt_in:
            return self

        # Fetch the `user_id`
        if self.in_server:
            from zenml.zen_server.auth import get_auth_context
            from zenml.zen_server.utils import server_config

            # If the code is running on the server, use the auth context.
            auth_context = get_auth_context()
            if auth_context is not None:
                self.user_id = auth_context.user.id
                self.executed_by_service_account = (
                    auth_context.user.is_service_account
                )
                self.external_user_id = auth_context.user.external_user_id

            self.external_server_id = server_config().external_server_id
        else:
            # If the code is running on the client, use the default user.
            active_user = gc.zen_store.get_user()
            self.user_id = active_user.id
            self.executed_by_service_account = (
                active_user.is_service_account
            )
            self.external_user_id = active_user.external_user_id

        # Fetch the `client_id`
        if self.in_server:
            # If the code is running on the server, there is no client id.
            self.client_id = None
        else:
            # If the code is running on the client, attach the client id.
            self.client_id = gc.user_id

        self.server_id = store_info.id
        self.deployment_type = store_info.deployment_type
        self.database_type = store_info.database_type
        self.server_metadata = store_info.metadata
    except Exception as e:
        self.analytics_opt_in = False
        logger.debug(f"Analytics initialization failed: {e}")

    return self

`exit(self, exc_type, exc_val, exc_tb)` `special`

Exit context manager.

Parameters:

Name	Type	Description	Default
`exc_type`	`Optional[Type[BaseException]]`	Exception type.	required
`exc_val`	`Optional[BaseException]`	Exception value.	required
`exc_tb`	`Optional[traceback]`	Exception traceback.	required

Returns:

Type	Description
`bool`	True.

Source code in zenml/analytics/context.py

def __exit__(
    self,
    exc_type: Optional[Type[BaseException]],
    exc_val: Optional[BaseException],
    exc_tb: Optional[TracebackType],
) -> bool:
    """Exit context manager.

    Args:
        exc_type: Exception type.
        exc_val: Exception value.
        exc_tb: Exception traceback.

    Returns:
        True.
    """
    if exc_val is not None:
        logger.debug(f"Sending telemetry data failed: {exc_val}")

    return True

`init(self)` `special`

Initialization.

Use this as a context manager to ensure that analytics are initialized properly, only tracked when configured to do so and that any errors are handled gracefully.

Source code in zenml/analytics/context.py

def __init__(self) -> None:
    """Initialization.

    Use this as a context manager to ensure that analytics are initialized
    properly, only tracked when configured to do so and that any errors
    are handled gracefully.
    """
    self.analytics_opt_in: bool = False

    self.user_id: Optional[UUID] = None
    self.external_user_id: Optional[UUID] = None
    self.executed_by_service_account: Optional[bool] = None
    self.client_id: Optional[UUID] = None
    self.server_id: Optional[UUID] = None
    self.external_server_id: Optional[UUID] = None
    self.server_metadata: Optional[Dict[str, str]] = None

    self.database_type: Optional["ServerDatabaseType"] = None
    self.deployment_type: Optional["ServerDeploymentType"] = None

`alias(self, user_id, previous_id)`

Alias user IDs.

Parameters:

Name	Type	Description	Default
`user_id`	`UUID`	The user ID.	required
`previous_id`	`UUID`	Previous ID for the alias.	required

Returns:

Type	Description
`bool`	True if alias information was sent, False otherwise.

Source code in zenml/analytics/context.py

def alias(self, user_id: UUID, previous_id: UUID) -> bool:
    """Alias user IDs.

    Args:
        user_id: The user ID.
        previous_id: Previous ID for the alias.

    Returns:
        True if alias information was sent, False otherwise.
    """
    success = False
    if self.analytics_opt_in:
        success, _ = default_client.alias(
            user_id=user_id,
            previous_id=previous_id,
        )

    return success

`group(self, group_id, traits=None)`

Group the user.

Parameters:

Name	Type	Description	Default
`group_id`	`UUID`	Group ID.	required
`traits`	`Optional[Dict[str, Any]]`	Traits of the group.	`None`

Returns:

Type	Description
`bool`	True if tracking information was sent, False otherwise.

Source code in zenml/analytics/context.py

def group(
    self,
    group_id: UUID,
    traits: Optional[Dict[str, Any]] = None,
) -> bool:
    """Group the user.

    Args:
        group_id: Group ID.
        traits: Traits of the group.

    Returns:
        True if tracking information was sent, False otherwise.
    """
    success = False
    if self.analytics_opt_in and self.user_id is not None:
        success, _ = default_client.group(
            user_id=self.user_id,
            group_id=group_id,
            traits=traits or {},
        )

    return success

`identify(self, traits=None)`

Identify the user through segment.

Parameters:

Name	Type	Description	Default
`traits`	`Optional[Dict[str, Any]]`	Traits of the user.	`None`

Returns:

Type	Description
`bool`	True if tracking information was sent, False otherwise.

Source code in zenml/analytics/context.py

def identify(self, traits: Optional[Dict[str, Any]] = None) -> bool:
    """Identify the user through segment.

    Args:
        traits: Traits of the user.

    Returns:
        True if tracking information was sent, False otherwise.
    """
    success = False
    if self.analytics_opt_in and self.user_id is not None:
        success, _ = default_client.identify(
            user_id=self.user_id,
            traits=traits,
        )

    return success

`track(self, event, properties=None)`

Track an event.

Parameters:

Name	Type	Description	Default
`event`	`AnalyticsEvent`	Event to track.	required
`properties`	`Optional[Dict[str, Any]]`	Event properties.	`None`

Returns:

Type	Description
`bool`	True if tracking information was sent, False otherwise.

Source code in zenml/analytics/context.py

def track(
    self,
    event: "AnalyticsEvent",
    properties: Optional[Dict[str, Any]] = None,
) -> bool:
    """Track an event.

    Args:
        event: Event to track.
        properties: Event properties.

    Returns:
        True if tracking information was sent, False otherwise.
    """
    from zenml.analytics.enums import AnalyticsEvent

    if properties is None:
        properties = {}

    if (
        not self.analytics_opt_in
        and event.value
        not in {
            AnalyticsEvent.OPT_OUT_ANALYTICS,
            AnalyticsEvent.OPT_IN_ANALYTICS,
        }
        or self.user_id is None
    ):
        return False

    # add basics
    properties.update(Environment.get_system_info())
    properties.update(
        {
            "environment": get_environment(),
            "python_version": Environment.python_version(),
            "version": __version__,
            "client_id": str(self.client_id),
            "user_id": str(self.user_id),
            "server_id": str(self.server_id),
            "deployment_type": str(self.deployment_type),
            "database_type": str(self.database_type),
            "executed_by_service_account": self.executed_by_service_account,
        }
    )

    try:
        # Timezone as tzdata
        tz = (
            datetime.datetime.now(datetime.timezone.utc)
            .astimezone()
            .tzname()
        )
        if tz is not None:
            properties.update({"timezone": tz})

        # Language code such as "en_DE"
        language_code, encoding = locale.getlocale()
        if language_code is not None:
            properties.update({"locale": language_code})
    except Exception:
        pass

    if self.external_user_id:
        properties["external_user_id"] = self.external_user_id

    if self.external_server_id:
        properties["external_server_id"] = self.external_server_id

    if self.server_metadata:
        properties.update(self.server_metadata)

    for k, v in properties.items():
        if isinstance(v, UUID):
            properties[k] = str(v)

    success, _ = default_client.track(
        user_id=self.user_id,
        event=event,
        properties=properties,
    )

    logger.debug(
        f"Sending analytics: User: {self.user_id}, Event: {event}, "
        f"Metadata: {properties}"
    )

    return success

`enums`

Collection of analytics events for ZenML.

`AnalyticsEvent (str, Enum)`

Enum of events to track in segment.

Source code in zenml/analytics/enums.py

class AnalyticsEvent(str, Enum):
    """Enum of events to track in segment."""

    # Login
    DEVICE_VERIFIED = "Device verified"

    # Onboarding
    USER_ENRICHED = "User Enriched"

    # Pipelines
    RUN_PIPELINE = "Pipeline run"
    RUN_PIPELINE_ENDED = "Pipeline run ended"
    CREATE_PIPELINE = "Pipeline created"
    BUILD_PIPELINE = "Pipeline built"

    # Template
    GENERATE_TEMPLATE = "Template generated"

    # Components
    REGISTERED_STACK_COMPONENT = "Stack component registered"

    # Code repository
    REGISTERED_CODE_REPOSITORY = "Code repository registered"

    # Stack
    REGISTERED_STACK = "Stack registered"
    UPDATED_STACK = "Stack updated"

    # Trigger
    CREATED_TRIGGER = "Trigger created"
    UPDATED_TRIGGER = "Trigger updated"

    # Templates
    CREATED_RUN_TEMPLATE = "Run template created"
    EXECUTED_RUN_TEMPLATE = "Run templated executed"

    # Model Control Plane
    MODEL_DEPLOYED = "Model deployed"
    CREATED_MODEL = "Model created"
    CREATED_MODEL_VERSION = "Model Version created"

    # Analytics opt in and out
    OPT_IN_ANALYTICS = "Analytics opt-in"
    OPT_OUT_ANALYTICS = "Analytics opt-out"
    OPT_IN_OUT_EMAIL = "Response for Email prompt"

    # Examples
    RUN_ZENML_GO = "ZenML go"

    # Workspaces
    CREATED_WORKSPACE = "Workspace created"

    # Flavor
    CREATED_FLAVOR = "Flavor created"

    # Secret
    CREATED_SECRET = "Secret created"

    # Service connector
    CREATED_SERVICE_CONNECTOR = "Service connector created"

    # Service account and API keys
    CREATED_SERVICE_ACCOUNT = "Service account created"

    # Stack recipes
    RUN_STACK_RECIPE = "Stack recipe ran"
    DEPLOY_STACK = "Stack deployed"
    DESTROY_STACK = "Stack destroyed"

    # Stack component deploy
    DEPLOY_STACK_COMPONENT = "Stack component deployed"
    DESTROY_STACK_COMPONENT = "Stack component destroyed"

    # Full stack infrastructure deployment
    DEPLOY_FULL_STACK = "Full stack deployed"

    # Tag created
    CREATED_TAG = "Tag created"

    # ZenML server events
    ZENML_SERVER_DEPLOYED = "ZenML server deployed"
    ZENML_SERVER_DESTROYED = "ZenML server destroyed"

    # ZenML Hub events
    ZENML_HUB_PLUGIN_INSTALL = "ZenML Hub plugin installed"
    ZENML_HUB_PLUGIN_UNINSTALL = "ZenML Hub plugin uninstalled"
    ZENML_HUB_PLUGIN_CLONE = "ZenML Hub plugin pulled"
    ZENML_HUB_PLUGIN_SUBMIT = "ZenML Hub plugin pushed"

    # Server Settings
    SERVER_SETTINGS_UPDATED = "Server Settings Updated"

`format(self, format_spec)` `special`

Default object formatter.

Source code in zenml/analytics/enums.py

def __format__(self, format_spec):
    return str.__format__(str(self), format_spec)

`models`

Helper models for ZenML analytics.

`AnalyticsTrackedModelMixin (BaseModel)`

Mixin for models that are tracked through analytics events.

Classes that have information tracked in analytics events can inherit from this mixin and implement the abstract methods. The @track_method decorator will detect function arguments and return values that inherit from this class and will include the ANALYTICS_FIELDS attributes as tracking metadata.

Source code in zenml/analytics/models.py

class AnalyticsTrackedModelMixin(BaseModel):
    """Mixin for models that are tracked through analytics events.

    Classes that have information tracked in analytics events can inherit
    from this mixin and implement the abstract methods. The `@track_method`
    decorator will detect function arguments and return values that inherit
    from this class and will include the `ANALYTICS_FIELDS` attributes as
    tracking metadata.
    """

    ANALYTICS_FIELDS: ClassVar[List[str]] = []

    def get_analytics_metadata(self) -> Dict[str, Any]:
        """Get the analytics metadata for the model.

        Returns:
            Dict of analytics metadata.
        """
        metadata = {}
        for field_name in self.ANALYTICS_FIELDS:
            metadata[field_name] = getattr(self, field_name, None)
        return metadata

`get_analytics_metadata(self)`

Get the analytics metadata for the model.

Returns:

Type	Description
`Dict[str, Any]`	Dict of analytics metadata.

Source code in zenml/analytics/models.py

def get_analytics_metadata(self) -> Dict[str, Any]:
    """Get the analytics metadata for the model.

    Returns:
        Dict of analytics metadata.
    """
    metadata = {}
    for field_name in self.ANALYTICS_FIELDS:
        metadata[field_name] = getattr(self, field_name, None)
    return metadata

`request`

The 'analytics' module of ZenML.

This module is based on the 'analytics-python' package created by Segment. The base functionalities are adapted to work with the ZenML analytics server.

`post(batch, timeout=15)`

Post a batch of messages to the ZenML analytics server.

Parameters:

Name	Type	Description	Default
`batch`	`List[str]`	The messages to send.	required
`timeout`	`int`	Timeout in seconds.	`15`

Returns:

Type	Description
`Response`	The response.

Exceptions:

Type	Description
`AnalyticsAPIError`	If the post request has failed.

Source code in zenml/analytics/request.py

def post(batch: List[str], timeout: int = 15) -> requests.Response:
    """Post a batch of messages to the ZenML analytics server.

    Args:
        batch: The messages to send.
        timeout: Timeout in seconds.

    Returns:
        The response.

    Raises:
        AnalyticsAPIError: If the post request has failed.
    """
    from zenml.analytics import source_context

    headers = {
        "accept": "application/json",
        "content-type": "application/json",
        source_context.name: source_context.get().value,
    }
    response = requests.post(
        url=ANALYTICS_SERVER_URL + "/batch",
        headers=headers,
        data=f"[{','.join(batch)}]",
        timeout=timeout,
    )

    if response.status_code == 200:
        logger.debug("data uploaded successfully")
        return response

    raise AnalyticsAPIError(response.status_code, response.text)

`utils`

Utility functions and classes for ZenML analytics.

`AnalyticsAPIError (Exception)`

Custom exception class for API-related errors.

Source code in zenml/analytics/utils.py

class AnalyticsAPIError(Exception):
    """Custom exception class for API-related errors."""

    def __init__(self, status: int, message: str) -> None:
        """Initialization.

        Args:
            status: The status code of the response.
            message: The text of the response.
        """
        self.message = message
        self.status = status

    def __str__(self) -> str:
        """Method to represent the instance as a string.

        Returns:
            A representation of the message and the status code.
        """
        msg = "[ZenML Analytics] {1}: {0}"
        return msg.format(self.message, self.status)

`init(self, status, message)` `special`

Initialization.

Parameters:

Name	Type	Description	Default
`status`	`int`	The status code of the response.	required
`message`	`str`	The text of the response.	required

Source code in zenml/analytics/utils.py

def __init__(self, status: int, message: str) -> None:
    """Initialization.

    Args:
        status: The status code of the response.
        message: The text of the response.
    """
    self.message = message
    self.status = status

`str(self)` `special`

Method to represent the instance as a string.

Returns:

Type	Description
`str`	A representation of the message and the status code.

Source code in zenml/analytics/utils.py

def __str__(self) -> str:
    """Method to represent the instance as a string.

    Returns:
        A representation of the message and the status code.
    """
    msg = "[ZenML Analytics] {1}: {0}"
    return msg.format(self.message, self.status)

`AnalyticsEncoder (JSONEncoder)`

Helper encoder class for JSON serialization.

Source code in zenml/analytics/utils.py

class AnalyticsEncoder(json.JSONEncoder):
    """Helper encoder class for JSON serialization."""

    def default(self, obj: Any) -> Any:
        """The default method to handle UUID and 'AnalyticsEvent' objects.

        Args:
            obj: The object to encode.

        Returns:
            The encoded object.
        """
        from zenml.analytics.enums import AnalyticsEvent

        # If the object is UUID, we simply return the value of UUID
        if isinstance(obj, UUID):
            return str(obj)

        # If the object is an AnalyticsEvent, return its value
        elif isinstance(obj, AnalyticsEvent):
            return str(obj.value)

        return json.JSONEncoder.default(self, obj)

`default(self, obj)`

The default method to handle UUID and 'AnalyticsEvent' objects.

Parameters:

Name	Type	Description	Default
`obj`	`Any`	The object to encode.	required

Returns:

Type	Description
`Any`	The encoded object.

Source code in zenml/analytics/utils.py

def default(self, obj: Any) -> Any:
    """The default method to handle UUID and 'AnalyticsEvent' objects.

    Args:
        obj: The object to encode.

    Returns:
        The encoded object.
    """
    from zenml.analytics.enums import AnalyticsEvent

    # If the object is UUID, we simply return the value of UUID
    if isinstance(obj, UUID):
        return str(obj)

    # If the object is an AnalyticsEvent, return its value
    elif isinstance(obj, AnalyticsEvent):
        return str(obj.value)

    return json.JSONEncoder.default(self, obj)

`analytics_disabler`

Context manager which disables ZenML analytics.

Source code in zenml/analytics/utils.py

class analytics_disabler:
    """Context manager which disables ZenML analytics."""

    def __init__(self) -> None:
        """Initialization of the context manager."""
        self.original_value = os.environ.get(ENV_ZENML_ANALYTICS_OPT_IN)

    def __enter__(self) -> None:
        """Disable the analytics."""
        os.environ[ENV_ZENML_ANALYTICS_OPT_IN] = str(False)

    def __exit__(
        self,
        exc_type: Optional[Any],
        exc_value: Optional[Any],
        traceback: Optional[Any],
    ) -> None:
        """Set it back to the original state.

        Args:
            exc_type: The class of the exception
            exc_value: The instance of the exception
            traceback: The traceback of the exception
        """
        if self.original_value is not None:
            os.environ[ENV_ZENML_ANALYTICS_OPT_IN] = self.original_value
        else:
            del os.environ[ENV_ZENML_ANALYTICS_OPT_IN]

`enter(self)` `special`

Disable the analytics.

Source code in zenml/analytics/utils.py

def __enter__(self) -> None:
    """Disable the analytics."""
    os.environ[ENV_ZENML_ANALYTICS_OPT_IN] = str(False)

`exit(self, exc_type, exc_value, traceback)` `special`

Set it back to the original state.

Parameters:

Name	Type	Description	Default
`exc_type`	`Optional[Any]`	The class of the exception	required
`exc_value`	`Optional[Any]`	The instance of the exception	required
`traceback`	`Optional[Any]`	The traceback of the exception	required

Source code in zenml/analytics/utils.py

def __exit__(
    self,
    exc_type: Optional[Any],
    exc_value: Optional[Any],
    traceback: Optional[Any],
) -> None:
    """Set it back to the original state.

    Args:
        exc_type: The class of the exception
        exc_value: The instance of the exception
        traceback: The traceback of the exception
    """
    if self.original_value is not None:
        os.environ[ENV_ZENML_ANALYTICS_OPT_IN] = self.original_value
    else:
        del os.environ[ENV_ZENML_ANALYTICS_OPT_IN]

`init(self)` `special`

Initialization of the context manager.

Source code in zenml/analytics/utils.py

def __init__(self) -> None:
    """Initialization of the context manager."""
    self.original_value = os.environ.get(ENV_ZENML_ANALYTICS_OPT_IN)

`track_handler`

Context handler to enable tracking the success status of an event.

Source code in zenml/analytics/utils.py

class track_handler(object):
    """Context handler to enable tracking the success status of an event."""

    def __init__(
        self,
        event: AnalyticsEvent,
        metadata: Optional[Dict[str, Any]] = None,
    ):
        """Initialization of the context manager.

        Args:
            event: The type of the analytics event
            metadata: The metadata of the event.
        """
        self.event: AnalyticsEvent = event
        self.metadata: Dict[str, Any] = metadata or {}

    def __enter__(self) -> "track_handler":
        """Enter function of the event handler.

        Returns:
            the handler instance.
        """
        return self

    def __exit__(
        self,
        type_: Optional[Any],
        value: Optional[Any],
        traceback: Optional[Any],
    ) -> Any:
        """Exit function of the event handler.

        Checks whether there was a traceback and updates the metadata
        accordingly. Following the check, it calls the function to track the
        event.

        Args:
            type_: The class of the exception
            value: The instance of the exception
            traceback: The traceback of the exception

        """
        if traceback is not None:
            self.metadata.update({"event_success": False})
        else:
            self.metadata.update({"event_success": True})

        if type_ is not None:
            self.metadata.update({"event_error_type": type_.__name__})

        track(self.event, self.metadata)

`enter(self)` `special`

Enter function of the event handler.

Returns:

Type	Description
`track_handler`	the handler instance.

Source code in zenml/analytics/utils.py

def __enter__(self) -> "track_handler":
    """Enter function of the event handler.

    Returns:
        the handler instance.
    """
    return self

`exit(self, type_, value, traceback)` `special`

Exit function of the event handler.

Checks whether there was a traceback and updates the metadata accordingly. Following the check, it calls the function to track the event.

Parameters:

Name	Type	Description	Default
`type_`	`Optional[Any]`	The class of the exception	required
`value`	`Optional[Any]`	The instance of the exception	required
`traceback`	`Optional[Any]`	The traceback of the exception	required

Source code in zenml/analytics/utils.py

def __exit__(
    self,
    type_: Optional[Any],
    value: Optional[Any],
    traceback: Optional[Any],
) -> Any:
    """Exit function of the event handler.

    Checks whether there was a traceback and updates the metadata
    accordingly. Following the check, it calls the function to track the
    event.

    Args:
        type_: The class of the exception
        value: The instance of the exception
        traceback: The traceback of the exception

    """
    if traceback is not None:
        self.metadata.update({"event_success": False})
    else:
        self.metadata.update({"event_success": True})

    if type_ is not None:
        self.metadata.update({"event_error_type": type_.__name__})

    track(self.event, self.metadata)

`init(self, event, metadata=None)` `special`

Initialization of the context manager.

Parameters:

Name	Type	Description	Default
`event`	`AnalyticsEvent`	The type of the analytics event	required
`metadata`	`Optional[Dict[str, Any]]`	The metadata of the event.	`None`

Source code in zenml/analytics/utils.py

def __init__(
    self,
    event: AnalyticsEvent,
    metadata: Optional[Dict[str, Any]] = None,
):
    """Initialization of the context manager.

    Args:
        event: The type of the analytics event
        metadata: The metadata of the event.
    """
    self.event: AnalyticsEvent = event
    self.metadata: Dict[str, Any] = metadata or {}

`email_opt_int(opted_in, email, source)`

Track the event of the users response to the email prompt, identify them.

Parameters:

Name	Type	Description	Default
`opted_in`	`bool`	Did the user decide to opt-in	required
`email`	`Optional[str]`	The email the user optionally provided	required
`source`	`str`	Location when the user replied ["zenml go", "zenml server"]	required

Source code in zenml/analytics/utils.py

def email_opt_int(opted_in: bool, email: Optional[str], source: str) -> None:
    """Track the event of the users response to the email prompt, identify them.

    Args:
        opted_in: Did the user decide to opt-in
        email: The email the user optionally provided
        source: Location when the user replied ["zenml go", "zenml server"]
    """
    # If the user opted in, associate email with the anonymous distinct ID
    if opted_in and email is not None and email != "":
        identify(metadata={"email": email, "source": source})

    # Track that the user answered the prompt
    track(
        AnalyticsEvent.OPT_IN_OUT_EMAIL,
        {"opted_in": opted_in, "source": source},
    )

`track_decorator(event)`

Decorator to track event.

If the decorated function takes in a AnalyticsTrackedModelMixin object as an argument or returns one, it will be called to track the event. The return value takes precedence over the argument when determining which object is called to track the event.

If the decorated function is a method of a class that inherits from AnalyticsTrackerMixin, the parent object will be used to intermediate tracking analytics.

Parameters:

Name	Type	Description	Default
`event`	`AnalyticsEvent`	Event string to stamp with.	required

Returns:

Type	Description
`Callable[[~F], ~F]`	A decorator that applies the analytics tracking to a function.

Source code in zenml/analytics/utils.py

def track_decorator(event: AnalyticsEvent) -> Callable[[F], F]:
    """Decorator to track event.

    If the decorated function takes in a `AnalyticsTrackedModelMixin` object as
    an argument or returns one, it will be called to track the event. The return
    value takes precedence over the argument when determining which object is
    called to track the event.

    If the decorated function is a method of a class that inherits from
    `AnalyticsTrackerMixin`, the parent object will be used to intermediate
    tracking analytics.

    Args:
        event: Event string to stamp with.

    Returns:
        A decorator that applies the analytics tracking to a function.
    """

    def inner_decorator(func: F) -> F:
        """Inner decorator function.

        Args:
            func: Function to decorate.

        Returns:
            Decorated function.
        """

        @wraps(func)
        def inner_func(*args: Any, **kwargs: Any) -> Any:
            """Inner function.

            Args:
                *args: Arguments to be passed to the function.
                **kwargs: Keyword arguments to be passed to the function.

            Returns:
                Result of the function.
            """
            with track_handler(event=event) as handler:
                try:
                    for obj in list(args) + list(kwargs.values()):
                        if isinstance(obj, AnalyticsTrackedModelMixin):
                            handler.metadata = obj.get_analytics_metadata()
                            break
                except Exception as e:
                    logger.debug(f"Analytics tracking failure for {func}: {e}")

                result = func(*args, **kwargs)

                try:
                    if isinstance(result, AnalyticsTrackedModelMixin):
                        handler.metadata = result.get_analytics_metadata()
                except Exception as e:
                    logger.debug(f"Analytics tracking failure for {func}: {e}")

                return result

        return cast(F, inner_func)

    return inner_decorator

Analytics

zenml.analytics special

alias(user_id, previous_id)

group(group_id, group_metadata=None)

identify(metadata=None)

track(event, metadata=None)

client

Client

__init__(self, send=True, timeout=15) special

alias(self, user_id, previous_id)

group(self, user_id, group_id, traits)

identify(self, user_id, traits)

track(self, user_id, event, properties)

context

AnalyticsContext

in_server: bool property readonly

__enter__(self) special

__exit__(self, exc_type, exc_val, exc_tb) special

__init__(self) special

alias(self, user_id, previous_id)

group(self, group_id, traits=None)

identify(self, traits=None)

track(self, event, properties=None)

enums

AnalyticsEvent (str, Enum)

__format__(self, format_spec) special

models

AnalyticsTrackedModelMixin (BaseModel)

get_analytics_metadata(self)

request

post(batch, timeout=15)

utils

AnalyticsAPIError (Exception)

__init__(self, status, message) special

__str__(self) special

AnalyticsEncoder (JSONEncoder)

default(self, obj)

analytics_disabler

__enter__(self) special

__exit__(self, exc_type, exc_value, traceback) special

__init__(self) special

track_handler

__enter__(self) special

__exit__(self, type_, value, traceback) special

__init__(self, event, metadata=None) special

email_opt_int(opted_in, email, source)

track_decorator(event)

`zenml.analytics` `special`

`alias(user_id, previous_id)`

`group(group_id, group_metadata=None)`

`identify(metadata=None)`

`track(event, metadata=None)`

`client`

`Client`

`init(self, send=True, timeout=15)` `special`

`alias(self, user_id, previous_id)`

`group(self, user_id, group_id, traits)`

`identify(self, user_id, traits)`

`track(self, user_id, event, properties)`

`context`

`AnalyticsContext`

`in_server: bool` `property` `readonly`

`enter(self)` `special`

`exit(self, exc_type, exc_val, exc_tb)` `special`

`init(self)` `special`

`alias(self, user_id, previous_id)`

`group(self, group_id, traits=None)`

`identify(self, traits=None)`

`track(self, event, properties=None)`

`enums`

`AnalyticsEvent (str, Enum)`

`format(self, format_spec)` `special`

`models`

`AnalyticsTrackedModelMixin (BaseModel)`

`get_analytics_metadata(self)`

`request`

`post(batch, timeout=15)`

`utils`

`AnalyticsAPIError (Exception)`

`init(self, status, message)` `special`

`str(self)` `special`

`AnalyticsEncoder (JSONEncoder)`

`default(self, obj)`

`analytics_disabler`

`enter(self)` `special`

`exit(self, exc_type, exc_value, traceback)` `special`

`init(self)` `special`

`track_handler`

`enter(self)` `special`

`exit(self, type_, value, traceback)` `special`

`init(self, event, metadata=None)` `special`

`email_opt_int(opted_in, email, source)`

`track_decorator(event)`