Skip to content

Chat Sessions (🧪Beta)

albert.collections.chat_sessions.ChatSessionCollection

ChatSessionCollection(*, session: AsyncAlbertSession)

Async collection for managing chat sessions (🧪Beta).

Beta Feature!

Please do not use in production or without explicit guidance from Albert. You might otherwise have a bad experience. This feature currently falls outside of the Albert support contract, but we'd love your feedback!

Parameters:

Name Type Description Default
session AsyncAlbertSession

The Albert async session instance.

required

Attributes:

Name Type Description
base_path str

The base URL for chat session API requests.

Methods:

Name Description
create

Creates a new chat session.

get_by_id

Retrieves a chat session by its ID.

get_by_source_session_id

Retrieves a chat session by its external source session ID.

get_all

Iterates over chat sessions with optional filters.

update

Updates a chat session by ID.

delete

Deletes a chat session by its ID.

Parameters:

Name Type Description Default
session AsyncAlbertSession

The async session used to make API requests.

required
Source code in src/albert/collections/chat_sessions.py
def __init__(self, *, session: AsyncAlbertSession):
    """
    Initializes the ChatSessionCollection with the provided session.

    Parameters
    ----------
    session : AsyncAlbertSession
        The async session used to make API requests.
    """
    self._session = session
    self.base_path: str = f"/api/{self._api_version}/chats/sessions"

base_path

base_path: str = f'/api/{_api_version}/chats/sessions'

create

create(*, session: ChatSession) -> ChatSession

Create a new chat session.

Parameters:

Name Type Description Default
session ChatSession

The session to create.

required

Returns:

Type Description
ChatSession

The created session.

Source code in src/albert/collections/chat_sessions.py
@validate_call
async def create(self, *, session: ChatSession) -> ChatSession:
    """
    Create a new chat session.

    Parameters
    ----------
    session : ChatSession
        The session to create.

    Returns
    -------
    ChatSession
        The created session.
    """
    response = await self._session.post(
        self.base_path,
        json=session.model_dump(by_alias=True, exclude_unset=True, mode="json"),
    )
    return ChatSession(**response.json())

get_by_id

get_by_id(*, id: str) -> ChatSession

Retrieve a chat session by its ID.

Parameters:

Name Type Description Default
id str

The session ID.

required

Returns:

Type Description
ChatSession

The matching session.

Source code in src/albert/collections/chat_sessions.py
@validate_call
async def get_by_id(self, *, id: str) -> ChatSession:
    """
    Retrieve a chat session by its ID.

    Parameters
    ----------
    id : str
        The session ID.

    Returns
    -------
    ChatSession
        The matching session.
    """
    response = await self._session.get(f"{self.base_path}/{id}")
    return ChatSession(**response.json())

get_by_source_session_id

get_by_source_session_id(
    *, source_session_id: str
) -> ChatSession

Retrieve a chat session by its external source session ID.

Parameters:

Name Type Description Default
source_session_id str

The external source session identifier.

required

Returns:

Type Description
ChatSession

The matching session.

Source code in src/albert/collections/chat_sessions.py
@validate_call
async def get_by_source_session_id(self, *, source_session_id: str) -> ChatSession:
    """
    Retrieve a chat session by its external source session ID.

    Parameters
    ----------
    source_session_id : str
        The external source session identifier.

    Returns
    -------
    ChatSession
        The matching session.
    """
    response = await self._session.get(f"{self.base_path}/source/{source_session_id}")
    return ChatSession(**response.json())

get_all

get_all(
    *,
    name: list[str] | None = None,
    exact_match: bool = False,
    parent_id: str | None = None,
    max_items: int | None = None,
) -> AsyncIterator[ChatSession]

Iterate over chat sessions with optional filters.

Parameters:

Name Type Description Default
name list[str] | None

Filter by session name(s).

None
exact_match bool

Whether name filtering uses exact matching (default False).

False
parent_id str | None

Filter sessions by folder ID.

None
max_items int | None

Maximum number of items to return in total. If None, fetches all available items.

None

Yields:

Type Description
ChatSession

Sessions matching the given filters.

Source code in src/albert/collections/chat_sessions.py
@validate_call
async def get_all(
    self,
    *,
    name: list[str] | None = None,
    exact_match: bool = False,
    parent_id: str | None = None,
    max_items: int | None = None,
) -> AsyncIterator[ChatSession]:
    """
    Iterate over chat sessions with optional filters.

    Parameters
    ----------
    name : list[str] | None, optional
        Filter by session name(s).
    exact_match : bool, optional
        Whether name filtering uses exact matching (default False).
    parent_id : str | None, optional
        Filter sessions by folder ID.
    max_items : int | None, optional
        Maximum number of items to return in total. If None, fetches all available items.

    Yields
    ------
    ChatSession
        Sessions matching the given filters.
    """
    params: dict[str, str | list[str]] = {}
    if name:
        params["name"] = name
    if exact_match:
        params["exactMatch"] = "true"
    if parent_id is not None:
        params["parentId"] = parent_id

    async for session in AsyncAlbertPaginator(
        session=self._session,
        path=self.base_path,
        deserialize=lambda item: ChatSession(**item),
        params=params,
        max_items=max_items,
    ):
        yield session

update

update(
    *,
    id: str,
    name: str | None = None,
    parent_id: str | None | _UnsetType = _UNSET,
) -> ChatSession

Update a chat session.

Parameters:

Name Type Description Default
id str

The ID of the session to update.

required
name str | None

New display name for the session.

None
parent_id str | None

New parent folder ID for the session. Pass None to remove the session from its current folder.

_UNSET

Returns:

Type Description
ChatSession

The updated session.

Notes

The following fields can be updated: name, parent_id.

Source code in src/albert/collections/chat_sessions.py
@validate_call
async def update(
    self,
    *,
    id: str,
    name: str | None = None,
    parent_id: str | None | _UnsetType = _UNSET,
) -> ChatSession:
    """
    Update a chat session.

    Parameters
    ----------
    id : str
        The ID of the session to update.
    name : str | None, optional
        New display name for the session.
    parent_id : str | None, optional
        New parent folder ID for the session. Pass ``None`` to remove the session from its
        current folder.

    Returns
    -------
    ChatSession
        The updated session.

    Notes
    -----
    The following fields can be updated: ``name``, ``parent_id``.
    """
    data = []
    if name is not None:
        data.append({"operation": "update", "attribute": "name", "newValue": name})
    if parent_id is not _UNSET:
        data.append({"operation": "update", "attribute": "parentId", "newValue": parent_id})
    if not data:
        return await self.get_by_id(id=id)
    await self._session.patch(f"{self.base_path}/{id}", json={"data": data})
    return await self.get_by_id(id=id)

delete

delete(*, id: str) -> None

Delete a chat session by ID.

Parameters:

Name Type Description Default
id str

The ID of the session to delete.

required

Returns:

Type Description
None
Source code in src/albert/collections/chat_sessions.py
@validate_call
async def delete(self, *, id: str) -> None:
    """
    Delete a chat session by ID.

    Parameters
    ----------
    id : str
        The ID of the session to delete.

    Returns
    -------
    None
    """
    await self._session.delete(f"{self.base_path}/{id}")