File size: 1,711 Bytes
9c6594c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from typing import Optional

from wandb.proto import wandb_internal_pb2 as pb

from . import AuthenticationError, CommError, Error, UnsupportedError, UsageError

to_exception_map = {
    pb.ErrorInfo.UNKNOWN: Error,
    pb.ErrorInfo.COMMUNICATION: CommError,
    pb.ErrorInfo.AUTHENTICATION: AuthenticationError,
    pb.ErrorInfo.USAGE: UsageError,
    pb.ErrorInfo.UNSUPPORTED: UnsupportedError,
}

from_exception_map = {v: k for k, v in to_exception_map.items()}


class ProtobufErrorHandler:
    """Converts protobuf errors to exceptions and vice versa."""

    @staticmethod
    def to_exception(error: pb.ErrorInfo) -> Optional[Error]:
        """Convert a protobuf error to an exception.

        Args:
            error: The protobuf error to convert.

        Returns:
            The corresponding exception.

        """
        if not error.SerializeToString():
            return None

        if error.code in to_exception_map:
            return to_exception_map[error.code](error.message)
        return Error(error.message)

    @classmethod
    def from_exception(cls, exc: Error) -> "pb.ErrorInfo":
        """Convert an wandb error to a protobuf error message.

        Args:
            exc: The exception to convert.

        Returns:
            The corresponding protobuf error message.
        """
        if not isinstance(exc, Error):
            raise TypeError("exc must be a subclass of wandb.errors.Error")

        code = None
        for subclass in type(exc).__mro__:
            if subclass in from_exception_map:
                code = from_exception_map[subclass]  # type: ignore
                break
        return pb.ErrorInfo(code=code, message=str(exc))  # type: ignore