Skip to content

feat: add Event.list() for listing existing events #2686

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
threcc wants to merge 1 commit into
base: main
Choose a base branch
from
+76 −0
Open

feat: add Event.list() for listing existing events #2686

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions ocp_resources/event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import warnings
from collections.abc import Generator
from datetime import datetime, timedelta, timezone
from typing import Any

from kubernetes.dynamic import DynamicClient
Expand Down Expand Up @@ -90,6 +91,81 @@ def get(
timeout=timeout,
)

@staticmethod
def _parse_timestamp(event: Any) -> datetime | None:
"""Parse event timestamp, preferring lastTimestamp over creationTimestamp."""
timestamp = event.get("lastTimestamp") or event.get("metadata", {}).get("creationTimestamp")
if not timestamp:
return None
try:
return datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
except (ValueError, TypeError):
LOGGER.debug(f"Failed to parse event timestamp: {timestamp}")
Comment on lines +100 to +103
Copy link
Copy Markdown

@coderabbitai coderabbitai bot Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Verifies Python datetime behavior that can trigger the runtime error path.
python - <<'PY'
from datetime import datetime, timezone
samples = ["2026-04-03T12:00:00", "2026-04-03T12:00:00Z"]
cutoff = datetime.now(timezone.utc)
for ts in samples:
    parsed = datetime.fromisoformat(ts.replace("Z", "+00:00"))
    print(f"{ts} -> tzinfo={parsed.tzinfo}")
    try:
        print("comparison_ok:", parsed >= cutoff)
    except Exception as exc:
        print("comparison_error:", type(exc).__name__, str(exc))
PY

Repository: RedHatQE/openshift-python-wrapper

Length of output: 249

🏁 Script executed:

# Locate and examine the event.py file
git ls-files | grep -E "event\.py$"

Repository: RedHatQE/openshift-python-wrapper

Length of output: 99

🏁 Script executed:

# Read the file to understand the full context of _parse_timestamp and list methods
cat -n ocp_resources/event.py | head -220

Repository: RedHatQE/openshift-python-wrapper

Length of output: 10161

🏁 Script executed:

# Check for generated-code markers per coding guidelines
grep -n "Generated using\|End of generated code" ocp_resources/event.py

Repository: RedHatQE/openshift-python-wrapper

Length of output: 59

Normalize parsed timestamps to timezone-aware UTC to prevent comparison error at runtime.

When a timestamp lacks a timezone indicator (e.g., "2026-04-03T12:00:00"), datetime.fromisoformat() returns a naive datetime. Comparing this naive datetime to the timezone-aware cutoff at line 163 raises TypeError: can't compare offset-naive and offset-aware datetimes, silently bypassing the try-except block at line 102 and crashing the list() method.

Proposed fix 
     try:
-            return datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+            parsed = datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
+            if parsed.tzinfo is None:
+                parsed = parsed.replace(tzinfo=timezone.utc)
+            return parsed.astimezone(timezone.utc)
     except (ValueError, TypeError):
             LOGGER.debug(f"Failed to parse event timestamp: {timestamp}")
             return None

Also applies to: line 163

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@ocp_resources/event.py` around lines 100 - 103, The parsed timestamp can be
offset-naive and later compared to the timezone-aware cutoff in list(), causing
a TypeError; after calling datetime.fromisoformat(...) (and after replacing "Z"
with "+00:00"), ensure the resulting datetime is timezone-aware by setting
tzinfo to UTC when dt.tzinfo is None (use datetime.timezone.utc or equivalent)
before returning it from the parsing helper; this guarantees safe comparisons
against the timezone-aware cutoff used in list().

return None

@classmethod
def list(
cls,
client: DynamicClient,
namespace: str | None = None,
field_selector: str | None = None,
label_selector: str | None = None,
since_seconds: int = 300,
Copy link
Copy Markdown

@coderabbitai coderabbitai bot Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Validate since_seconds lower bound.

Negative values currently produce a future cutoff and unintuitive filtering. Add a guard to fail fast.

💡 Proposed fix 
     def list(
         cls,
         client: DynamicClient,
         namespace: str | None = None,
         field_selector: str | None = None,
         label_selector: str | None = None,
         since_seconds: int = 300,
     ) -> list[Any]:
+        if since_seconds < 0:
+            raise ValueError("since_seconds must be >= 0")

Also applies to: 159-159

🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@ocp_resources/event.py` at line 113, Add a guard to validate the
since_seconds parameter (the argument declared as "since_seconds: int = 300") to
ensure it is non‑negative: if since_seconds < 0, raise a clear ValueError (or
similar) with a descriptive message so the call fails fast; apply this same
validation to the other occurrence of since_seconds in the file (the second
declaration referenced in the review).

) -> list[Any]:
"""
List existing K8s events using a standard API list call (not watch).
Unlike ``Event.get()`` which uses watch and streams events in real-time,
this method returns already-existing events immediately.
Args:
client: K8s dynamic client.
namespace: Filter events to this namespace.
field_selector: Filter events by fields (e.g. ``"type==Warning"``).
label_selector: Filter events by labels.
since_seconds: Only return events from the last N seconds (default: 300 = 5 minutes).
Returns:
List of event resource objects, sorted by ``lastTimestamp`` descending (most recent first).
Example:
List Warning events from the last 5 minutes in a namespace::
events = Event.list(
client=client,
namespace="my-namespace",
field_selector="type==Warning",
)
"""
LOGGER.info("Listing events")
LOGGER.debug(
f"list events parameters: namespace={namespace},"
f" field_selector='{field_selector}', label_selector='{label_selector}',"
f" since_seconds={since_seconds}"
)

resource = client.resources.get(api_version=cls.api_version, kind=cls.__name__)
kwargs: dict[str, Any] = {}
if namespace:
kwargs["namespace"] = namespace
if field_selector:
kwargs["field_selector"] = field_selector
if label_selector:
kwargs["label_selector"] = label_selector

response = resource.get(**kwargs)
events = response.items or []

cutoff = datetime.now(tz=timezone.utc) - timedelta(seconds=since_seconds)
timed_events: list[tuple[datetime, Any]] = []
for event in events:
event_time = cls._parse_timestamp(event)
if event_time and event_time >= cutoff:
timed_events.append((event_time, event))

timed_events.sort(key=lambda pair: pair[0], reverse=True)
return [event for _, event in timed_events]

@classmethod
def delete_events(
cls,
Expand Down