Package glean
Top-level package for Glean SDK.
Expand source code
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
"""Top-level package for Glean SDK."""
import warnings
from pkg_resources import get_distribution, DistributionNotFound
from semver import VersionInfo # type: ignore
import glean_parser # type: ignore
from .glean import Glean
from .config import Configuration
from ._loader import load_metrics, load_pings
__version__: str = "unknown"
try:
__version__ = str(get_distribution("glean-sdk").version)
except DistributionNotFound: # pragma: no cover
pass
__author__ = "The Glean Team"
__email__ = "glean-team@mozilla.com"
GLEAN_PARSER_VERSION = "10.0.0"
parser_version = VersionInfo.parse(GLEAN_PARSER_VERSION)
parser_version_next_major = parser_version.bump_major()
current_parser = VersionInfo.parse(glean_parser.__version__)
if current_parser < parser_version or current_parser >= parser_version_next_major:
warnings.warn(
f"glean_sdk expected glean_parser ~= v{GLEAN_PARSER_VERSION}, "
f"found v{glean_parser.__version__}",
Warning,
stacklevel=1,
)
__all__ = [
"__author__",
"__email__",
"__version__",
"Glean",
"Configuration",
"load_metrics",
"load_pings",
]
# Tell pdoc3 to ignore the libglean_ffi.so, which is a Rust shared library, not
# a Python extension module.
__pdoc__ = {"libglean_ffi": False}
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: Union[str, pathlib.Path, List[Union[str, pathlib.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")
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
def load_pings(filepath: Union[str, pathlib.Path, List[Union[str, pathlib.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()
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
Classes
class Configuration (server_endpoint: Optional[str] = None, channel: Optional[str] = None, max_events: int = 500, ping_uploader: Optional[BaseUploader] = None, allow_multiprocessing: bool = True, enable_event_timestamps: bool = False, experimentation_id: Optional[str] = None)
-
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
- (Experimental) Whether to add a
wallclock timestamp to all events. Default:
false
.
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 = False, experimentation_id: Optional[str] = None, ): """ 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): (Experimental) Whether to add a wallclock timestamp to all events. Default: `false`. """ 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 @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: """(Experimental) 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 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
Instance variables
var channel : Optional[str]
-
The release channel the application is on, if known.
Expand source code
@property def channel(self) -> Optional[str]: """The release channel the application is on, if known.""" return self._channel
var enable_event_timestamps : bool
-
(Experimental) Whether to add a wallclock timestamp to all events.
Expand source code
@property def enable_event_timestamps(self) -> bool: """(Experimental) Whether to add a wallclock timestamp to all events.""" return self._enable_event_timestamps
var experimentation_id : Optional[str]
-
An experimentation id that will be sent in all pings
Expand source code
@property def experimentation_id(self) -> Optional[str]: """An experimentation id that will be sent in all pings""" return self._experimentation_id
var max_events : int
-
The number of events to store before force-sending.
Expand source code
@property def max_events(self) -> int: """The number of events to store before force-sending.""" return self._max_events
var ping_uploader : BaseUploader
-
The ping uploader implementation.
Expand source code
@property def ping_uploader(self) -> net.BaseUploader: """The ping uploader implementation.""" return self._ping_uploader
var server_endpoint : str
-
The server pings are sent to.
Expand source code
@property def server_endpoint(self) -> str: """The server pings are sent to.""" return self._server_endpoint
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", ... )
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="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, ) _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: """ 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. 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_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 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 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()
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).Expand source code
@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()
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).Expand source code
@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()
def initialize(application_id: str, application_version: str, upload_enabled: bool, configuration: Optional[Configuration] = None, data_dir: Optional[pathlib.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
: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 Pythonlogging
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.
Expand source code
@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="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, ) _uniffi.glean_initialize(cfg, client_info, callbacks) cls._initialized = True
def is_initialized() ‑> bool
-
Returns True if the Glean SDK has been initialized.
Expand source code
@classmethod def is_initialized(cls) -> bool: """ Returns True if the Glean SDK has been initialized. """ return cls._initialized
def set_experiment_active(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
Expand source code
@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)
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.
Expand source code
@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)
def set_upload_enabled(enabled: bool) ‑> None
-
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.
Args
enabled
:bool
- When True, enable metric collection.
Expand source code
@classmethod def set_upload_enabled(cls, enabled: bool) -> None: """ 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. 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)
def shutdown()
-
Shuts down Glean in an orderly fashion.
Expand source code
@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()
def test_get_experiment_data(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.
Expand source code
@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")
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.
Expand source code
@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")
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.
Expand source code
@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
Instance variables
var configuration
-
Decorator for creating a property on a class (rather than an instance).
Expand source code
def __get__(self, obj, owner): return self.f(owner)