Skip to content

time_parser.py

This module provides interfaces to parse and convert timestamp.

__all__ = ['TimeParser'] module-attribute

InvalidTimeFormatException

Bases: Exception

Exception for invalid time format.

Source code in solnlib/time_parser.py 
31
32
33
34
class InvalidTimeFormatException(Exception):
    """Exception for invalid time format."""

    pass

TimeParser

Datetime parser.

Use splunkd rest to parse datetime.

Examples:

>>> from solnlib import time_parser
>>> tp = time_parser.TimeParser(session_key)
>>> tp.to_seconds('2011-07-06T21:54:23.000-07:00')
>>> tp.to_utc('2011-07-06T21:54:23.000-07:00')
>>> tp.to_local('2011-07-06T21:54:23.000-07:00')
Source code in solnlib/time_parser.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class TimeParser:
    """Datetime parser.

    Use splunkd rest to parse datetime.

    Examples:
       >>> from solnlib import time_parser
       >>> tp = time_parser.TimeParser(session_key)
       >>> tp.to_seconds('2011-07-06T21:54:23.000-07:00')
       >>> tp.to_utc('2011-07-06T21:54:23.000-07:00')
       >>> tp.to_local('2011-07-06T21:54:23.000-07:00')
    """

    URL = "/services/search/timeparser"

    def __init__(
        self,
        session_key: str,
        scheme: str = None,
        host: str = None,
        port: int = None,
        **context: Any,
    ):
        """Initializes TimeParser.

        Arguments:
            session_key: Splunk access token.
            scheme: (optional) The access scheme, default is None.
            host: (optional) The host name, default is None.
            port: (optional) The port number, default is None.
            context: Other configurations for Splunk rest client.

        Raises:
            ValueError: if scheme, host or port are invalid.
        """
        self._rest_client = rest_client.SplunkRestClient(
            session_key, "-", scheme=scheme, host=host, port=port, **context
        )

    @retry(exceptions=[binding.HTTPError])
    def to_seconds(self, time_str: str) -> float:
        """Parse `time_str` and convert to seconds since epoch.

        Arguments:
            time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

        Raises:
            binding.HTTPError: rest client returns an exception (everything
                else than 400 code).
            InvalidTimeFormatException: when time format is invalid (rest
                client returns 400 code).

        Returns:
            Seconds since epoch.
        """

        try:
            response = self._rest_client.get(
                self.URL, output_mode="json", time=time_str, output_time_format="%s"
            ).body.read()
        except binding.HTTPError as e:
            if e.status != 400:
                raise

            raise InvalidTimeFormatException(f"Invalid time format: {time_str}.")

        seconds = json.loads(response)[time_str]
        return float(seconds)

    def to_utc(self, time_str: str) -> datetime.datetime:
        """Parse `time_str` and convert to UTC timestamp.

        Arguments:
            time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

        Raises:
            binding.HTTPError: rest client returns an exception (everything
                else than 400 code).
            InvalidTimeFormatException: when time format is invalid (rest
                client returns 400 code).

        Returns:
            UTC timestamp.
        """

        return datetime.datetime.utcfromtimestamp(self.to_seconds(time_str))

    @retry(exceptions=[binding.HTTPError])
    def to_local(self, time_str: str) -> str:
        """Parse `time_str` and convert to local timestamp.

        Arguments:
            time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

        Raises:
            binding.HTTPError: rest client returns an exception (everything
                else than 400 code).
            InvalidTimeFormatException: when time format is invalid (rest
                client returns 400 code).

        Returns:
            Local timestamp in ISO8601 format.
        """

        try:
            response = self._rest_client.get(
                self.URL, output_mode="json", time=time_str
            ).body.read()
        except binding.HTTPError as e:
            if e.status != 400:
                raise

            raise InvalidTimeFormatException(f"Invalid time format: {time_str}.")

        return json.loads(response)[time_str]

URL = '/services/search/timeparser' class-attribute instance-attribute

__init__(session_key, scheme=None, host=None, port=None, **context)

Initializes TimeParser.

Parameters:

Name Type Description Default
session_key str

Splunk access token.

 required
scheme str

(optional) The access scheme, default is None.

 None
host str

(optional) The host name, default is None.

 None
port int

(optional) The port number, default is None.

 None
context Any

Other configurations for Splunk rest client.

 {}

Raises:

Type Description
ValueError

if scheme, host or port are invalid.
Source code in solnlib/time_parser.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    session_key: str,
    scheme: str = None,
    host: str = None,
    port: int = None,
    **context: Any,
):
    """Initializes TimeParser.

    Arguments:
        session_key: Splunk access token.
        scheme: (optional) The access scheme, default is None.
        host: (optional) The host name, default is None.
        port: (optional) The port number, default is None.
        context: Other configurations for Splunk rest client.

    Raises:
        ValueError: if scheme, host or port are invalid.
    """
    self._rest_client = rest_client.SplunkRestClient(
        session_key, "-", scheme=scheme, host=host, port=port, **context
    )

to_local(time_str)

Parse time_str and convert to local timestamp.

Parameters:

Name Type Description Default
time_str str

ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

 required

Raises:

Type Description
binding.HTTPError

rest client returns an exception (everything else than 400 code).
InvalidTimeFormatException

when time format is invalid (rest client returns 400 code).

Returns:

Type Description
str

Local timestamp in ISO8601 format.
Source code in solnlib/time_parser.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@retry(exceptions=[binding.HTTPError])
def to_local(self, time_str: str) -> str:
    """Parse `time_str` and convert to local timestamp.

    Arguments:
        time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

    Raises:
        binding.HTTPError: rest client returns an exception (everything
            else than 400 code).
        InvalidTimeFormatException: when time format is invalid (rest
            client returns 400 code).

    Returns:
        Local timestamp in ISO8601 format.
    """

    try:
        response = self._rest_client.get(
            self.URL, output_mode="json", time=time_str
        ).body.read()
    except binding.HTTPError as e:
        if e.status != 400:
            raise

        raise InvalidTimeFormatException(f"Invalid time format: {time_str}.")

    return json.loads(response)[time_str]

to_seconds(time_str)

Parse time_str and convert to seconds since epoch.

Parameters:

Name Type Description Default
time_str str

ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

 required

Raises:

Type Description
binding.HTTPError

rest client returns an exception (everything else than 400 code).
InvalidTimeFormatException

when time format is invalid (rest client returns 400 code).

Returns:

Type Description
float

Seconds since epoch.
Source code in solnlib/time_parser.py 
 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
@retry(exceptions=[binding.HTTPError])
def to_seconds(self, time_str: str) -> float:
    """Parse `time_str` and convert to seconds since epoch.

    Arguments:
        time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

    Raises:
        binding.HTTPError: rest client returns an exception (everything
            else than 400 code).
        InvalidTimeFormatException: when time format is invalid (rest
            client returns 400 code).

    Returns:
        Seconds since epoch.
    """

    try:
        response = self._rest_client.get(
            self.URL, output_mode="json", time=time_str, output_time_format="%s"
        ).body.read()
    except binding.HTTPError as e:
        if e.status != 400:
            raise

        raise InvalidTimeFormatException(f"Invalid time format: {time_str}.")

    seconds = json.loads(response)[time_str]
    return float(seconds)

to_utc(time_str)

Parse time_str and convert to UTC timestamp.

Parameters:

Name Type Description Default
time_str str

ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

 required

Raises:

Type Description
binding.HTTPError

rest client returns an exception (everything else than 400 code).
InvalidTimeFormatException

when time format is invalid (rest client returns 400 code).

Returns:

Type Description
datetime.datetime

UTC timestamp.
Source code in solnlib/time_parser.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def to_utc(self, time_str: str) -> datetime.datetime:
    """Parse `time_str` and convert to UTC timestamp.

    Arguments:
        time_str: ISO8601 format timestamp, example: 2011-07-06T21:54:23.000-07:00.

    Raises:
        binding.HTTPError: rest client returns an exception (everything
            else than 400 code).
        InvalidTimeFormatException: when time format is invalid (rest
            client returns 400 code).

    Returns:
        UTC timestamp.
    """

    return datetime.datetime.utcfromtimestamp(self.to_seconds(time_str))