Module `glean.metrics.object`

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/.


from typing import Any, Generic, TypeVar, Optional

import json
from json import JSONEncoder
from dataclasses import asdict

from .._uniffi import CommonMetricData
from .._uniffi import ObjectMetric
from ..testing import ErrorType


class EncodeObject(JSONEncoder):
    def default(self, o):
        # Remove `None` items for objects that are `ObjectSerialize`
        if isinstance(o, ObjectSerialize):
            d = {k: v for k, v in asdict(o).items() if v is not None}
            return d

        return super().default(o)


class ObjectSerialize:
    """
    A class that can be converted into key-value pairs of event extras.
    This will be automatically implemented for event properties of an [ObjectMetricType].
    """

    def into_serialized_object(self) -> str:
        """
        Convert the event extras into a key-value dict:
        """
        return json.dumps(self, cls=EncodeObject)


T = TypeVar("T", bound=ObjectSerialize)


class ObjectMetricType(Generic[T]):
    """
    This implements the developer facing API for recording events.

    Instances of this class type are automatically generated by
    `glean.load_metrics`, allowing developers to record values that were
    previously registered in the metrics.yaml file.

    The event API only exposes the `ObjectMetricType.set` method, which
    takes care of validating the input data and making sure that limits are
    enforced.
    """

    def __init__(
        self,
        common_metric_data: CommonMetricData,
        ty: Any,
    ):
        self._inner = ObjectMetric(common_metric_data)
        self._objty = ty

    def set(self, obj: T) -> None:
        """
        Set the object.
        Args:
            extra: optional. The extra keys and values for this event.
                   The maximum length for values is 100.
        """

        if isinstance(obj, self._objty):
            inner_obj = obj.into_serialized_object()
            self._inner.set_string(inner_obj)
        else:
            self._inner.record_schema_error()

    def test_get_value(self, ping_name: Optional[str] = None) -> Optional[dict]:
        """
        Returns the stored value for testing purposes only.

        Args:
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            value (list of RecordedEventData): value of the stored events.
        """
        data = self._inner.test_get_value(ping_name)
        if data:
            return json.loads(data)

        return None

    def test_get_num_recorded_errors(self, error_type: ErrorType) -> int:
        """
        Returns the number of errors recorded for the given metric.

        Args:
            error_type (ErrorType): The type of error recorded.
            ping_name (str): (default: first value in send_in_pings) The name
                of the ping to retrieve the metric for.

        Returns:
            num_errors (int): The number of errors recorded for the metric for
                the given error type.
        """
        return self._inner.test_get_num_recorded_errors(error_type)


__all__ = ["ObjectMetric", "ObjectSerialize"]

Classes

class ObjectMetric (meta: "'CommonMetricData'")

Expand source code

class ObjectMetric:
    _pointer: ctypes.c_void_p
    def __init__(self, meta: "CommonMetricData"):
        _UniffiConverterTypeCommonMetricData.check_lower(meta)
        
        self._pointer = _rust_call(_UniffiLib.uniffi_glean_core_fn_constructor_objectmetric_new,
        _UniffiConverterTypeCommonMetricData.lower(meta))

    def __del__(self):
        # In case of partial initialization of instances.
        pointer = getattr(self, "_pointer", None)
        if pointer is not None:
            _rust_call(_UniffiLib.uniffi_glean_core_fn_free_objectmetric, pointer)

    def _uniffi_clone_pointer(self):
        return _rust_call(_UniffiLib.uniffi_glean_core_fn_clone_objectmetric, self._pointer)

    # Used by alternative constructors or any methods which return this type.
    @classmethod
    def _make_instance_(cls, pointer):
        # Lightly yucky way to bypass the usual __init__ logic
        # and just create a new instance with the required pointer.
        inst = cls.__new__(cls)
        inst._pointer = pointer
        return inst


    def record_schema_error(self, ) -> None:
        _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_record_schema_error,self._uniffi_clone_pointer(),)






    def set_string(self, object: "str") -> None:
        _UniffiConverterString.check_lower(object)
        
        _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_set_string,self._uniffi_clone_pointer(),
        _UniffiConverterString.lower(object))






    def test_get_num_recorded_errors(self, error: "ErrorType") -> "int":
        _UniffiConverterTypeErrorType.check_lower(error)
        
        return _UniffiConverterInt32.lift(
            _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_test_get_num_recorded_errors,self._uniffi_clone_pointer(),
        _UniffiConverterTypeErrorType.lower(error))
        )





    def test_get_value(self, ping_name: "typing.Union[object, typing.Optional[str]]" = _DEFAULT) -> "typing.Optional[JsonValue]":
        if ping_name is _DEFAULT:
            ping_name = None
        _UniffiConverterOptionalString.check_lower(ping_name)
        
        return _UniffiConverterOptionalTypeJsonValue.lift(
            _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_test_get_value,self._uniffi_clone_pointer(),
        _UniffiConverterOptionalString.lower(ping_name))
        )

Methods

def record_schema_error(self) ‑> None

Expand source code

def record_schema_error(self, ) -> None:
    _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_record_schema_error,self._uniffi_clone_pointer(),)

def set_string(self, object: "'str'") ‑> None

Expand source code

def set_string(self, object: "str") -> None:
    _UniffiConverterString.check_lower(object)
    
    _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_set_string,self._uniffi_clone_pointer(),
    _UniffiConverterString.lower(object))

def test_get_num_recorded_errors(self, error: "'ErrorType'") ‑> int

Expand source code

def test_get_num_recorded_errors(self, error: "ErrorType") -> "int":
    _UniffiConverterTypeErrorType.check_lower(error)
    
    return _UniffiConverterInt32.lift(
        _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_test_get_num_recorded_errors,self._uniffi_clone_pointer(),
    _UniffiConverterTypeErrorType.lower(error))
    )

def test_get_value(self, ping_name: "'typing.Union[object, typing.Optional[str]]'" = <object object>) ‑> Optional[str]

Expand source code

def test_get_value(self, ping_name: "typing.Union[object, typing.Optional[str]]" = _DEFAULT) -> "typing.Optional[JsonValue]":
    if ping_name is _DEFAULT:
        ping_name = None
    _UniffiConverterOptionalString.check_lower(ping_name)
    
    return _UniffiConverterOptionalTypeJsonValue.lift(
        _rust_call(_UniffiLib.uniffi_glean_core_fn_method_objectmetric_test_get_value,self._uniffi_clone_pointer(),
    _UniffiConverterOptionalString.lower(ping_name))
    )

class ObjectSerialize

A class that can be converted into key-value pairs of event extras. This will be automatically implemented for event properties of an [ObjectMetricType].

Expand source code

class ObjectSerialize:
    """
    A class that can be converted into key-value pairs of event extras.
    This will be automatically implemented for event properties of an [ObjectMetricType].
    """

    def into_serialized_object(self) -> str:
        """
        Convert the event extras into a key-value dict:
        """
        return json.dumps(self, cls=EncodeObject)

Methods

def into_serialized_object(self) ‑> str

Convert the event extras into a key-value dict:

Expand source code

def into_serialized_object(self) -> str:
    """
    Convert the event extras into a key-value dict:
    """
    return json.dumps(self, cls=EncodeObject)