"""Session API module for HoneyHive."""
# pylint: disable=useless-parent-delegation
# Note: BaseAPI.__init__ performs important setup (error_handler, _client_name)
# The delegation is not useless despite pylint's false positive
from typing import TYPE_CHECKING, Any, Optional
from ..models import Event, SessionStartRequest
from .base import BaseAPI
if TYPE_CHECKING:
from .client import HoneyHive
[docs]
class SessionStartResponse: # pylint: disable=too-few-public-methods
"""Response from starting a session.
Contains the result of a session creation operation including
the session ID.
"""
def __init__(self, session_id: str):
"""Initialize the response.
Args:
session_id: Unique identifier for the created session
"""
self.session_id = session_id
@property
def id(self) -> str:
"""Alias for session_id for compatibility.
Returns:
The session ID
"""
return self.session_id
@property
def _id(self) -> str:
"""Alias for session_id for compatibility.
Returns:
The session ID
"""
return self.session_id
[docs]
class SessionResponse: # pylint: disable=too-few-public-methods
"""Response from getting a session.
Contains the session data retrieved from the API.
"""
def __init__(self, event: Event):
"""Initialize the response.
Args:
event: Event object containing session information
"""
self.event = event
[docs]
class SessionAPI(BaseAPI):
"""API for session operations."""
def __init__(self, client: "HoneyHive") -> None:
"""Initialize the SessionAPI."""
super().__init__(client)
# Session-specific initialization can be added here if needed
[docs]
def create_session(self, session: SessionStartRequest) -> SessionStartResponse:
"""Create a new session using SessionStartRequest model."""
response = self.client.request(
"POST",
"/session/start",
json={"session": session.model_dump(mode="json", exclude_none=True)},
)
data = response.json()
return SessionStartResponse(session_id=data["session_id"])
[docs]
def create_session_from_dict(self, session_data: dict) -> SessionStartResponse:
"""Create a new session from session data dictionary (legacy method)."""
# Handle both direct session data and nested session data
if "session" in session_data:
request_data = session_data
else:
request_data = {"session": session_data}
response = self.client.request("POST", "/session/start", json=request_data)
data = response.json()
return SessionStartResponse(session_id=data["session_id"])
[docs]
async def create_session_async(
self, session: SessionStartRequest
) -> SessionStartResponse:
"""Create a new session asynchronously using SessionStartRequest model."""
response = await self.client.request_async(
"POST",
"/session/start",
json={"session": session.model_dump(mode="json", exclude_none=True)},
)
data = response.json()
return SessionStartResponse(session_id=data["session_id"])
[docs]
async def create_session_from_dict_async(
self, session_data: dict
) -> SessionStartResponse:
"""Create a new session asynchronously from session data dictionary \
(legacy method)."""
# Handle both direct session data and nested session data
if "session" in session_data:
request_data = session_data
else:
request_data = {"session": session_data}
response = await self.client.request_async(
"POST", "/session/start", json=request_data
)
data = response.json()
return SessionStartResponse(session_id=data["session_id"])
[docs]
def start_session(
self,
project: str,
session_name: str,
source: str,
session_id: Optional[str] = None,
**kwargs: Any,
) -> SessionStartResponse:
"""Start a new session using SessionStartRequest model."""
request_data = SessionStartRequest(
project=project,
session_name=session_name,
source=source,
session_id=session_id,
**kwargs,
)
response = self.client.request(
"POST",
"/session/start",
json={"session": request_data.model_dump(mode="json", exclude_none=True)},
)
data = response.json()
self.client._log( # pylint: disable=protected-access
"debug", "Session API response", honeyhive_data={"response_data": data}
)
# Check if session_id exists in the response
if "session_id" in data:
return SessionStartResponse(session_id=data["session_id"])
if "session" in data and "session_id" in data["session"]:
return SessionStartResponse(session_id=data["session"]["session_id"])
self.client._log( # pylint: disable=protected-access
"warning",
"Unexpected session response structure",
honeyhive_data={"response_data": data},
)
# Try to find session_id in nested structures
if "session" in data:
session_data = data["session"]
if isinstance(session_data, dict) and "session_id" in session_data:
return SessionStartResponse(session_id=session_data["session_id"])
# If we still can't find it, raise an error with the full response
raise ValueError(f"Session ID not found in response: {data}")
[docs]
async def start_session_async(
self,
project: str,
session_name: str,
source: str,
session_id: Optional[str] = None,
**kwargs: Any,
) -> SessionStartResponse:
"""Start a new session asynchronously using SessionStartRequest model."""
request_data = SessionStartRequest(
project=project,
session_name=session_name,
source=source,
session_id=session_id,
**kwargs,
)
response = await self.client.request_async(
"POST",
"/session/start",
json={"session": request_data.model_dump(mode="json", exclude_none=True)},
)
data = response.json()
return SessionStartResponse(session_id=data["session_id"])
[docs]
def get_session(self, session_id: str) -> SessionResponse:
"""Get a session by ID."""
response = self.client.request("GET", f"/session/{session_id}")
data = response.json()
return SessionResponse(event=Event(**data))
[docs]
async def get_session_async(self, session_id: str) -> SessionResponse:
"""Get a session by ID asynchronously."""
response = await self.client.request_async("GET", f"/session/{session_id}")
data = response.json()
return SessionResponse(event=Event(**data))
[docs]
def delete_session(self, session_id: str) -> bool:
"""Delete a session by ID."""
context = self._create_error_context(
operation="delete_session",
method="DELETE",
path=f"/session/{session_id}",
additional_context={"session_id": session_id},
)
with self.error_handler.handle_operation(context):
response = self.client.request("DELETE", f"/session/{session_id}")
return response.status_code == 200
[docs]
async def delete_session_async(self, session_id: str) -> bool:
"""Delete a session by ID asynchronously."""
context = self._create_error_context(
operation="delete_session_async",
method="DELETE",
path=f"/session/{session_id}",
additional_context={"session_id": session_id},
)
with self.error_handler.handle_operation(context):
response = await self.client.request_async(
"DELETE", f"/session/{session_id}"
)
return response.status_code == 200