cozeloop-python/cozeloop/internal/trace/noop_span.py at a41add3f5659870d98d9428febd225dc594e4c4e · coze-dev/cozeloop-python · GitHub

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
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
# Copyright (c) 2025 Bytedance Ltd. and/or its affiliates
# SPDX-License-Identifier: MIT

import types
from abc import ABC
from datetime import datetime
from typing import Dict, Any

from cozeloop.entities.prompt import Prompt
from cozeloop.span import Span
from cozeloop.spec.tracespec import Runtime


class NoopSpan(Span, ABC):
    @property
    def get_span_id(self) -> str:
        return ''

    @property
    def get_trace_id(self) -> str:
        return ''

    @property
    def get_baggage(self) -> Dict[str, str]:
        return {}

    @property
    def get_start_time(self) -> datetime:
        return datetime.now()

    def set_tags(self, tag_kvs: Dict[str, Any]) -> None:
        pass

    def set_baggage(self, baggage_items: Dict[str, str]) -> None:
        pass

    def discard(self) -> None:
        pass

    def finish(self) -> None:
        pass

    @property
    def baggage(self) -> Dict[str, str]:
        return {}

    @property
    def trace_id(self) -> str:
        return ""

    @property
    def span_id(self) -> str:
        return ""

    @property
    def start_time(self) -> int:
        return 0

    def to_header(self) -> Dict[str, str]:
        return {}

    def set_input(self, input: Any) -> None:
        pass

    def set_output(self, output: Any) -> None:
        pass

    def set_error(self, err: Exception) -> None:
        pass

    def set_status_code(self, code: int) -> None:
        pass

    def set_user_id(self, user_id: str) -> None:
        pass

    def set_user_id_baggage(self, user_id: str) -> None:
        pass

    def set_message_id(self, message_id: str) -> None:
        pass

    def set_message_id_baggage(self, message_id: str) -> None:
        pass

    def set_thread_id(self, thread_id: str) -> None:
        pass

    def set_thread_id_baggage(self, thread_id: str) -> None:
        pass

    def set_prompt(self, prompt: Prompt) -> None:
        pass

    def set_model_provider(self, model_provider: str) -> None:
        pass

    def set_model_name(self, model_name: str) -> None:
        pass

    def set_model_call_options(self, call_options: Any) -> None:
        pass

    def set_input_tokens(self, input_tokens: int) -> None:
        pass

    def set_output_tokens(self, output_tokens: int) -> None:
        pass

    def set_start_time_first_resp(self, start_time_first_resp: int) -> None:
        pass

    def set_runtime(self, runtime: Runtime) -> None:
        pass

    def set_service_name(self, service_name: str) -> None:
        pass

    def set_log_id(self, log_id: str) -> None:
        pass

    def set_system_tags(self, system_tags: Dict[str, Any]) -> None:
        pass

    def set_deployment_env(self, deployment_env: str) -> None:
        pass

    def set_finish_time(self, finish_time: datetime) -> None:
        pass

    def __enter__(self):
        return self

    def __exit__(self, exc, value, tb):
        self.finish()