Skip to content

Gufo Err Example: Fail-fast

Errors may be unrecoverable. Application should be stopped as soon as possible to minimise the possible damage. Lets implement the simple fail-fast behavior. Consider the RuntimeError is fatal.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Lets see.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Type hints is the great help, so lets import the necessary types.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

All error configuration is performed via err singleton, so we need to import it first. We also need the BaseFailFast class to implement our handler.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Lets define our fail-fast handler. It must be derived from BaseFailFast.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")
Our handler accepts exception type to check as its own configuration parameter.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Do not forget always call base class constructor unless you know what you do. Otherwise, you code may be broken with future update.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Lets store our configuration as exc_type argument.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

must_die function is the key function for fail-fast handler. It accepts the result of sys.exc_info() function. First parameter is the exception type. Second is the exception value. Last is the frame information. Fail-fast handlers return boolean value. True should be returned if the error is unrecoverable, False - otherwise.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

The logic is simple. If the exception type is matched with configured one - we must fail.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

err.setup() function must be called to initialize and confugure the error protection. None, we pass to the fail_fast argument a list of configured fail-fast handler instances, not a classes. fail_fast_code parameter is optional and sets the exit code on fail-fast termination. Default code is 1, but we set it to 5 for our example. See Err.setup() for details.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Lets define the function which will intentionally fail.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Lets wrap our error domain.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

And call our faulty function.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")

Run all the error processing machinery.

failfast.py
 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
from types import TracebackType
from typing import Type

from gufo.err import BaseFailFast, err


class FailOnType(BaseFailFast):
    def __init__(self, exc_type) -> None:
        super().__init__()
        self.exc_type = exc_type

    def must_die(
        self, t: Type[BaseException], v: BaseException, tb: TracebackType
    ) -> bool:
        return t == self.exc_type


err.setup(fail_fast=[FailOnType(RuntimeError)], fail_fast_code=5)


def fail():
    msg = "failing"
    raise RuntimeError(msg)


try:
    fail()
except Exception:
    err.process()
print("Stopping")
Here we print the debug message. If our fail-fast code works correctly, we will not see this message.

Running

Run example as:

$ python3 examples/failfast.py

And got the empty output. Let check our error code:

$ echo $?
5

Note, we didn't saw "Stopping" message and our process returns error code 5. All just we configured.