Skip to content

LogSentry

dotflow.providers.log_sentry.LogSentry

Bases: Log

Import

You can import the LogSentry class with:

from dotflow.providers import LogSentry
Example

class dotflow.providers.log_sentry.LogSentry

from dotflow import Config, DotFlow
from dotflow.providers import LogSentry

config = Config(
    log=LogSentry(
        dsn="https://xxx@sentry.io/123",
        environment="production",
    ),
)

workflow = DotFlow(config=config)

Parameters:

Name Type Description Default
dsn str

Sentry DSN for the project.

 required
environment str

Environment tag sent to Sentry. Defaults to None.

 None
traces_sample_rate float

Sample rate for performance traces (0.0 to 1.0). Defaults to 0.0.

 0.0
Source code in dotflow/providers/log_sentry.py 
 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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class LogSentry(Log):
    """
    Import:
        You can import the **LogSentry** class with:

            from dotflow.providers import LogSentry

    Example:
        `class` dotflow.providers.log_sentry.LogSentry

            from dotflow import Config, DotFlow
            from dotflow.providers import LogSentry

            config = Config(
                log=LogSentry(
                    dsn="https://xxx@sentry.io/123",
                    environment="production",
                ),
            )

            workflow = DotFlow(config=config)

    Args:
        dsn (str): Sentry DSN for the project.
        environment (str): Environment tag sent to Sentry.
            Defaults to None.
        traces_sample_rate (float): Sample rate for performance traces (0.0 to 1.0).
            Defaults to 0.0.
    """

    def __init__(
        self,
        dsn: str,
        environment: str | None = None,
        traces_sample_rate: float = 0.0,
    ) -> None:
        try:
            import sentry_sdk
        except ImportError as err:
            raise ModuleNotFound(
                module="sentry-sdk", library="dotflow[sentry]"
            ) from err

        sentry_sdk.init(
            dsn=dsn,
            environment=environment,
            traces_sample_rate=traces_sample_rate,
        )
        self._sentry = sentry_sdk

    def info(self, **kwargs) -> None:
        task = kwargs.get("task")
        if task:
            self._sentry.add_breadcrumb(
                category="dotflow",
                message=f"task:{task.task_id} {task.status}",
                level="info",
            )

    def error(self, **kwargs) -> None:
        task = kwargs.get("task")
        if not task or not task.errors:
            return

        last_error = task.errors[-1]
        self._sentry.capture_message(
            f"Task {task.task_id} failed: {last_error.message}",
            level="error",
            extras={
                "workflow_id": str(task.workflow_id),
                "task_id": str(task.task_id),
                "exception": str(last_error.exception),
                "attempt": last_error.attempt,
                "traceback": last_error.traceback,
            },
        )

    def warning(self, **kwargs) -> None:
        task = kwargs.get("task")
        if task:
            self._sentry.add_breadcrumb(
                category="dotflow",
                message=f"task:{task.task_id} {task.status}",
                level="warning",
            )

    def debug(self, **kwargs) -> None:
        pass

__init__(dsn, environment=None, traces_sample_rate=0.0)

Source code in dotflow/providers/log_sentry.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    dsn: str,
    environment: str | None = None,
    traces_sample_rate: float = 0.0,
) -> None:
    try:
        import sentry_sdk
    except ImportError as err:
        raise ModuleNotFound(
            module="sentry-sdk", library="dotflow[sentry]"
        ) from err

    sentry_sdk.init(
        dsn=dsn,
        environment=environment,
        traces_sample_rate=traces_sample_rate,
    )
    self._sentry = sentry_sdk

debug(**kwargs)

Source code in dotflow/providers/log_sentry.py 
95
96
def debug(self, **kwargs) -> None:
    pass

error(**kwargs)

Source code in dotflow/providers/log_sentry.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def error(self, **kwargs) -> None:
    task = kwargs.get("task")
    if not task or not task.errors:
        return

    last_error = task.errors[-1]
    self._sentry.capture_message(
        f"Task {task.task_id} failed: {last_error.message}",
        level="error",
        extras={
            "workflow_id": str(task.workflow_id),
            "task_id": str(task.task_id),
            "exception": str(last_error.exception),
            "attempt": last_error.attempt,
            "traceback": last_error.traceback,
        },
    )

info(**kwargs)

Source code in dotflow/providers/log_sentry.py 
59
60
61
62
63
64
65
66
def info(self, **kwargs) -> None:
    task = kwargs.get("task")
    if task:
        self._sentry.add_breadcrumb(
            category="dotflow",
            message=f"task:{task.task_id} {task.status}",
            level="info",
        )

warning(**kwargs)

Source code in dotflow/providers/log_sentry.py 
86
87
88
89
90
91
92
93
def warning(self, **kwargs) -> None:
    task = kwargs.get("task")
    if task:
        self._sentry.add_breadcrumb(
            category="dotflow",
            message=f"task:{task.task_id} {task.status}",
            level="warning",
        )