Module zep_python.user.client

Expand source code
from typing import AsyncGenerator, Generator, List, Optional

import httpx
from httpx import AsyncClient, Client

from zep_python.exceptions import handle_response

from ..memory.models import Session
from .models import CreateUserRequest, UpdateUserRequest, User


class UserClient:
    """
    UserClient class implementation for user APIs.

    Attributes
    ----------
    aclient : httpx.AsyncClient
        The async client used for making API requests.
    client : httpx.Client
        The client used for making API requests.
    """

    def __init__(self, aclient: AsyncClient, client: Client) -> None:
        """
        Initialize the UserClient.

        Parameters
        ----------
        aclient : httpx.AsyncClient
            The async client used for making API requests.
        client : httpx.Client
            The client used for making API requests.
        """

        self.aclient = aclient
        self.client = client

    def add(self, user: CreateUserRequest) -> User:
        """
        Add a user.

        Parameters
        ----------
        user : CreateUserRequest
            The user to add.

        Returns
        -------
        User
            The user that was added.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.

        """
        try:
            response = self.client.post(
                "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    async def aadd(self, user: CreateUserRequest) -> User:
        """
        Async add a user.

        Parameters
        ----------
        user : CreateUserRequest
            The user to add.

        Returns
        -------
        User
            The user that was added.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.

        """
        try:
            response = await self.aclient.post(
                "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    def get(self, user_id: str) -> User:
        """
        Get a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to get.

        Returns
        -------
        User
            The user that was retrieved.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = self.client.get(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    async def aget(self, user_id: str) -> User:
        """
        Async get a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to get.

        Returns
        -------
        User
            The user that was retrieved.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        if user_id is None:
            raise ValueError("user_id must be provided")
        try:
            response = await self.aclient.get(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    def update(self, user: UpdateUserRequest) -> User:
        """
        Update a user.

        Parameters
        ----------
        user : UpdateUserRequest
            The user to update.

        Returns
        -------
        User
            The user that was updated.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist
        """
        if user.user_id is None:
            raise ValueError("user_id must be provided")

        try:
            response = self.client.patch(
                f"/users/{user.user_id}",
                json=user.model_dump(exclude_none=True, exclude_unset=True),
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)

        return User.model_validate(response.json())

    async def aupdate(self, user: UpdateUserRequest) -> User:
        """
        Async update a user.

        Parameters
        ----------
        user : UpdateUserRequest
            The user to update.

        Returns
        -------
        User
            The user that was updated.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        if user.user_id is None:
            raise ValueError("user_id must be provided")
        try:
            response = await self.aclient.patch(
                f"/users/{user.user_id}",
                json=user.model_dump(exclude_none=True, exclude_unset=True),
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

        return User.model_validate(response.json())

    def delete(self, user_id: str) -> None:
        """
        Delete a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to delete.

        Returns
        -------
        None

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = self.client.delete(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

    async def adelete(self, user_id: str) -> None:
        """
        Async delete a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to delete.

        Returns
        -------
        None

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = await self.aclient.delete(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

    def list(
        self, limit: Optional[int] = None, cursor: Optional[int] = None
    ) -> List[User]:
        """
        List users.

        Parameters
        ----------
        limit : Optional[int]
            The maximum number of users to return.
        cursor : Optional[int]
            The cursor to use for pagination.

        Returns
        -------
        List[User]
            The list of users. If no users are found, an empty list is returned.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = self.client.get(
                "/users", params={"limit": limit, "cursor": cursor}
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)
        return [User.model_validate(user) for user in response.json()]

    async def alist(
        self, limit: Optional[int] = None, cursor: Optional[int] = None
    ) -> List[User]:
        """
        Async list users.

        Parameters
        ----------
        limit : Optional[int]
            The maximum number of users to return.
        cursor : Optional[int]
            The cursor to use for pagination.

        Returns
        -------
        List[User]
            The list of users. An empty list is returned if there are no users.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = await self.aclient.get(
                "/users", params={"limit": limit, "cursor": cursor}
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)
        return [User.model_validate(user) for user in response.json()]

    def list_chunked(self, chunk_size: int = 100) -> Generator[List[User], None, None]:
        """
        List all users in chunks.

        This method uses pagination to retrieve the users in chunks and returns a
        generator that yields each chunk of users as a list.

        Parameters
        ----------
        chunk_size : int, optional
            The number of users to retrieve in each chunk, by default 100

        Returns
        -------
        Generator[List[User], None, None]
            A generator that yields each chunk of users as a list.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        cursor: Optional[int] = None

        while True:
            response = self.list(limit=chunk_size, cursor=cursor)

            if len(response) == 0:
                # We've reached the last page
                break

            yield response

            if cursor is None:
                cursor = 0
            cursor += chunk_size

    async def alist_chunked(
        self, chunk_size: int = 100
    ) -> AsyncGenerator[List[User], None]:
        """
        Async list all users in chunks.

        This method uses pagination to retrieve the users in chunks and returns a
        generator that yields each chunk of users as a list.

        Parameters
        ----------
        chunk_size : int, optional
            The number of users to retrieve in each chunk, by default 100

        Returns
        -------
        Generator[List[User], None, None]
            A generator that yields each chunk of users as a list.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        cursor: Optional[int] = None

        while True:
            response = await self.alist(limit=chunk_size, cursor=cursor)
            if len(response) == 0:
                # We've reached the last page
                break

            yield response

            if cursor is None:
                cursor = 0
            cursor += chunk_size

    def get_sessions(self, user_id: str) -> List[Session]:
        """
        List all sessions associated with this user.

        Parameters
        ----------
        user_id : str
            The user_id of the user whose sessions to list.

        Returns
        -------
        List[Session]
            The list of sessions.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = self.client.get(f"/users/{user_id}/sessions")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return [Session.model_validate(session) for session in response.json()]

    async def aget_sessions(self, user_id: str) -> List[Session]:
        """
        Async list all sessions associated with this user.

        Parameters
        ----------
        user_id : str
            The user_id of the user whose sessions to list.

        Returns
        -------
        List[Session]
            The list of sessions.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = await self.aclient.get(f"/users/{user_id}/sessions")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return [Session.model_validate(session) for session in response.json()]

Classes

class UserClient (aclient: httpx.AsyncClient, client: httpx.Client)

UserClient class implementation for user APIs.

Attributes

aclient : httpx.AsyncClient
The async client used for making API requests.
client : httpx.Client
The client used for making API requests.

Initialize the UserClient.

Parameters

aclient : httpx.AsyncClient
The async client used for making API requests.
client : httpx.Client
The client used for making API requests.
Expand source code
class UserClient:
    """
    UserClient class implementation for user APIs.

    Attributes
    ----------
    aclient : httpx.AsyncClient
        The async client used for making API requests.
    client : httpx.Client
        The client used for making API requests.
    """

    def __init__(self, aclient: AsyncClient, client: Client) -> None:
        """
        Initialize the UserClient.

        Parameters
        ----------
        aclient : httpx.AsyncClient
            The async client used for making API requests.
        client : httpx.Client
            The client used for making API requests.
        """

        self.aclient = aclient
        self.client = client

    def add(self, user: CreateUserRequest) -> User:
        """
        Add a user.

        Parameters
        ----------
        user : CreateUserRequest
            The user to add.

        Returns
        -------
        User
            The user that was added.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.

        """
        try:
            response = self.client.post(
                "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    async def aadd(self, user: CreateUserRequest) -> User:
        """
        Async add a user.

        Parameters
        ----------
        user : CreateUserRequest
            The user to add.

        Returns
        -------
        User
            The user that was added.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.

        """
        try:
            response = await self.aclient.post(
                "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    def get(self, user_id: str) -> User:
        """
        Get a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to get.

        Returns
        -------
        User
            The user that was retrieved.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = self.client.get(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    async def aget(self, user_id: str) -> User:
        """
        Async get a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to get.

        Returns
        -------
        User
            The user that was retrieved.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        if user_id is None:
            raise ValueError("user_id must be provided")
        try:
            response = await self.aclient.get(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return User.model_validate(response.json())

    def update(self, user: UpdateUserRequest) -> User:
        """
        Update a user.

        Parameters
        ----------
        user : UpdateUserRequest
            The user to update.

        Returns
        -------
        User
            The user that was updated.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist
        """
        if user.user_id is None:
            raise ValueError("user_id must be provided")

        try:
            response = self.client.patch(
                f"/users/{user.user_id}",
                json=user.model_dump(exclude_none=True, exclude_unset=True),
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)

        return User.model_validate(response.json())

    async def aupdate(self, user: UpdateUserRequest) -> User:
        """
        Async update a user.

        Parameters
        ----------
        user : UpdateUserRequest
            The user to update.

        Returns
        -------
        User
            The user that was updated.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        if user.user_id is None:
            raise ValueError("user_id must be provided")
        try:
            response = await self.aclient.patch(
                f"/users/{user.user_id}",
                json=user.model_dump(exclude_none=True, exclude_unset=True),
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

        return User.model_validate(response.json())

    def delete(self, user_id: str) -> None:
        """
        Delete a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to delete.

        Returns
        -------
        None

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = self.client.delete(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

    async def adelete(self, user_id: str) -> None:
        """
        Async delete a user.

        Parameters
        ----------
        user_id : str
            The user_id of the user to delete.

        Returns
        -------
        None

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        NotFoundError
            If the user does not exist.
        """
        try:
            response = await self.aclient.delete(f"/users/{user_id}")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)

    def list(
        self, limit: Optional[int] = None, cursor: Optional[int] = None
    ) -> List[User]:
        """
        List users.

        Parameters
        ----------
        limit : Optional[int]
            The maximum number of users to return.
        cursor : Optional[int]
            The cursor to use for pagination.

        Returns
        -------
        List[User]
            The list of users. If no users are found, an empty list is returned.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = self.client.get(
                "/users", params={"limit": limit, "cursor": cursor}
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)
        return [User.model_validate(user) for user in response.json()]

    async def alist(
        self, limit: Optional[int] = None, cursor: Optional[int] = None
    ) -> List[User]:
        """
        Async list users.

        Parameters
        ----------
        limit : Optional[int]
            The maximum number of users to return.
        cursor : Optional[int]
            The cursor to use for pagination.

        Returns
        -------
        List[User]
            The list of users. An empty list is returned if there are no users.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = await self.aclient.get(
                "/users", params={"limit": limit, "cursor": cursor}
            )
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e
        handle_response(response)
        return [User.model_validate(user) for user in response.json()]

    def list_chunked(self, chunk_size: int = 100) -> Generator[List[User], None, None]:
        """
        List all users in chunks.

        This method uses pagination to retrieve the users in chunks and returns a
        generator that yields each chunk of users as a list.

        Parameters
        ----------
        chunk_size : int, optional
            The number of users to retrieve in each chunk, by default 100

        Returns
        -------
        Generator[List[User], None, None]
            A generator that yields each chunk of users as a list.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        cursor: Optional[int] = None

        while True:
            response = self.list(limit=chunk_size, cursor=cursor)

            if len(response) == 0:
                # We've reached the last page
                break

            yield response

            if cursor is None:
                cursor = 0
            cursor += chunk_size

    async def alist_chunked(
        self, chunk_size: int = 100
    ) -> AsyncGenerator[List[User], None]:
        """
        Async list all users in chunks.

        This method uses pagination to retrieve the users in chunks and returns a
        generator that yields each chunk of users as a list.

        Parameters
        ----------
        chunk_size : int, optional
            The number of users to retrieve in each chunk, by default 100

        Returns
        -------
        Generator[List[User], None, None]
            A generator that yields each chunk of users as a list.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        cursor: Optional[int] = None

        while True:
            response = await self.alist(limit=chunk_size, cursor=cursor)
            if len(response) == 0:
                # We've reached the last page
                break

            yield response

            if cursor is None:
                cursor = 0
            cursor += chunk_size

    def get_sessions(self, user_id: str) -> List[Session]:
        """
        List all sessions associated with this user.

        Parameters
        ----------
        user_id : str
            The user_id of the user whose sessions to list.

        Returns
        -------
        List[Session]
            The list of sessions.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = self.client.get(f"/users/{user_id}/sessions")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return [Session.model_validate(session) for session in response.json()]

    async def aget_sessions(self, user_id: str) -> List[Session]:
        """
        Async list all sessions associated with this user.

        Parameters
        ----------
        user_id : str
            The user_id of the user whose sessions to list.

        Returns
        -------
        List[Session]
            The list of sessions.

        Raises
        ------
        ConnectionError
            If the client fails to connect to the server.
        APIError
            If the server returns an error.
        """
        try:
            response = await self.aclient.get(f"/users/{user_id}/sessions")
        except httpx.NetworkError as e:
            raise ConnectionError("Failed to connect to server") from e

        handle_response(response)
        return [Session.model_validate(session) for session in response.json()]

Methods

async def aadd(self, user: CreateUserRequest) ‑> User

Async add a user.

Parameters

user : CreateUserRequest
The user to add.

Returns

User
The user that was added.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
async def aadd(self, user: CreateUserRequest) -> User:
    """
    Async add a user.

    Parameters
    ----------
    user : CreateUserRequest
        The user to add.

    Returns
    -------
    User
        The user that was added.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.

    """
    try:
        response = await self.aclient.post(
            "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return User.model_validate(response.json())
def add(self, user: CreateUserRequest) ‑> User

Add a user.

Parameters

user : CreateUserRequest
The user to add.

Returns

User
The user that was added.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
def add(self, user: CreateUserRequest) -> User:
    """
    Add a user.

    Parameters
    ----------
    user : CreateUserRequest
        The user to add.

    Returns
    -------
    User
        The user that was added.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.

    """
    try:
        response = self.client.post(
            "/users", json=user.model_dump(exclude_none=True, exclude_unset=True)
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return User.model_validate(response.json())
async def adelete(self, user_id: str) ‑> None

Async delete a user.

Parameters

user_id : str
The user_id of the user to delete.

Returns

None
 

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist.
Expand source code
async def adelete(self, user_id: str) -> None:
    """
    Async delete a user.

    Parameters
    ----------
    user_id : str
        The user_id of the user to delete.

    Returns
    -------
    None

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist.
    """
    try:
        response = await self.aclient.delete(f"/users/{user_id}")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
async def aget(self, user_id: str) ‑> User

Async get a user.

Parameters

user_id : str
The user_id of the user to get.

Returns

User
The user that was retrieved.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist.
Expand source code
async def aget(self, user_id: str) -> User:
    """
    Async get a user.

    Parameters
    ----------
    user_id : str
        The user_id of the user to get.

    Returns
    -------
    User
        The user that was retrieved.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist.
    """
    if user_id is None:
        raise ValueError("user_id must be provided")
    try:
        response = await self.aclient.get(f"/users/{user_id}")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return User.model_validate(response.json())
async def aget_sessions(self, user_id: str) ‑> List[Session]

Async list all sessions associated with this user.

Parameters

user_id : str
The user_id of the user whose sessions to list.

Returns

List[Session]
The list of sessions.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
async def aget_sessions(self, user_id: str) -> List[Session]:
    """
    Async list all sessions associated with this user.

    Parameters
    ----------
    user_id : str
        The user_id of the user whose sessions to list.

    Returns
    -------
    List[Session]
        The list of sessions.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    try:
        response = await self.aclient.get(f"/users/{user_id}/sessions")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return [Session.model_validate(session) for session in response.json()]
async def alist(self, limit: Optional[int] = None, cursor: Optional[int] = None) ‑> List[User]

Async list users.

Parameters

limit : Optional[int]
The maximum number of users to return.
cursor : Optional[int]
The cursor to use for pagination.

Returns

List[User]
The list of users. An empty list is returned if there are no users.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
async def alist(
    self, limit: Optional[int] = None, cursor: Optional[int] = None
) -> List[User]:
    """
    Async list users.

    Parameters
    ----------
    limit : Optional[int]
        The maximum number of users to return.
    cursor : Optional[int]
        The cursor to use for pagination.

    Returns
    -------
    List[User]
        The list of users. An empty list is returned if there are no users.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    try:
        response = await self.aclient.get(
            "/users", params={"limit": limit, "cursor": cursor}
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e
    handle_response(response)
    return [User.model_validate(user) for user in response.json()]
async def alist_chunked(self, chunk_size: int = 100) ‑> AsyncGenerator[List[User], None]

Async list all users in chunks.

This method uses pagination to retrieve the users in chunks and returns a generator that yields each chunk of users as a list.

Parameters

chunk_size : int, optional
The number of users to retrieve in each chunk, by default 100

Returns

Generator[List[User], None, None]
A generator that yields each chunk of users as a list.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
async def alist_chunked(
    self, chunk_size: int = 100
) -> AsyncGenerator[List[User], None]:
    """
    Async list all users in chunks.

    This method uses pagination to retrieve the users in chunks and returns a
    generator that yields each chunk of users as a list.

    Parameters
    ----------
    chunk_size : int, optional
        The number of users to retrieve in each chunk, by default 100

    Returns
    -------
    Generator[List[User], None, None]
        A generator that yields each chunk of users as a list.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    cursor: Optional[int] = None

    while True:
        response = await self.alist(limit=chunk_size, cursor=cursor)
        if len(response) == 0:
            # We've reached the last page
            break

        yield response

        if cursor is None:
            cursor = 0
        cursor += chunk_size
async def aupdate(self, user: UpdateUserRequest) ‑> User

Async update a user.

Parameters

user : UpdateUserRequest
The user to update.

Returns

User
The user that was updated.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist.
Expand source code
async def aupdate(self, user: UpdateUserRequest) -> User:
    """
    Async update a user.

    Parameters
    ----------
    user : UpdateUserRequest
        The user to update.

    Returns
    -------
    User
        The user that was updated.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist.
    """
    if user.user_id is None:
        raise ValueError("user_id must be provided")
    try:
        response = await self.aclient.patch(
            f"/users/{user.user_id}",
            json=user.model_dump(exclude_none=True, exclude_unset=True),
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)

    return User.model_validate(response.json())
def delete(self, user_id: str) ‑> None

Delete a user.

Parameters

user_id : str
The user_id of the user to delete.

Returns

None
 

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist.
Expand source code
def delete(self, user_id: str) -> None:
    """
    Delete a user.

    Parameters
    ----------
    user_id : str
        The user_id of the user to delete.

    Returns
    -------
    None

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist.
    """
    try:
        response = self.client.delete(f"/users/{user_id}")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
def get(self, user_id: str) ‑> User

Get a user.

Parameters

user_id : str
The user_id of the user to get.

Returns

User
The user that was retrieved.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist.
Expand source code
def get(self, user_id: str) -> User:
    """
    Get a user.

    Parameters
    ----------
    user_id : str
        The user_id of the user to get.

    Returns
    -------
    User
        The user that was retrieved.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist.
    """
    try:
        response = self.client.get(f"/users/{user_id}")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return User.model_validate(response.json())
def get_sessions(self, user_id: str) ‑> List[Session]

List all sessions associated with this user.

Parameters

user_id : str
The user_id of the user whose sessions to list.

Returns

List[Session]
The list of sessions.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
def get_sessions(self, user_id: str) -> List[Session]:
    """
    List all sessions associated with this user.

    Parameters
    ----------
    user_id : str
        The user_id of the user whose sessions to list.

    Returns
    -------
    List[Session]
        The list of sessions.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    try:
        response = self.client.get(f"/users/{user_id}/sessions")
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e

    handle_response(response)
    return [Session.model_validate(session) for session in response.json()]
def list(self, limit: Optional[int] = None, cursor: Optional[int] = None) ‑> List[User]

List users.

Parameters

limit : Optional[int]
The maximum number of users to return.
cursor : Optional[int]
The cursor to use for pagination.

Returns

List[User]
The list of users. If no users are found, an empty list is returned.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
def list(
    self, limit: Optional[int] = None, cursor: Optional[int] = None
) -> List[User]:
    """
    List users.

    Parameters
    ----------
    limit : Optional[int]
        The maximum number of users to return.
    cursor : Optional[int]
        The cursor to use for pagination.

    Returns
    -------
    List[User]
        The list of users. If no users are found, an empty list is returned.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    try:
        response = self.client.get(
            "/users", params={"limit": limit, "cursor": cursor}
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e
    handle_response(response)
    return [User.model_validate(user) for user in response.json()]
def list_chunked(self, chunk_size: int = 100) ‑> Generator[List[User], None, None]

List all users in chunks.

This method uses pagination to retrieve the users in chunks and returns a generator that yields each chunk of users as a list.

Parameters

chunk_size : int, optional
The number of users to retrieve in each chunk, by default 100

Returns

Generator[List[User], None, None]
A generator that yields each chunk of users as a list.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
Expand source code
def list_chunked(self, chunk_size: int = 100) -> Generator[List[User], None, None]:
    """
    List all users in chunks.

    This method uses pagination to retrieve the users in chunks and returns a
    generator that yields each chunk of users as a list.

    Parameters
    ----------
    chunk_size : int, optional
        The number of users to retrieve in each chunk, by default 100

    Returns
    -------
    Generator[List[User], None, None]
        A generator that yields each chunk of users as a list.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    """
    cursor: Optional[int] = None

    while True:
        response = self.list(limit=chunk_size, cursor=cursor)

        if len(response) == 0:
            # We've reached the last page
            break

        yield response

        if cursor is None:
            cursor = 0
        cursor += chunk_size
def update(self, user: UpdateUserRequest) ‑> User

Update a user.

Parameters

user : UpdateUserRequest
The user to update.

Returns

User
The user that was updated.

Raises

ConnectionError
If the client fails to connect to the server.
APIError
If the server returns an error.
NotFoundError
If the user does not exist
Expand source code
def update(self, user: UpdateUserRequest) -> User:
    """
    Update a user.

    Parameters
    ----------
    user : UpdateUserRequest
        The user to update.

    Returns
    -------
    User
        The user that was updated.

    Raises
    ------
    ConnectionError
        If the client fails to connect to the server.
    APIError
        If the server returns an error.
    NotFoundError
        If the user does not exist
    """
    if user.user_id is None:
        raise ValueError("user_id must be provided")

    try:
        response = self.client.patch(
            f"/users/{user.user_id}",
            json=user.model_dump(exclude_none=True, exclude_unset=True),
        )
    except httpx.NetworkError as e:
        raise ConnectionError("Failed to connect to server") from e
    handle_response(response)

    return User.model_validate(response.json())