Skip to content

channels

Classes

ChannelsClient

ChannelsClient(client: OWUIClientBase)

Bases: ResourceBase

Source code in src/owui_client/client_base.py
def __init__(self, client: OWUIClientBase):
    self._client = client

Functions

list
list() -> List[ChannelListItemResponse]

Get list of channels for the current user.

Returns:

Type Description
List[ChannelListItemResponse]

List[ChannelListItemResponse]: A list of channels with additional user stats (last message, unread count).

Source code in src/owui_client/routers/channels.py
async def list(self) -> List[ChannelListItemResponse]:
    """
    Get list of channels for the current user.

    Returns:
        List[ChannelListItemResponse]: A list of channels with additional user stats (last message, unread count).
    """
    return await self._request(
        "GET",
        "/v1/channels/",
        model=ChannelListItemResponse,
    )
list_all
list_all() -> List[ChannelModel]

Get all channels available to the user.

For admins, this returns all channels. For regular users, this returns channels they are a member of.

Returns:

Type Description
List[ChannelModel]

List[ChannelModel]: A list of basic channel models.

Source code in src/owui_client/routers/channels.py
async def list_all(self) -> List[ChannelModel]:
    """
    Get all channels available to the user.

    For admins, this returns all channels. For regular users, this returns channels they are a member of.

    Returns:
        List[ChannelModel]: A list of basic channel models.
    """
    return await self._request(
        "GET",
        "/v1/channels/list",
        model=ChannelModel,
    )
get_dm_by_user
get_dm_by_user(user_id: str) -> Optional[ChannelModel]

Get or create a DM channel with a specific user.

If a DM channel already exists, it is returned. If not, a new one is created.

Parameters:

Name Type Description Default
user_id str

The ID of the user to start a DM with.

required

Returns:

Type Description
Optional[ChannelModel]

Optional[ChannelModel]: The DM channel model.

Source code in src/owui_client/routers/channels.py
async def get_dm_by_user(self, user_id: str) -> Optional[ChannelModel]:
    """
    Get or create a DM channel with a specific user.

    If a DM channel already exists, it is returned. If not, a new one is created.

    Args:
        user_id: The ID of the user to start a DM with.

    Returns:
        Optional[ChannelModel]: The DM channel model.
    """
    return await self._request(
        "GET",
        f"/v1/channels/users/{user_id}",
        model=Optional[ChannelModel],
    )
create
create(
    form_data: CreateChannelForm,
) -> Optional[ChannelModel]

Create a new channel.

Parameters:

Name Type Description Default
form_data CreateChannelForm

The form data for creating the channel (name, type, members, etc.).

required

Returns:

Type Description
Optional[ChannelModel]

Optional[ChannelModel]: The created channel model, or None if creation failed.

Source code in src/owui_client/routers/channels.py
async def create(self, form_data: CreateChannelForm) -> Optional[ChannelModel]:
    """
    Create a new channel.

    Args:
        form_data: The form data for creating the channel (name, type, members, etc.).

    Returns:
        Optional[ChannelModel]: The created channel model, or None if creation failed.
    """
    return await self._request(
        "POST",
        "/v1/channels/create",
        model=Optional[ChannelModel],
        json=form_data.model_dump(),
    )
get
get(id: str) -> Optional[ChannelFullResponse]

Get detailed channel information by ID.

Parameters:

Name Type Description Default
id str

The channel ID.

required

Returns:

Type Description
Optional[ChannelFullResponse]

Optional[ChannelFullResponse]: Detailed channel info including members and user permissions.

Source code in src/owui_client/routers/channels.py
async def get(self, id: str) -> Optional[ChannelFullResponse]:
    """
    Get detailed channel information by ID.

    Args:
        id: The channel ID.

    Returns:
        Optional[ChannelFullResponse]: Detailed channel info including members and user permissions.
    """
    return await self._request(
        "GET",
        f"/v1/channels/{id}",
        model=Optional[ChannelFullResponse],
    )
get_members
get_members(
    id: str,
    query: Optional[str] = None,
    order_by: Optional[str] = None,
    direction: Optional[str] = None,
    page: Optional[int] = 1,
) -> UserListResponse

Get members of a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
query Optional[str]

Optional search query for filtering members.

None
order_by Optional[str]

Field to order by.

None
direction Optional[str]

Sort direction ('asc' or 'desc').

None
page Optional[int]

Page number for pagination.

1

Returns:

Type Description
UserListResponse

UserListResponse: List of users and total count.

Source code in src/owui_client/routers/channels.py
async def get_members(
    self,
    id: str,
    query: Optional[str] = None,
    order_by: Optional[str] = None,
    direction: Optional[str] = None,
    page: Optional[int] = 1,
) -> UserListResponse:
    """
    Get members of a channel.

    Args:
        id: The channel ID.
        query: Optional search query for filtering members.
        order_by: Field to order by.
        direction: Sort direction ('asc' or 'desc').
        page: Page number for pagination.

    Returns:
        `UserListResponse`: List of users and total count.
    """
    params = {}
    if query:
        params["query"] = query
    if order_by:
        params["order_by"] = order_by
    if direction:
        params["direction"] = direction
    if page:
        params["page"] = page

    return await self._request(
        "GET",
        f"/v1/channels/{id}/members",
        model=UserListResponse,
        params=params,
    )
update_member_active_status
update_member_active_status(
    id: str, is_active: bool
) -> bool

Update the active status of the current user in the channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
is_active bool

The new active status.

required

Returns:

Name Type Description
bool bool

True if successful.

Source code in src/owui_client/routers/channels.py
async def update_member_active_status(self, id: str, is_active: bool) -> bool:
    """
    Update the active status of the current user in the channel.

    Args:
        id: The channel ID.
        is_active: The new active status.

    Returns:
        bool: True if successful.
    """
    form = UpdateActiveMemberForm(is_active=is_active)
    return await self._request(
        "POST",
        f"/v1/channels/{id}/members/active",
        model=bool,
        json=form.model_dump(),
    )
add_members
add_members(
    id: str,
    user_ids: List[str] = [],
    group_ids: List[str] = [],
) -> List[ChannelMemberModel]

Add members to a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
user_ids List[str]

List of user IDs to add.

[]
group_ids List[str]

List of group IDs to add.

[]

Returns:

Type Description
List[ChannelMemberModel]

List[ChannelMemberModel]: List of added memberships.

Source code in src/owui_client/routers/channels.py
async def add_members(
    self, id: str, user_ids: List[str] = [], group_ids: List[str] = []
) -> List[ChannelMemberModel]:
    """
    Add members to a channel.

    Args:
        id: The channel ID.
        user_ids: List of user IDs to add.
        group_ids: List of group IDs to add.

    Returns:
        List[ChannelMemberModel]: List of added memberships.
    """
    form = UpdateMembersForm(user_ids=user_ids, group_ids=group_ids)
    return await self._request(
        "POST",
        f"/v1/channels/{id}/update/members/add",
        model=ChannelMemberModel,
        json=form.model_dump(),
    )
remove_members
remove_members(id: str, user_ids: List[str]) -> int

Remove members from a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
user_ids List[str]

List of user IDs to remove.

required

Returns:

Name Type Description
int int

The number of members removed.

Source code in src/owui_client/routers/channels.py
async def remove_members(self, id: str, user_ids: List[str]) -> int:
    """
    Remove members from a channel.

    Args:
        id: The channel ID.
        user_ids: List of user IDs to remove.

    Returns:
        int: The number of members removed.
    """
    form = RemoveMembersForm(user_ids=user_ids)
    return await self._request(
        "POST",
        f"/v1/channels/{id}/update/members/remove",
        model=int,
        json=form.model_dump(),
    )
update
update(
    id: str, form_data: ChannelForm
) -> Optional[ChannelModel]

Update a channel by ID.

Parameters:

Name Type Description Default
id str

The channel ID.

required
form_data ChannelForm

The update form data.

required

Returns:

Type Description
Optional[ChannelModel]

Optional[ChannelModel]: The updated channel model.

Source code in src/owui_client/routers/channels.py
async def update(self, id: str, form_data: ChannelForm) -> Optional[ChannelModel]:
    """
    Update a channel by ID.

    Args:
        id: The channel ID.
        form_data: The update form data.

    Returns:
        Optional[ChannelModel]: The updated channel model.
    """
    return await self._request(
        "POST",
        f"/v1/channels/{id}/update",
        model=Optional[ChannelModel],
        json=form_data.model_dump(),
    )
delete
delete(id: str) -> bool

Delete a channel by ID.

Parameters:

Name Type Description Default
id str

The channel ID.

required

Returns:

Name Type Description
bool bool

True if successful.

Source code in src/owui_client/routers/channels.py
async def delete(self, id: str) -> bool:
    """
    Delete a channel by ID.

    Args:
        id: The channel ID.

    Returns:
        bool: True if successful.
    """
    return await self._request(
        "DELETE",
        f"/v1/channels/{id}/delete",
        model=bool,
    )
get_messages
get_messages(
    id: str, skip: int = 0, limit: int = 50
) -> List[MessageUserResponse]

Get messages from a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
skip int

Number of messages to skip.

0
limit int

Number of messages to return.

50

Returns:

Type Description
List[MessageUserResponse]

List[MessageUserResponse]: List of messages with user details.

Source code in src/owui_client/routers/channels.py
async def get_messages(
    self, id: str, skip: int = 0, limit: int = 50
) -> List[MessageUserResponse]:
    """
    Get messages from a channel.

    Args:
        id: The channel ID.
        skip: Number of messages to skip.
        limit: Number of messages to return.

    Returns:
        List[MessageUserResponse]: List of messages with user details.
    """
    params = {"skip": skip, "limit": limit}
    return await self._request(
        "GET",
        f"/v1/channels/{id}/messages",
        model=MessageUserResponse,
        params=params,
    )
get_pinned_messages
get_pinned_messages(
    id: str, page: int = 1
) -> List[MessageWithReactionsResponse]

Get pinned messages from a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
page int

Page number (1-based).

1

Returns:

Type Description
List[MessageWithReactionsResponse]

List[MessageWithReactionsResponse]: List of pinned messages.

Source code in src/owui_client/routers/channels.py
async def get_pinned_messages(
    self, id: str, page: int = 1
) -> List[MessageWithReactionsResponse]:
    """
    Get pinned messages from a channel.

    Args:
        id: The channel ID.
        page: Page number (1-based).

    Returns:
        List[MessageWithReactionsResponse]: List of pinned messages.
    """
    params = {"page": page}
    return await self._request(
        "GET",
        f"/v1/channels/{id}/messages/pinned",
        model=MessageWithReactionsResponse,
        params=params,
    )
post_message
post_message(
    id: str, form_data: MessageForm
) -> Optional[MessageModel]

Post a new message to a channel.

Parameters:

Name Type Description Default
id str

The channel ID.

required
form_data MessageForm

The message content and metadata.

required

Returns:

Type Description
Optional[MessageModel]

Optional[MessageModel]: The created message model.

Source code in src/owui_client/routers/channels.py
async def post_message(
    self, id: str, form_data: MessageForm
) -> Optional[MessageModel]:
    """
    Post a new message to a channel.

    Args:
        id: The channel ID.
        form_data: The message content and metadata.

    Returns:
        Optional[MessageModel]: The created message model.
    """
    return await self._request(
        "POST",
        f"/v1/channels/{id}/messages/post",
        model=Optional[MessageModel],
        json=form_data.model_dump(),
    )
get_message
get_message(
    id: str, message_id: str
) -> Optional[MessageUserResponse]

Get a specific message by ID.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required

Returns:

Type Description
Optional[MessageUserResponse]

Optional[MessageUserResponse]: The message details.

Source code in src/owui_client/routers/channels.py
async def get_message(
    self, id: str, message_id: str
) -> Optional[MessageUserResponse]:
    """
    Get a specific message by ID.

    Args:
        id: The channel ID.
        message_id: The message ID.

    Returns:
        Optional[MessageUserResponse]: The message details.
    """
    return await self._request(
        "GET",
        f"/v1/channels/{id}/messages/{message_id}",
        model=Optional[MessageUserResponse],
    )
pin_message
pin_message(
    id: str, message_id: str, is_pinned: bool
) -> Optional[MessageUserResponse]

Pin or unpin a message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required
is_pinned bool

True to pin, False to unpin.

required

Returns:

Type Description
Optional[MessageUserResponse]

Optional[MessageUserResponse]: The updated message details.

Source code in src/owui_client/routers/channels.py
async def pin_message(
    self, id: str, message_id: str, is_pinned: bool
) -> Optional[MessageUserResponse]:
    """
    Pin or unpin a message.

    Args:
        id: The channel ID.
        message_id: The message ID.
        is_pinned: True to pin, False to unpin.

    Returns:
        Optional[MessageUserResponse]: The updated message details.
    """
    return await self._request(
        "POST",
        f"/v1/channels/{id}/messages/{message_id}/pin",
        model=Optional[MessageUserResponse],
        json={"is_pinned": is_pinned},
    )
get_thread_messages
get_thread_messages(
    id: str, message_id: str, skip: int = 0, limit: int = 50
) -> List[MessageUserResponse]

Get thread messages for a specific message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The parent message ID.

required
skip int

Number of messages to skip.

0
limit int

Number of messages to return.

50

Returns:

Type Description
List[MessageUserResponse]

List[MessageUserResponse]: List of thread messages.

Source code in src/owui_client/routers/channels.py
async def get_thread_messages(
    self, id: str, message_id: str, skip: int = 0, limit: int = 50
) -> List[MessageUserResponse]:
    """
    Get thread messages for a specific message.

    Args:
        id: The channel ID.
        message_id: The parent message ID.
        skip: Number of messages to skip.
        limit: Number of messages to return.

    Returns:
        List[MessageUserResponse]: List of thread messages.
    """
    params = {"skip": skip, "limit": limit}
    return await self._request(
        "GET",
        f"/v1/channels/{id}/messages/{message_id}/thread",
        model=MessageUserResponse,
        params=params,
    )
update_message
update_message(
    id: str, message_id: str, form_data: MessageForm
) -> Optional[MessageModel]

Update a message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required
form_data MessageForm

The update form data.

required

Returns:

Type Description
Optional[MessageModel]

Optional[MessageModel]: The updated message model.

Source code in src/owui_client/routers/channels.py
async def update_message(
    self, id: str, message_id: str, form_data: MessageForm
) -> Optional[MessageModel]:
    """
    Update a message.

    Args:
        id: The channel ID.
        message_id: The message ID.
        form_data: The update form data.

    Returns:
        Optional[MessageModel]: The updated message model.
    """
    return await self._request(
        "POST",
        f"/v1/channels/{id}/messages/{message_id}/update",
        model=Optional[MessageModel],
        json=form_data.model_dump(),
    )
add_reaction
add_reaction(
    id: str, message_id: str, reaction_name: str
) -> bool

Add a reaction to a message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required
reaction_name str

The name of the reaction (e.g. emoji or shortcode).

required

Returns:

Name Type Description
bool bool

True if successful.

Source code in src/owui_client/routers/channels.py
async def add_reaction(self, id: str, message_id: str, reaction_name: str) -> bool:
    """
    Add a reaction to a message.

    Args:
        id: The channel ID.
        message_id: The message ID.
        reaction_name: The name of the reaction (e.g. emoji or shortcode).

    Returns:
        bool: True if successful.
    """
    form = ReactionForm(name=reaction_name)
    return await self._request(
        "POST",
        f"/v1/channels/{id}/messages/{message_id}/reactions/add",
        model=bool,
        json=form.model_dump(),
    )
remove_reaction
remove_reaction(
    id: str, message_id: str, reaction_name: str
) -> bool

Remove a reaction from a message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required
reaction_name str

The name of the reaction to remove.

required

Returns:

Name Type Description
bool bool

True if successful.

Source code in src/owui_client/routers/channels.py
async def remove_reaction(
    self, id: str, message_id: str, reaction_name: str
) -> bool:
    """
    Remove a reaction from a message.

    Args:
        id: The channel ID.
        message_id: The message ID.
        reaction_name: The name of the reaction to remove.

    Returns:
        bool: True if successful.
    """
    form = ReactionForm(name=reaction_name)
    return await self._request(
        "POST",
        f"/v1/channels/{id}/messages/{message_id}/reactions/remove",
        model=bool,
        json=form.model_dump(),
    )
delete_message
delete_message(id: str, message_id: str) -> bool

Delete a message.

Parameters:

Name Type Description Default
id str

The channel ID.

required
message_id str

The message ID.

required

Returns:

Name Type Description
bool bool

True if successful.

Source code in src/owui_client/routers/channels.py
async def delete_message(self, id: str, message_id: str) -> bool:
    """
    Delete a message.

    Args:
        id: The channel ID.
        message_id: The message ID.

    Returns:
        bool: True if successful.
    """
    return await self._request(
        "DELETE",
        f"/v1/channels/{id}/messages/{message_id}/delete",
        model=bool,
    )