Skip to content

Config

provide.foundation.integrations.openobserve.config

TODO: Add module docstring.

Classes

OpenObserveConfig

Bases: RuntimeConfig

Configuration for OpenObserve integration.

Functions
get_otlp_endpoint 
get_otlp_endpoint() -> str | None

Get OTLP endpoint derived from OpenObserve URL.

Returns:

Type Description
str | None

OTLP endpoint URL or None if not configured
Source code in provide/foundation/integrations/openobserve/config.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_otlp_endpoint(self) -> str | None:
    """Get OTLP endpoint derived from OpenObserve URL.

    Returns:
        OTLP endpoint URL or None if not configured

    """
    if not self.url:
        return None

    # Remove /api/{org} suffix if present
    base_url = self.url
    if "/api/" in base_url:
        # Extract base URL before /api/
        base_url = base_url.split("/api/")[0]

    # Construct OTLP endpoint
    org = self.org or "default"
    return f"{base_url}/api/{org}"
is_available 
is_available() -> bool

Test if OpenObserve is available and reachable.

Returns:

Type Description
bool

True if connection test succeeds
Source code in provide/foundation/integrations/openobserve/config.py 
 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
def is_available(self) -> bool:
    """Test if OpenObserve is available and reachable.

    Returns:
        True if connection test succeeds

    """
    if not self.is_configured():
        return False

    try:
        # Import here to avoid circular dependency
        import asyncio

        from provide.foundation.integrations.openobserve.client import OpenObserveClient

        client = OpenObserveClient(
            url=self.url,  # type: ignore[arg-type]
            username=self.user,  # type: ignore[arg-type]
            password=self.password,  # type: ignore[arg-type]
            organization=self.org or "default",
        )
        return asyncio.run(client.test_connection())
    except Exception:
        return False
is_configured 
is_configured() -> bool

Check if OpenObserve is configured with required settings.

Returns:

Type Description
bool

True if URL, user, and password are all set
Source code in provide/foundation/integrations/openobserve/config.py 
48
49
50
51
52
53
54
55
def is_configured(self) -> bool:
    """Check if OpenObserve is configured with required settings.

    Returns:
        True if URL, user, and password are all set

    """
    return bool(self.url and self.user and self.password)

Functions