feat! Remove list_events from session service interface as it has many overlap with get_session.

PiperOrigin-RevId: 759184310
This commit is contained in:
Shangjie Chen 2025-05-15 09:40:39 -07:00 committed by Copybara-Service
parent 2214c1558f
commit 3ea9eca929
4 changed files with 0 additions and 69 deletions

View File

@ -40,13 +40,6 @@ class ListSessionsResponse(BaseModel):
sessions: list[Session] = Field(default_factory=list) sessions: list[Session] = Field(default_factory=list)
class ListEventsResponse(BaseModel):
"""The response of listing events in a session."""
events: list[Event] = Field(default_factory=list)
next_page_token: Optional[str] = None
class BaseSessionService(abc.ABC): class BaseSessionService(abc.ABC):
"""Base class for session services. """Base class for session services.
@ -102,17 +95,6 @@ class BaseSessionService(abc.ABC):
"""Deletes a session.""" """Deletes a session."""
pass pass
@abc.abstractmethod
def list_events(
self,
*,
app_name: str,
user_id: str,
session_id: str,
) -> ListEventsResponse:
"""Lists events in a session."""
pass
def close_session(self, *, session: Session): def close_session(self, *, session: Session):
"""Closes a session.""" """Closes a session."""
# TODO: determine whether we want to finalize the session here. # TODO: determine whether we want to finalize the session here.

View File

@ -49,7 +49,6 @@ from ..events.event import Event
from . import _session_util from . import _session_util
from .base_session_service import BaseSessionService from .base_session_service import BaseSessionService
from .base_session_service import GetSessionConfig from .base_session_service import GetSessionConfig
from .base_session_service import ListEventsResponse
from .base_session_service import ListSessionsResponse from .base_session_service import ListSessionsResponse
from .session import Session from .session import Session
from .state import State from .state import State
@ -556,17 +555,6 @@ class DatabaseSessionService(BaseSessionService):
super().append_event(session=session, event=event) super().append_event(session=session, event=event)
return event return event
@override
def list_events(
self,
*,
app_name: str,
user_id: str,
session_id: str,
) -> ListEventsResponse:
raise NotImplementedError()
def convert_event(event: StorageEvent) -> Event: def convert_event(event: StorageEvent) -> Event:
"""Converts a storage event to an event.""" """Converts a storage event to an event."""
return Event( return Event(

View File

@ -24,7 +24,6 @@ from typing_extensions import override
from ..events.event import Event from ..events.event import Event
from .base_session_service import BaseSessionService from .base_session_service import BaseSessionService
from .base_session_service import GetSessionConfig from .base_session_service import GetSessionConfig
from .base_session_service import ListEventsResponse
from .base_session_service import ListSessionsResponse from .base_session_service import ListSessionsResponse
from .session import Session from .session import Session
from .state import State from .state import State
@ -292,13 +291,3 @@ class InMemorySessionService(BaseSessionService):
storage_session.last_update_time = event.timestamp storage_session.last_update_time = event.timestamp
return event return event
@override
def list_events(
self,
*,
app_name: str,
user_id: str,
session_id: str,
) -> ListEventsResponse:
raise NotImplementedError()

View File

@ -25,7 +25,6 @@ from ..events.event_actions import EventActions
from . import _session_util from . import _session_util
from .base_session_service import BaseSessionService from .base_session_service import BaseSessionService
from .base_session_service import GetSessionConfig from .base_session_service import GetSessionConfig
from .base_session_service import ListEventsResponse
from .base_session_service import ListSessionsResponse from .base_session_service import ListSessionsResponse
from .session import Session from .session import Session
@ -213,33 +212,6 @@ class VertexAiSessionService(BaseSessionService):
request_dict={}, request_dict={},
) )
@override
def list_events(
self,
*,
app_name: str,
user_id: str,
session_id: str,
) -> ListEventsResponse:
reasoning_engine_id = _parse_reasoning_engine_id(app_name)
api_response = self.api_client.request(
http_method='GET',
path=f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}/events',
request_dict={},
)
logger.info(f'List events response {api_response}')
# Handles empty response case
if api_response.get('httpHeaders', None):
return ListEventsResponse()
session_events = api_response['sessionEvents']
return ListEventsResponse(
events=[_from_api_event(event) for event in session_events]
)
@override @override
def append_event(self, session: Session, event: Event) -> Event: def append_event(self, session: Session, event: Event) -> Event:
# Update the in-memory session. # Update the in-memory session.