CoCalc -- logging.py

GitHub Repository: wiseplat/python-code
Path: blob/master/ invest-robot-contest_TinkoffBotTwitch-main/venv/lib/python3.8/site-packages/tinkoff/invest/logging.py
⁷⁸²⁶ views
1
import logging
2
from collections import namedtuple
3
from contextvars import ContextVar
4
from typing import Any, Optional
5

6
from .constants import (
7
    MESSAGE,
8
    X_RATELIMIT_LIMIT,
9
    X_RATELIMIT_REMAINING,
10
    X_RATELIMIT_RESET,
11
    X_TRACKING_ID,
12
)
13

14
__all__ = (
15
    "get_current_tracking_id",
16
    "get_tracking_id_from_call",
17
    "get_tracking_id_from_coro",
18
    "get_metadata_from_call",
19
    "get_metadata_from_aio_error",
20
    "log_request",
21
    "log_error",
22
)
23

24
logger = logging.getLogger(__name__)
25

26
_TRACKING_ID: ContextVar[Optional[str]] = ContextVar("tracking_id", default=None)
27
Metadata = namedtuple(
28
    "Metadata",
29
    (
30
        "tracking_id",
31
        "ratelimit_limit",
32
        "ratelimit_remaining",
33
        "ratelimit_reset",
34
        "message",
35
    ),
36
)
37

38

39
def get_current_tracking_id() -> Optional[str]:
40
    return _TRACKING_ID.get() or None
41

42

43
def log_request(tracking_id: Optional[str], name: str) -> None:
44
    _TRACKING_ID.set(tracking_id)
45
    logger.info("%s %s", tracking_id, name)
46

47

48
def log_error(tracking_id: Optional[str], name: str, text: str) -> None:
49
    _TRACKING_ID.set(tracking_id)
50
    logger.error("%s %s %s", tracking_id, name, text)
51

52

53
def get_tracking_id_from_call(call: Any) -> Optional[str]:
54
    metadata = call.initial_metadata() or call.trailing_metadata()
55
    for item in metadata:
56
        if item.key == X_TRACKING_ID:
57
            return item.value
58
    return None
59

60

61
async def get_tracking_id_from_coro(coro: Any) -> Optional[str]:
62
    metadata = await coro.initial_metadata() or await coro.trailing_metadata()
63
    for key, value in metadata:
64
        if key == X_TRACKING_ID:
65
            return value
66
    return None
67

68

69
def get_metadata_from_call(call: Any) -> Optional[Metadata]:
70
    metadata = call.initial_metadata() or call.trailing_metadata()
71
    tracking_id = None
72
    ratelimit_limit = None
73
    ratelimit_remaining = None
74
    ratelimit_reset = None
75
    message = None
76
    for item in metadata:
77
        if item.key == X_TRACKING_ID:
78
            tracking_id = item.value
79
        elif item.key == X_RATELIMIT_LIMIT:
80
            ratelimit_limit = item.value
81
        elif item.key == X_RATELIMIT_REMAINING:
82
            ratelimit_remaining = int(item.value)
83
        elif item.key == X_RATELIMIT_RESET:
84
            ratelimit_reset = int(item.value)
85
        elif item.key == MESSAGE:
86
            message = item.value
87
    if not any(
88
        (tracking_id, ratelimit_limit, ratelimit_remaining, ratelimit_reset, message)
89
    ):
90
        return None
91
    return Metadata(
92
        tracking_id, ratelimit_limit, ratelimit_remaining, ratelimit_reset, message
93
    )
94

95

96
def get_metadata_from_aio_error(err: Any) -> Optional[Metadata]:
97
    metadata = err.initial_metadata() or err.trailing_metadata()
98
    tracking_id = None
99
    ratelimit_limit = None
100
    ratelimit_remaining = None
101
    ratelimit_reset = None
102
    message = None
103
    for key, value in metadata:
104
        if key == X_TRACKING_ID:
105
            tracking_id = value
106
        elif key == X_RATELIMIT_LIMIT:
107
            ratelimit_limit = value
108
        elif key == X_RATELIMIT_REMAINING:
109
            ratelimit_remaining = int(value)
110
        elif key == X_RATELIMIT_RESET:
111
            ratelimit_reset = int(value)
112
        elif key == MESSAGE:
113
            message = value
114
    if not any(
115
        (tracking_id, ratelimit_limit, ratelimit_remaining, ratelimit_reset, message)
116
    ):
117
        return None
118
    return Metadata(
119
        tracking_id, ratelimit_limit, ratelimit_remaining, ratelimit_reset, message
120
    )
121

122
Product

Resources

Company
