26751910360

Committed 01 Jun 2026 11:24AM UTC coverage: 52.046% (-40.7%) from 92.792%

Build # 26751910360

Build Type

Pull #23395

github

Committed by

web-flow

Commit Message

Merge 41608d741 into c8127c1f4

Pull Request Pull Request #23395: Bump the gha-deps group with 8 updates

Coverage Stats

31994 of 61473 relevant lines covered (52.05%)

1.04 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/python/pants/backend/observability/opentelemetry/processor.py

# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import datetime
import enum
from dataclasses import dataclass
from typing import Any, Mapping, Protocol

from pants.util.frozendict import FrozenDict


class Level(enum.Enum):
    DEBUG = "DEBUG"
    ERROR = "ERROR"
    INFO = "INFO"
    TRACE = "TRACE"
    WARN = "WARN"


@dataclass(frozen=True)
class IncompleteWorkunit:
    """An incomplete workunit which only knows its start time."""

    name: str
    span_id: str
    parent_ids: tuple[str, ...]
    level: Level
    description: str | None
    start_time: datetime.datetime

    @property
    def primary_parent_id(self) -> str | None:
        if len(self.parent_ids) > 0:
            return self.parent_ids[0]
        return None


@dataclass(frozen=True)
class Workunit(IncompleteWorkunit):
    """The final workunit which knows when it completed as well."""

    end_time: datetime.datetime
    metadata: FrozenDict[str, Any]


class ProcessorContext(Protocol):
    def get_metrics(self) -> Mapping[str, int]: ...


class Processor(Protocol):
    """Protocol for emitter implementations."""

    def initialize(self) -> None: ...

    def start_workunit(
        self, workunit: IncompleteWorkunit, *, context: ProcessorContext
    ) -> None: ...

    def complete_workunit(self, workunit: Workunit, *, context: ProcessorContext) -> None: ...

    def finish(
        self, timeout: datetime.timedelta | None = None, *, context: ProcessorContext
    ) -> None: ...

1	# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	×
5
6	import datetime	×
7	import enum	×
8	from dataclasses import dataclass	×
9	from typing import Any, Mapping, Protocol	×
10
11	from pants.util.frozendict import FrozenDict	×
12
13
14	class Level(enum.Enum):	×
15	DEBUG = "DEBUG"	×
16	ERROR = "ERROR"	×
17	INFO = "INFO"	×
18	TRACE = "TRACE"	×
19	WARN = "WARN"	×
20
21
22	@dataclass(frozen=True)	×
23	class IncompleteWorkunit:	×
24	"""An incomplete workunit which only knows its start time."""
25
26	name: str	×
27	span_id: str	×
28	parent_ids: tuple[str, ...]	×
29	level: Level	×
30	description: str \| None	×
31	start_time: datetime.datetime	×
32
33	@property	×
34	def primary_parent_id(self) -> str \| None:	×
35	if len(self.parent_ids) > 0:	×
36	return self.parent_ids[0]	×
37	return None	×
38
39
40	@dataclass(frozen=True)	×
41	class Workunit(IncompleteWorkunit):	×
42	"""The final workunit which knows when it completed as well."""
43
44	end_time: datetime.datetime	×
45	metadata: FrozenDict[str, Any]	×
46
47
48	class ProcessorContext(Protocol):	×
49	def get_metrics(self) -> Mapping[str, int]: ...
50
51
52	class Processor(Protocol):	×
53	"""Protocol for emitter implementations."""
54
55	def initialize(self) -> None: ...
56
57	def start_workunit(
58	self, workunit: IncompleteWorkunit, *, context: ProcessorContext
59	) -> None: ...
60
61	def complete_workunit(self, workunit: Workunit, *, context: ProcessorContext) -> None: ...
62
63	def finish(
64	self, timeout: datetime.timedelta \| None = None, *, context: ProcessorContext
65	) -> None: ...

pantsbuild / pants / 26751910360

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous