Package glean

Top-level package for Glean SDK.

Sub-modules

glean.config

Provides an object to pass configuration to Glean.

glean.glean

The main Glean general API.

glean.metrics

This module contains all of the metric types.

glean.net

Network functionality for Glean.

glean.testing

Utilities for writing unit tests involving Glean.

Functions

def load_metrics(filepath: str | pathlib._local.Path | List[str | pathlib._local.Path],
config: dict | None = None) ‑> Any

Expand source code

def load_metrics(
    filepath: Union[Union[str, Path], List[Union[str, Path]]],
    config: Optional[dict] = None,
) -> Any:
    """
    Load metrics from a `metrics.yaml` file.

    Args:
        filepath (Path): The path to the file, or a list of paths, to load.
        config (dict): A dictionary of options that change parsing behavior.
            These are documented in glean_parser:
            https://mozilla.github.io/glean_parser/glean_parser.html#glean_parser.parser.parse_objects
    Returns:
        metrics (object): An object containing a tree of metrics, as defined in
            the `metrics.yaml` file.
    Example:
        >>> metrics = load_metrics("metrics.yaml")
        >>> metrics.category.name.set("value")
    """
    if config is None:
        config = {}

    if not isinstance(filepath, list):
        filepath = [filepath]

    filepath = [Path(x) for x in filepath]

    result = parse_objects(filepath, config)

    errors = list(result)
    if len(errors):
        raise ValueError("\n\n".join(errors))

    metrics = result.value
    if len(metrics) == 0:
        raise ValueError(f"Didn't find any metrics in '{filepath}'")

    root = type("Metrics", (object,), {})

    for category_name, category in metrics.items():
        cursor = root
        for part in category_name.split("."):
            if not hasattr(cursor, part):
                setattr(cursor, part, type(category_name, (object,), {}))
            cursor = getattr(cursor, part)
        for name, metric in category.items():
            for actual_name, glean_metric in _get_metric_objects(name, metric):
                setattr(cursor, _normalize_name(actual_name), glean_metric)

    return root

Load metrics from a metrics.yaml file.

Args

filepath : Path

The path to the file, or a list of paths, to load.

config : dict

A dictionary of options that change parsing behavior. These are documented in glean_parser: https://mozilla.github.io/glean_parser/glean_parser.html#glean_parser.parser.parse_objects

Returns

metrics (object): An object containing a tree of metrics, as defined in the metrics.yaml file.

Example

>>> metrics = load_metrics("metrics.yaml")
>>> metrics.category.name.set("value")

def load_pings(filepath: str | pathlib._local.Path | List[str | pathlib._local.Path],
config: dict | None = None) ‑> Any

Expand source code

def load_pings(
    filepath: Union[Union[str, Path], List[Union[str, Path]]],
    config: Optional[dict] = None,
) -> Any:
    """
    Load pings from a `pings.yaml` file.

    Args:
        filepath (Path): The path to the file, or a list of paths, to load.
        config (dict): A dictionary of options that change parsing behavior.
            These are documented in glean_parser:
            https://mozilla.github.io/glean_parser/glean_parser.html#glean_parser.parser.parse_objects
    Returns:
        pings (object): An object where the attributes are pings, as defined in
            the `pings.yaml` file.
    Example:
        >>> pings = load_pings("pings.yaml")
        >>> pings.baseline.submit()
    """
    metrics = load_metrics(filepath, config)

    return metrics.pings

Load pings from a pings.yaml file.

Args

filepath : Path

The path to the file, or a list of paths, to load.

config : dict

A dictionary of options that change parsing behavior. These are documented in glean_parser: https://mozilla.github.io/glean_parser/glean_parser.html#glean_parser.parser.parse_objects

Returns

pings (object): An object where the attributes are pings, as defined in the pings.yaml file.

Example

>>> pings = load_pings("pings.yaml")
>>> pings.baseline.submit()

Classes

class Configuration (server_endpoint: str | None = None,
channel: str | None = None,
max_events: int = 500,
ping_uploader: BaseUploader | None = None,
allow_multiprocessing: bool = True,
enable_event_timestamps: bool = True,
experimentation_id: str | None = None,
enable_internal_pings: bool = True,
max_pending_pings_count: int | None = None,
max_pending_pings_directory_size: int | None = None,
session_mode: glean._uniffi.glean.SessionMode = SessionMode.AUTO,
session_sample_rate: float = 1.0,
session_inactivity_timeout_ms: int = 1800000)

Expand source code

class Configuration:
    """
    Configuration values for Glean.
    """

    def __init__(
        self,
        server_endpoint: Optional[str] = None,
        channel: Optional[str] = None,
        max_events: int = DEFAULT_MAX_EVENTS,
        ping_uploader: Optional[net.BaseUploader] = None,
        allow_multiprocessing: bool = True,
        enable_event_timestamps: bool = True,
        experimentation_id: Optional[str] = None,
        enable_internal_pings: bool = True,
        max_pending_pings_count: Optional[int] = None,
        max_pending_pings_directory_size: Optional[int] = None,
        session_mode: SessionMode = SessionMode.AUTO,
        session_sample_rate: float = 1.0,
        session_inactivity_timeout_ms: int = DEFAULT_SESSION_INACTIVITY_TIMEOUT_MS,
    ):
        """
        Args:
            server_endpoint (str): Optional. The server pings are sent to.
                Defaults to `DEFAULT_TELEMETRY_ENDPOINT`.
            channel (str): Optional. The release channel the application is on,
                if known.
            max_events (int): Optional.The number of events to store before
                force-sending. Defaults to `DEFAULT_MAX_EVENTS`.
            ping_uploader (glean.net.BaseUploader): Optional. The ping uploader
                implementation. Defaults to `glean.net.HttpClientUploader`.
            allow_multiprocessing (bool): When True (default), use a subprocess
                to offload some work (such as ping uploading).
            enable_event_timestamps (bool): Whether to add a wallclock timestamp
                to all events. Default: `True`.
            experimentation_id (string): An experimentation identifier derived
                by the application to be sent with all pings. Default: None.
            enable_internal_pings (bool): Whether to enable internal pings. Default: `True`.
            max_pending_pings_count (int): Optional. The maximum number of pending
                pings stored on disk. When exceeded, the oldest pings are deleted.
                Defaults to 500.
            max_pending_pings_directory_size (int): Optional. The maximum size in
                bytes of the pending pings directory. When exceeded, the oldest pings
                are deleted. Defaults to 50 MB.
            session_mode (SessionMode): How Glean manages session boundaries.
                Default: `SessionMode.AUTO`.
            session_sample_rate (float): Session sampling rate (0.0–1.0).
                Default: `1.0`.
            session_inactivity_timeout_ms (int): Inactivity timeout (milliseconds)
                before AUTO-mode sessions expire. Default: 30 minutes.
        """
        if server_endpoint is None:
            server_endpoint = DEFAULT_TELEMETRY_ENDPOINT
        self._server_endpoint = server_endpoint
        self._channel = channel
        self._max_events = max_events
        if ping_uploader is None:
            ping_uploader = net.HttpClientUploader()
        self._ping_uploader = ping_uploader
        self._allow_multiprocessing = allow_multiprocessing
        self._enable_event_timestamps = enable_event_timestamps
        self._experimentation_id = experimentation_id
        self._enable_internal_pings = enable_internal_pings
        self._max_pending_pings_count = max_pending_pings_count
        self._max_pending_pings_directory_size = max_pending_pings_directory_size
        self._session_mode = session_mode
        self._session_sample_rate = session_sample_rate
        self._session_inactivity_timeout_ms = session_inactivity_timeout_ms

    @property
    def server_endpoint(self) -> str:
        """The server pings are sent to."""
        return self._server_endpoint

    @server_endpoint.setter
    def server_endpoint(self, value: str):
        self._server_endpoint = value

    @property
    def channel(self) -> Optional[str]:
        """The release channel the application is on, if known."""
        return self._channel

    @channel.setter
    def channel(self, value: str):
        from ._builtins import metrics

        self._channel = value

        metrics.glean.internal.metrics.app_channel.set(value)

    @property
    def max_events(self) -> int:
        """The number of events to store before force-sending."""
        return self._max_events

    # max_events can't be changed after Glean is initialized

    @property
    def enable_event_timestamps(self) -> bool:
        """Whether to add a wallclock timestamp to all events."""
        return self._enable_event_timestamps

    @property
    def experimentation_id(self) -> Optional[str]:
        """An experimentation id that will be sent in all pings"""
        return self._experimentation_id

    @property
    def enable_internal_pings(self) -> bool:
        """Whether to enable internal pings."""
        return self._enable_internal_pings

    @property
    def max_pending_pings_count(self) -> Optional[int]:
        """The maximum number of pending pings stored on disk."""
        return self._max_pending_pings_count

    @property
    def max_pending_pings_directory_size(self) -> Optional[int]:
        """The maximum size in bytes of the pending pings directory."""
        return self._max_pending_pings_directory_size

    @property
    def ping_uploader(self) -> net.BaseUploader:
        """The ping uploader implementation."""
        return self._ping_uploader

    @ping_uploader.setter
    def ping_uploader(self, value: net.BaseUploader):
        self._ping_uploader = value

    @property
    def session_mode(self) -> SessionMode:
        """How Glean manages session boundaries."""
        return self._session_mode

    @property
    def session_sample_rate(self) -> float:
        """Session sampling rate (0.0–1.0)."""
        return self._session_sample_rate

    @property
    def session_inactivity_timeout_ms(self) -> int:
        """Inactivity timeout (milliseconds) before AUTO-mode sessions expire."""
        return self._session_inactivity_timeout_ms

Configuration values for Glean.

Args

server_endpoint : str

Optional. The server pings are sent to. Defaults to DEFAULT_TELEMETRY_ENDPOINT.

channel : str

Optional. The release channel the application is on, if known.

max_events : int

Optional.The number of events to store before force-sending. Defaults to DEFAULT_MAX_EVENTS.

ping_uploader : BaseUploader

Optional. The ping uploader implementation. Defaults to HttpClientUploader.

allow_multiprocessing : bool

When True (default), use a subprocess to offload some work (such as ping uploading).

enable_event_timestamps : bool

Whether to add a wallclock timestamp to all events. Default: True.

experimentation_id : string

An experimentation identifier derived by the application to be sent with all pings. Default: None.

enable_internal_pings : bool

Whether to enable internal pings. Default: True.

max_pending_pings_count : int

Optional. The maximum number of pending pings stored on disk. When exceeded, the oldest pings are deleted. Defaults to 500.

max_pending_pings_directory_size : int

Optional. The maximum size in bytes of the pending pings directory. When exceeded, the oldest pings are deleted. Defaults to 50 MB.

session_mode : SessionMode

How Glean manages session boundaries. Default: SessionMode.AUTO.

session_sample_rate : float

Session sampling rate (0.0–1.0). Default: 1.0.

session_inactivity_timeout_ms : int

Inactivity timeout (milliseconds) before AUTO-mode sessions expire. Default: 30 minutes.

Instance variables

prop channel : str | None

Expand source code

@property
def channel(self) -> Optional[str]:
    """The release channel the application is on, if known."""
    return self._channel

The release channel the application is on, if known.

prop enable_event_timestamps : bool

Expand source code

@property
def enable_event_timestamps(self) -> bool:
    """Whether to add a wallclock timestamp to all events."""
    return self._enable_event_timestamps

Whether to add a wallclock timestamp to all events.

prop enable_internal_pings : bool

Expand source code

@property
def enable_internal_pings(self) -> bool:
    """Whether to enable internal pings."""
    return self._enable_internal_pings

Whether to enable internal pings.

prop experimentation_id : str | None

Expand source code

@property
def experimentation_id(self) -> Optional[str]:
    """An experimentation id that will be sent in all pings"""
    return self._experimentation_id

An experimentation id that will be sent in all pings

prop max_events : int

Expand source code

@property
def max_events(self) -> int:
    """The number of events to store before force-sending."""
    return self._max_events

The number of events to store before force-sending.

prop max_pending_pings_count : int | None

Expand source code

@property
def max_pending_pings_count(self) -> Optional[int]:
    """The maximum number of pending pings stored on disk."""
    return self._max_pending_pings_count

The maximum number of pending pings stored on disk.

prop max_pending_pings_directory_size : int | None

Expand source code

@property
def max_pending_pings_directory_size(self) -> Optional[int]:
    """The maximum size in bytes of the pending pings directory."""
    return self._max_pending_pings_directory_size

The maximum size in bytes of the pending pings directory.

prop ping_uploader : BaseUploader

Expand source code

@property
def ping_uploader(self) -> net.BaseUploader:
    """The ping uploader implementation."""
    return self._ping_uploader

The ping uploader implementation.

prop server_endpoint : str

Expand source code

@property
def server_endpoint(self) -> str:
    """The server pings are sent to."""
    return self._server_endpoint

The server pings are sent to.

prop session_inactivity_timeout_ms : int

Expand source code

@property
def session_inactivity_timeout_ms(self) -> int:
    """Inactivity timeout (milliseconds) before AUTO-mode sessions expire."""
    return self._session_inactivity_timeout_ms

Inactivity timeout (milliseconds) before AUTO-mode sessions expire.

prop session_mode : glean._uniffi.glean.SessionMode

Expand source code

@property
def session_mode(self) -> SessionMode:
    """How Glean manages session boundaries."""
    return self._session_mode

How Glean manages session boundaries.

prop session_sample_rate : float

Expand source code

@property
def session_sample_rate(self) -> float:
    """Session sampling rate (0.0–1.0)."""
    return self._session_sample_rate

Session sampling rate (0.0–1.0).

class Glean

Expand source code

class Glean:
    """
    The main Glean API.

    Before any data collection can take place, the Glean SDK **must** be
    initialized from the application.

    >>> Glean.initialize(
    ...     application_id="my-app",
    ...     application_version="0.0.0",
    ...     upload_enabled=True,
    ...     data_dir=Path.home() / ".glean",
    ... )
    """

    # Whether Glean was initialized
    _initialized: bool = False
    # Set when `initialize()` returns.
    # This allows to detect calls that happen before `Glean.initialize()` was called.
    # Note: The initialization might still be in progress, as it runs in a separate thread.
    _init_finished: bool = False

    # Are we in testing mode?
    _testing_mode: bool = False

    # The Configuration that was passed to `initialize`
    _configuration: Configuration

    # The directory that Glean stores data in
    _data_dir: Path = Path()

    # Whether Glean "owns" the data directory and should destroy it upon reset.
    _destroy_data_dir: bool = False

    # Keep track of this setting before Glean is initialized
    _upload_enabled: bool = True

    # The ping types, so they can be registered prior to Glean initialization,
    # and saved between test runs.
    _ping_type_queue: Set["PingType"] = set()

    # The application id to send in the ping.
    _application_id: str

    # The version of the application sending Glean data.
    _application_version: str

    # The build identifier generated by the CI system.
    _application_build_id: str

    # A thread lock for Glean operations that need to be synchronized
    _thread_lock = threading.RLock()

    # Simple logging API log level
    _simple_log_level: Optional[int] = None

    @classmethod
    def initialize(
        cls,
        application_id: str,
        application_version: str,
        upload_enabled: bool,
        configuration: Optional[Configuration] = None,
        data_dir: Optional[Path] = None,
        application_build_id: Optional[str] = None,
        log_level: Optional[int] = None,
    ) -> None:
        """
        Initialize the Glean SDK.

        This should only be initialized once by the application, and not by
        libraries using the Glean SDK. A message is logged to error and no
        changes are made to the state if initialize is called a more than
        once.

        Args:
            application_id (str): The application id to use when sending pings.
            application_version (str): The version of the application sending
                Glean data. The meaning of this field is application-specific,
                but it is highly recommended to set this to something
                meaningful.
            upload_enabled (bool): Controls whether telemetry is enabled. If
                disabled, all persisted metrics, events and queued pings
                (except first_run_date) are cleared.
            configuration (glean.config.Configuration): (optional) An object with
                global settings.
            data_dir (pathlib.Path): The path to the Glean data directory.
            application_build_id (str): (optional) The build identifier generated
                by the CI system (e.g. "1234/A").
            log_level (int): (optional) The level of log messages that Glean
                will emit. One of the constants in the Python `logging` module:
                `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`. If you need a
                specialized logging configuration, such as to redirecting,
                filtering or reformatting, you should use the Python `logging`
                module's API directly, but that will not affect logging any of
                Glean's networking operations which happen in a subprocess.
                Details in the "Debugging Python applications with the Glean
                SDK" chapter in the docs.
        """
        if log_level is not None:
            cls._simple_log_level = log_level
            logging.basicConfig(level=log_level)

        with cls._thread_lock:
            if cls.is_initialized():
                return

            atexit.register(Glean._reset)

            if configuration is None:
                configuration = Configuration()

            if data_dir is None:
                raise TypeError("data_dir must be provided")
            cls._data_dir = data_dir
            cls._destroy_data_dir = False

            cls._configuration = configuration
            cls._application_id = application_id

            if application_version is None:
                cls._application_version = "Unknown"
            else:
                cls._application_version = application_version

            if application_build_id is None:
                cls._application_build_id = "Unknown"
            else:
                cls._application_build_id = application_build_id

        # FIXME: Require user to pass in build-date
        dt = _uniffi.Datetime(
            year=1970,
            month=1,
            day=1,
            hour=0,
            minute=0,
            second=0,
            nanosecond=0,
            offset_seconds=0,
        )
        client_info = _uniffi.ClientInfoMetrics(
            app_build=cls._application_build_id,
            app_display_version=cls._application_version,
            app_build_date=dt,
            channel=configuration.channel,
            architecture=platform.machine() or "Unknown",
            os_version="Unknown",
            locale=None,
            device_manufacturer=None,
            device_model=None,
            android_sdk_version=None,
            windows_build_number=None,
        )
        callbacks = OnGleanEventsImpl(cls)
        cfg = _uniffi.InternalConfiguration(
            data_path=str(cls._data_dir),
            application_id=application_id,
            language_binding_name="Python",
            upload_enabled=upload_enabled,
            max_events=configuration.max_events,
            delay_ping_lifetime_io=False,
            use_core_mps=False,
            app_build=cls._application_build_id,
            trim_data_to_registered_pings=False,
            log_level=None,
            rate_limit=None,
            enable_event_timestamps=configuration.enable_event_timestamps,
            experimentation_id=configuration.experimentation_id,
            enable_internal_pings=configuration.enable_internal_pings,
            ping_schedule={},
            ping_lifetime_threshold=0,
            ping_lifetime_max_time=0,
            max_pending_pings_count=configuration.max_pending_pings_count,
            max_pending_pings_directory_size=configuration.max_pending_pings_directory_size,
            session_mode=configuration.session_mode,
            session_sample_rate=configuration.session_sample_rate,
            session_inactivity_timeout_ms=configuration.session_inactivity_timeout_ms,
        )

        _uniffi.glean_initialize(cfg, client_info, callbacks)
        cls._initialized = True

    @classmethod
    def _initialize_with_tempdir_for_testing(
        cls,
        application_id: str,
        application_version: str,
        upload_enabled: bool,
        configuration: Optional[Configuration] = None,
        application_build_id: Optional[str] = None,
    ) -> None:
        """
        Initialize Glean to use a temporary data directory. Use for internal
        unit testing only.

        The temporary directory will be destroyed when Glean is initialized
        again or at process shutdown.
        """

        actual_data_dir = Path(tempfile.TemporaryDirectory().name)
        cls.initialize(
            application_id,
            application_version,
            upload_enabled,
            configuration=configuration,
            data_dir=actual_data_dir,
            application_build_id=application_build_id,
        )
        cls._destroy_data_dir = True

    @_util.classproperty
    def configuration(cls) -> Configuration:
        """
        Access the configuration object to change dynamic parameters.
        """
        return cls._configuration

    @classmethod
    def _reset(cls) -> None:
        """
        Resets the Glean singleton.
        """
        # TODO: 1594184 Send the metrics ping
        log.debug("Resetting Glean")

        # Wait for the subprocess to complete.  We only need to do this if
        # we know we are going to be deleting the data directory.
        if cls._destroy_data_dir and cls._data_dir.exists():
            ProcessDispatcher._wait_for_last_process()

        # Destroy the Glean object.
        # Importantly on Windows, this closes the handle to the database so
        # that the data directory can be deleted without a multiple access
        # violation.
        _uniffi.glean_test_destroy_glean(False)

        _uniffi.glean_set_test_mode(False)
        cls._init_finished = False
        cls._initialized = False
        cls._testing_mode = False

        # Remove the atexit handler or it will get called multiple times at
        # exit.
        atexit.unregister(cls._reset)

        if cls._destroy_data_dir and cls._data_dir.exists():
            # This needs to be run in the same one-at-a-time process as the
            # PingUploadWorker to avoid a race condition. This will block the
            # main thread waiting for all pending uploads to complete, but this
            # only happens during testing when the data directory is a
            # temporary directory, so there is no concern about delaying
            # application shutdown here.
            p = ProcessDispatcher.dispatch(_rmtree, (str(cls._data_dir),))
            p.wait()

    @classmethod
    def is_initialized(cls) -> bool:
        """
        Returns True if the Glean SDK has been initialized.
        """
        return cls._initialized

    @classmethod
    def set_upload_enabled(cls, enabled: bool) -> None:
        """
        **DEPRECATED** Enable or disable Glean collection and upload.

        Metric collection is enabled by default.

        When uploading is disabled, metrics aren't recorded at all and no data
        is uploaded.

        When disabling, all pending metrics, events and queued pings are cleared.

        When enabling, the core Glean metrics are recreated.

        **DEPRECATION NOTICE**:
        This API is deprecated. Use `set_collection_enabled` instead.

        Args:
            enabled (bool): When True, enable metric collection.
        """
        # Changing upload enabled always happens asynchronous.
        # That way it follows what a user expect when calling it inbetween other calls:
        # It executes in the right order.
        #
        # Because the dispatch queue is halted until Glean is fully initialized
        # we can safely enqueue here and it will execute after initialization.
        _uniffi.glean_set_upload_enabled(enabled)

    @classmethod
    def set_collection_enabled(cls, enabled: bool) -> None:
        """
        Enable or disable Glean collection and upload.

        Metric collection is enabled by default.

        When collection is disabled, metrics aren't recorded at all and no data
        is uploaded.
        **Note**: Individual pings can be enabled if they don't follow this setting.
        See `PingType.set_enabled`.

        When disabling, all pending metrics, events and queued pings are cleared.

        When enabling, the core Glean metrics are recreated.

        Args:
            enabled (bool): When True, enable metric collection.
        """
        cls.set_upload_enabled(enabled)

    @classmethod
    def set_experiment_active(
        cls, experiment_id: str, branch: str, extra: Optional[Dict[str, str]] = None
    ) -> None:
        """
        Indicate that an experiment is running. Glean will then add an
        experiment annotation to the environment which is sent with pings. This
        information is not persisted between runs.

        Args:
            experiment_id (str): The id of the active experiment (maximum 100
                bytes)
            branch (str): The experiment branch (maximum 100 bytes)
            extra (dict of str -> str): Optional metadata to output with the
                ping
        """
        map = {} if extra is None else extra
        _uniffi.glean_set_experiment_active(experiment_id, branch, map)

    @classmethod
    def set_experiment_inactive(cls, experiment_id: str) -> None:
        """
        Indicate that the experiment is no longer running.

        Args:
            experiment_id (str): The id of the experiment to deactivate.
        """
        _uniffi.glean_set_experiment_inactive(experiment_id)

    @classmethod
    def test_is_experiment_active(cls, experiment_id: str) -> bool:
        """
        Tests whether an experiment is active, for testing purposes only.

        Args:
            experiment_id (str): The id of the experiment to look for.

        Returns:
            is_active (bool): If the experiement is active and reported in
                pings.
        """
        return _uniffi.glean_test_get_experiment_data(experiment_id) is not None

    @classmethod
    def test_get_experiment_data(cls, experiment_id: str) -> "RecordedExperiment":
        """
        Returns the stored data for the requested active experiment, for testing purposes only.

        Args:
            experiment_id (str): The id of the experiment to look for.

        Returns:
            experiment_data (RecordedExperiment): The data associated with
                the experiment.
        """
        data = _uniffi.glean_test_get_experiment_data(experiment_id)
        if data is not None:
            return data
        else:
            raise RuntimeError("Experiment data is not set")

    @classmethod
    def set_experimentation_id(cls, experimentation_id: str):
        """
        Dynamically set the experimentation identifier, as opposed to setting it through
        the configuration during initialization.

        Args:
            experimentation_id (str): The string experimentation identifier to set
        """
        _uniffi.glean_set_experimentation_id(experimentation_id)

    @classmethod
    def test_get_experimentation_id(cls) -> str:
        """
        Returns the stored experimentation id, for testing purposes only.

        Returns:
            experimentation_id (str): The experimentation id set by the client.
        """
        experimentation_id = _uniffi.glean_test_get_experimentation_id()
        if experimentation_id is not None:
            return experimentation_id
        else:
            raise RuntimeError("Experimentation id is not set")

    @classmethod
    def handle_client_active(cls):
        """
        Performs the collection/cleanup operations required by becoming active.

        This functions generates a baseline ping with reason `active`
        and then sets the dirty bit.
        This should be called whenever the consuming product becomes active (e.g.
        getting to foreground).
        """
        _uniffi.glean_handle_client_active()

    @classmethod
    def handle_client_inactive(cls):
        """
        Performs the collection/cleanup operations required by becoming inactive.

        This functions generates a baseline and an events ping with reason
        `inactive` and then clears the dirty bit.
        This should be called whenever the consuming product becomes inactive (e.g.
        getting to background).
        """
        _uniffi.glean_handle_client_inactive()

    @classmethod
    def session_start(cls):
        """
        Starts a session manually.

        Only has an effect when Glean is configured with `SessionMode.MANUAL`.
        In `AUTO` or `LIFECYCLE` mode this is a no-op so automatic session
        state isn't corrupted.
        """
        _uniffi.glean_session_start()

    @classmethod
    def session_end(cls, reason: Optional[str] = None):
        """
        Ends a session manually.

        Only has an effect when Glean is configured with `SessionMode.MANUAL`.

        Args:
            reason (str): Optional application-provided string attached to the
                `glean.session_end` boundary event for downstream analysis.
        """
        _uniffi.glean_session_end(reason)

    @classmethod
    def shutdown(cls):
        """
        Shuts down Glean in an orderly fashion.
        """
        _uniffi.glean_shutdown()

        # On top of the Glean shutdown
        # we also wait for the process dispatcher to finish.
        ProcessDispatcher._wait_for_last_process()

    @classmethod
    def update_attribution(cls, attribution: "AttributionMetrics") -> None:
        """
        Updates attribution fields with new values.
        AttributionMetrics fields with `None` values will not overwrite older values.
        """
        _uniffi.glean_update_attribution(attribution)

    @classmethod
    def test_get_attribution(cls) -> "AttributionMetrics":
        """
        Test-only method for getting the current attribution metrics.
        """
        return _uniffi.glean_test_get_attribution()

    @classmethod
    def update_distribution(cls, distribution: "DistributionMetrics") -> None:
        """
        Updates distribution fields with new values.
        DistributionMetrics fields with `None` values will not overwrite older values.
        """
        _uniffi.glean_update_distribution(distribution)

    @classmethod
    def test_get_distribution(cls) -> "DistributionMetrics":
        """
        Test-only method for getting the current distribution metrics.
        """
        return _uniffi.glean_test_get_distribution()

The main Glean API.

Before any data collection can take place, the Glean SDK must be initialized from the application.

>>> Glean.initialize(
...     application_id="my-app",
...     application_version="0.0.0",
...     upload_enabled=True,
...     data_dir=Path.home() / ".glean",
... )

Static methods

def handle_client_active()

Performs the collection/cleanup operations required by becoming active.

This functions generates a baseline ping with reason active and then sets the dirty bit. This should be called whenever the consuming product becomes active (e.g. getting to foreground).

def handle_client_inactive()

Performs the collection/cleanup operations required by becoming inactive.

This functions generates a baseline and an events ping with reason inactive and then clears the dirty bit. This should be called whenever the consuming product becomes inactive (e.g. getting to background).

def initialize(application_id: str,
application_version: str,
upload_enabled: bool,
configuration: Configuration | None = None,
data_dir: pathlib._local.Path | None = None,
application_build_id: str | None = None,
log_level: int | None = None) ‑> None

Initialize the Glean SDK.

This should only be initialized once by the application, and not by libraries using the Glean SDK. A message is logged to error and no changes are made to the state if initialize is called a more than once.

Args

application_id : str

The application id to use when sending pings.

application_version : str

The version of the application sending Glean data. The meaning of this field is application-specific, but it is highly recommended to set this to something meaningful.

upload_enabled : bool

Controls whether telemetry is enabled. If disabled, all persisted metrics, events and queued pings (except first_run_date) are cleared.

configuration : Configuration

(optional) An object with global settings.

data_dir : pathlib.Path

The path to the Glean data directory.

application_build_id : str

(optional) The build identifier generated by the CI system (e.g. "1234/A").

log_level : int

(optional) The level of log messages that Glean will emit. One of the constants in the Python logging module: DEBUG, INFO, WARNING, ERROR, CRITICAL. If you need a specialized logging configuration, such as to redirecting, filtering or reformatting, you should use the Python logging module's API directly, but that will not affect logging any of Glean's networking operations which happen in a subprocess. Details in the "Debugging Python applications with the Glean SDK" chapter in the docs.

def is_initialized() ‑> bool

Returns True if the Glean SDK has been initialized.

def session_end(reason: str | None = None)

Ends a session manually.

Only has an effect when Glean is configured with SessionMode.MANUAL.

Args

reason : str

Optional application-provided string attached to the glean.session_end boundary event for downstream analysis.

def session_start()

Starts a session manually.

Only has an effect when Glean is configured with SessionMode.MANUAL. In AUTO or LIFECYCLE mode this is a no-op so automatic session state isn't corrupted.

def set_collection_enabled(enabled: bool) ‑> None

Enable or disable Glean collection and upload.

Metric collection is enabled by default.

When collection is disabled, metrics aren't recorded at all and no data is uploaded. Note: Individual pings can be enabled if they don't follow this setting. See PingType.set_enabled.

When disabling, all pending metrics, events and queued pings are cleared.

When enabling, the core Glean metrics are recreated.

Args

enabled : bool

When True, enable metric collection.

def set_experiment_active(experiment_id: str, branch: str, extra: Dict[str, str] | None = None) ‑> None

Indicate that an experiment is running. Glean will then add an experiment annotation to the environment which is sent with pings. This information is not persisted between runs.

Args

experiment_id : str

The id of the active experiment (maximum 100 bytes)

branch : str

The experiment branch (maximum 100 bytes)

extra (dict of str -> str): Optional metadata to output with the ping

def set_experiment_inactive(experiment_id: str) ‑> None

Indicate that the experiment is no longer running.

Args

experiment_id : str

The id of the experiment to deactivate.

def set_experimentation_id(experimentation_id: str)

Dynamically set the experimentation identifier, as opposed to setting it through the configuration during initialization.

Args

experimentation_id : str

The string experimentation identifier to set

def set_upload_enabled(enabled: bool) ‑> None

DEPRECATED Enable or disable Glean collection and upload.

Metric collection is enabled by default.

When uploading is disabled, metrics aren't recorded at all and no data is uploaded.

When disabling, all pending metrics, events and queued pings are cleared.

When enabling, the core Glean metrics are recreated.

DEPRECATION NOTICE: This API is deprecated. Use set_collection_enabled instead.

Args

enabled : bool

When True, enable metric collection.

def shutdown()

Shuts down Glean in an orderly fashion.

def test_get_attribution()

Test-only method for getting the current attribution metrics.

def test_get_distribution()

Test-only method for getting the current distribution metrics.

def test_get_experiment_data(experiment_id: str)

Returns the stored data for the requested active experiment, for testing purposes only.

Args

experiment_id : str

The id of the experiment to look for.

Returns

experiment_data (RecordedExperiment): The data associated with the experiment.

def test_get_experimentation_id() ‑> str

Returns the stored experimentation id, for testing purposes only.

Returns

experimentation_id (str): The experimentation id set by the client.

def test_is_experiment_active(experiment_id: str) ‑> bool

Tests whether an experiment is active, for testing purposes only.

Args

experiment_id : str

The id of the experiment to look for.

Returns

is_active (bool): If the experiement is active and reported in pings.

def update_attribution(attribution: AttributionMetrics)

Updates attribution fields with new values. AttributionMetrics fields with None values will not overwrite older values.

def update_distribution(distribution: DistributionMetrics)

Updates distribution fields with new values. DistributionMetrics fields with None values will not overwrite older values.

Instance variables

var configuration

Expand source code

def __get__(self, obj, owner):
    return self.f(owner)

Decorator for creating a property on a class (rather than an instance).

class SessionMode (*args, **kwds)

Expand source code

class SessionMode(enum.Enum):
    
    AUTO = 0
    
    LIFECYCLE = 1
    
    MANUAL = 2

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

• attribute access:

Color.RED

• value lookup:

Color(1)

• name lookup:

Color['RED']

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3

>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

Ancestors

• enum.Enum

Class variables

var AUTO

The type of the None singleton.

var LIFECYCLE

The type of the None singleton.

var MANUAL

The type of the None singleton.