Module glean.net.ping_uploader

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


import sys
from typing import Dict, Union

from .._uniffi import UploadResult


if sys.version_info >= (3, 8):
    from typing import Protocol
else:
    Protocol = object


class PingUploader(Protocol):
    def upload(
        self, url: str, data: bytes, headers: Dict[str, str]
    ) -> Union[
        UploadResult,
        UploadResult.UNRECOVERABLE_FAILURE,
        UploadResult.RECOVERABLE_FAILURE,
        UploadResult.HTTP_STATUS,
    ]:
        """
        Upload a ping to a server.

        Args:
            url (str): The URL path to upload the data to.
            data (bytes): The serialized data to send.
            headers (dict of (str, str)): Dictionary of header entries.

        Returns:
            result (UploadResult): the status code of the upload response.
        """
        pass


__all__ = [
    "PingUploader",
    "UploadResult",
]

Classes

class PingUploader (*args, **kwargs)

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing), for example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto(Protocol[T]):
    def meth(self) -> T:
        ...
Expand source code
class PingUploader(Protocol):
    def upload(
        self, url: str, data: bytes, headers: Dict[str, str]
    ) -> Union[
        UploadResult,
        UploadResult.UNRECOVERABLE_FAILURE,
        UploadResult.RECOVERABLE_FAILURE,
        UploadResult.HTTP_STATUS,
    ]:
        """
        Upload a ping to a server.

        Args:
            url (str): The URL path to upload the data to.
            data (bytes): The serialized data to send.
            headers (dict of (str, str)): Dictionary of header entries.

        Returns:
            result (UploadResult): the status code of the upload response.
        """
        pass

Ancestors

  • typing.Protocol
  • typing.Generic

Subclasses

Methods

def upload(self, url: str, data: bytes, headers: Dict[str, str]) ‑> Union[glean._uniffi.glean.UploadResult, glean._uniffi.glean.UploadResult.UNRECOVERABLE_FAILURE, glean._uniffi.glean.UploadResult.RECOVERABLE_FAILURE, glean._uniffi.glean.UploadResult.HTTP_STATUS]

Upload a ping to a server.

Args

url : str
The URL path to upload the data to.
data : bytes
The serialized data to send.

headers (dict of (str, str)): Dictionary of header entries.

Returns

result (UploadResult): the status code of the upload response.

Expand source code
def upload(
    self, url: str, data: bytes, headers: Dict[str, str]
) -> Union[
    UploadResult,
    UploadResult.UNRECOVERABLE_FAILURE,
    UploadResult.RECOVERABLE_FAILURE,
    UploadResult.HTTP_STATUS,
]:
    """
    Upload a ping to a server.

    Args:
        url (str): The URL path to upload the data to.
        data (bytes): The serialized data to send.
        headers (dict of (str, str)): Dictionary of header entries.

    Returns:
        result (UploadResult): the status code of the upload response.
    """
    pass
class UploadResult
Expand source code
class UploadResult:
    def __init__(self):
        raise RuntimeError("UploadResult cannot be instantiated directly")

    # Each enum variant is a nested class of the enum itself.
    class RECOVERABLE_FAILURE:
        unused: "int"

        @typing.no_type_check
        def __init__(self,unused: "int"):
            self.unused = unused

        def __str__(self):
            return "UploadResult.RECOVERABLE_FAILURE(unused={})".format(self.unused)

        def __eq__(self, other):
            if not other.is_recoverable_failure():
                return False
            if self.unused != other.unused:
                return False
            return True
    
    class UNRECOVERABLE_FAILURE:
        unused: "int"

        @typing.no_type_check
        def __init__(self,unused: "int"):
            self.unused = unused

        def __str__(self):
            return "UploadResult.UNRECOVERABLE_FAILURE(unused={})".format(self.unused)

        def __eq__(self, other):
            if not other.is_unrecoverable_failure():
                return False
            if self.unused != other.unused:
                return False
            return True
    
    class HTTP_STATUS:
        code: "int"

        @typing.no_type_check
        def __init__(self,code: "int"):
            self.code = code

        def __str__(self):
            return "UploadResult.HTTP_STATUS(code={})".format(self.code)

        def __eq__(self, other):
            if not other.is_http_status():
                return False
            if self.code != other.code:
                return False
            return True
    
    class DONE:
        unused: "int"

        @typing.no_type_check
        def __init__(self,unused: "int"):
            self.unused = unused

        def __str__(self):
            return "UploadResult.DONE(unused={})".format(self.unused)

        def __eq__(self, other):
            if not other.is_done():
                return False
            if self.unused != other.unused:
                return False
            return True
    
    

    # For each variant, we have an `is_NAME` method for easily checking
    # whether an instance is that variant.
    def is_recoverable_failure(self) -> bool:
        return isinstance(self, UploadResult.RECOVERABLE_FAILURE)
    def is_unrecoverable_failure(self) -> bool:
        return isinstance(self, UploadResult.UNRECOVERABLE_FAILURE)
    def is_http_status(self) -> bool:
        return isinstance(self, UploadResult.HTTP_STATUS)
    def is_done(self) -> bool:
        return isinstance(self, UploadResult.DONE)

Subclasses

  • glean._uniffi.glean.UploadResult.DONE
  • glean._uniffi.glean.UploadResult.HTTP_STATUS
  • glean._uniffi.glean.UploadResult.RECOVERABLE_FAILURE
  • glean._uniffi.glean.UploadResult.UNRECOVERABLE_FAILURE

Class variables

var DONE
var HTTP_STATUS
var RECOVERABLE_FAILURE
var UNRECOVERABLE_FAILURE

Methods

def is_done(self) ‑> bool
Expand source code
def is_done(self) -> bool:
    return isinstance(self, UploadResult.DONE)
def is_http_status(self) ‑> bool
Expand source code
def is_http_status(self) -> bool:
    return isinstance(self, UploadResult.HTTP_STATUS)
def is_recoverable_failure(self) ‑> bool
Expand source code
def is_recoverable_failure(self) -> bool:
    return isinstance(self, UploadResult.RECOVERABLE_FAILURE)
def is_unrecoverable_failure(self) ‑> bool
Expand source code
def is_unrecoverable_failure(self) -> bool:
    return isinstance(self, UploadResult.UNRECOVERABLE_FAILURE)