Skip to content

Models

provide.foundation.integrations.openobserve.models

TODO: Add module docstring.

Classes

SearchQuery

Search query parameters for OpenObserve.

Functions
to_dict 
to_dict() -> dict[str, Any]

Convert to API request format.

Source code in provide/foundation/integrations/openobserve/models.py 
28
29
30
31
32
33
34
35
36
37
38
def to_dict(self) -> dict[str, Any]:
    """Convert to API request format."""
    return {
        "query": {
            "sql": self.sql,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "from": self.from_offset,
            "size": self.size,
        },
    }

SearchResponse

Response from OpenObserve search API.

Functions
from_dict classmethod 
from_dict(data: dict[str, Any]) -> SearchResponse

Create from API response.

Source code in provide/foundation/integrations/openobserve/models.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def from_dict(cls, data: dict[str, Any]) -> SearchResponse:
    """Create from API response."""
    return cls(
        hits=data.get("hits", []),
        total=data.get("total", 0),
        took=data.get("took", 0),
        scan_size=data.get("scan_size", 0),
        trace_id=data.get("trace_id"),
        from_offset=data.get("from", 0),
        size=data.get("size", 0),
        is_partial=data.get("is_partial", False),
        function_error=data.get("function_error", []),
    )

StreamInfo

Information about an OpenObserve stream.

Functions
from_dict classmethod 
from_dict(data: dict[str, Any]) -> StreamInfo

Create from API response.

Source code in provide/foundation/integrations/openobserve/models.py 
82
83
84
85
86
87
88
89
90
91
92
@classmethod
def from_dict(cls, data: dict[str, Any]) -> StreamInfo:
    """Create from API response."""
    return cls(
        name=data.get("name", ""),
        storage_type=data.get("storage_type", ""),
        stream_type=data.get("stream_type", ""),
        doc_count=data.get("stats", {}).get("doc_count", 0),
        compressed_size=data.get("stats", {}).get("compressed_size", 0),
        original_size=data.get("stats", {}).get("original_size", 0),
    )

Functions

parse_relative_time 

parse_relative_time(
    time_str: str, now: datetime | None = None
) -> int

Parse relative time strings like '-1h', '-30m' to microseconds since epoch.

Parameters:

Name Type Description Default
time_str str

Time string (e.g., '-1h', '-30m', 'now')

 required
now datetime | None

Current time (for testing), defaults to datetime.now()

 None

Returns:

Type Description
int

Microseconds since epoch
Source code in provide/foundation/integrations/openobserve/models.py 
 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
def parse_relative_time(time_str: str, now: datetime | None = None) -> int:
    """Parse relative time strings like '-1h', '-30m' to microseconds since epoch.

    Args:
        time_str: Time string (e.g., '-1h', '-30m', 'now')
        now: Current time (for testing), defaults to datetime.now()

    Returns:
        Microseconds since epoch

    """
    from datetime import timedelta

    if now is None:
        now = datetime.now()

    if time_str == "now":
        return int(now.timestamp() * 1_000_000)

    if time_str.startswith("-"):
        # Parse relative time
        value = time_str[1:]
        if value.endswith("h"):
            delta = timedelta(hours=int(value[:-1]))
        elif value.endswith("m"):
            delta = timedelta(minutes=int(value[:-1]))
        elif value.endswith("s"):
            delta = timedelta(seconds=int(value[:-1]))
        elif value.endswith("d"):
            delta = timedelta(days=int(value[:-1]))
        else:
            # Assume seconds if no unit
            delta = timedelta(seconds=int(value))

        target_time = now - delta
        return int(target_time.timestamp() * 1_000_000)

    # Try to parse as timestamp
    try:
        timestamp = int(time_str)
        # If it's already in microseconds (large number), return as-is
        if timestamp > 1_000_000_000_000:
            return timestamp
        # Otherwise assume seconds and convert
        return timestamp * 1_000_000
    except ValueError:
        # Try to parse as ISO datetime
        dt = datetime.fromisoformat(time_str)
        return int(dt.timestamp() * 1_000_000)