# Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. # See LICENSE.txt for license information. """ Pythonic wrapper types for Mattermost Plugin API entities. This module provides dataclass-based wrappers that convert between the raw protobuf messages and Pythonic types. These wrappers are the public API surface for SDK users - they never need to interact with protobuf directly. Naming Convention: - `from_proto(proto)` - classmethod to create wrapper from protobuf - `to_proto()` - method to convert wrapper back to protobuf All wrapper types are immutable (frozen dataclasses) to prevent accidental modification and enable hashing. """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Dict, List, Optional, TYPE_CHECKING if TYPE_CHECKING: from mattermost_plugin.grpc import ( user_pb2, team_pb2, channel_pb2, api_user_team_pb2, api_channel_post_pb2, post_pb2, file_pb2, api_file_bot_pb2, api_kv_config_pb2, api_remaining_pb2, ) # ============================================================================= # ENUMS # ============================================================================= class TeamType(str, Enum): """Type of team visibility.""" OPEN = "O" INVITE = "I" class ChannelType(str, Enum): """Type of channel.""" OPEN = "O" PRIVATE = "P" DIRECT = "D" GROUP = "G" # ============================================================================= # USER TYPES # ============================================================================= @dataclass(frozen=True) class User: """ Represents a Mattermost user. This is a Pythonic wrapper around the protobuf User message that provides type safety and a clean API surface. Attributes: id: Unique identifier for the user. create_at: Unix timestamp (milliseconds) when user was created. update_at: Unix timestamp (milliseconds) when user was last updated. delete_at: Unix timestamp (milliseconds) when user was deleted (0 if not deleted). username: Unique username for the user. email: User's email address. email_verified: Whether the email has been verified. nickname: User's display nickname. first_name: User's first name. last_name: User's last name. position: User's job position/title. roles: Space-separated list of roles. locale: User's preferred locale. timezone: User's timezone settings. is_bot: Whether this user is a bot account. props: User properties. notify_props: Notification preferences. """ id: str username: str = "" email: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 email_verified: bool = False nickname: str = "" first_name: str = "" last_name: str = "" position: str = "" roles: str = "" locale: str = "" timezone: Dict[str, str] = field(default_factory=dict) is_bot: bool = False bot_description: str = "" auth_data: Optional[str] = None auth_service: str = "" props: Dict[str, str] = field(default_factory=dict) notify_props: Dict[str, str] = field(default_factory=dict) last_password_update: int = 0 last_picture_update: int = 0 failed_attempts: int = 0 mfa_active: bool = False remote_id: Optional[str] = None last_activity_at: int = 0 bot_last_icon_update: int = 0 terms_of_service_id: str = "" terms_of_service_create_at: int = 0 disable_welcome_email: bool = False last_login: int = 0 @classmethod def from_proto(cls, proto: "user_pb2.User") -> "User": """Create a User from a protobuf message.""" return cls( id=proto.id, username=proto.username, email=proto.email, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, email_verified=proto.email_verified, nickname=proto.nickname, first_name=proto.first_name, last_name=proto.last_name, position=proto.position, roles=proto.roles, locale=proto.locale, timezone=dict(proto.timezone), is_bot=proto.is_bot, bot_description=proto.bot_description, auth_data=proto.auth_data if proto.HasField("auth_data") else None, auth_service=proto.auth_service, props=dict(proto.props), notify_props=dict(proto.notify_props), last_password_update=proto.last_password_update, last_picture_update=proto.last_picture_update, failed_attempts=proto.failed_attempts, mfa_active=proto.mfa_active, remote_id=proto.remote_id if proto.HasField("remote_id") else None, last_activity_at=proto.last_activity_at, bot_last_icon_update=proto.bot_last_icon_update, terms_of_service_id=proto.terms_of_service_id, terms_of_service_create_at=proto.terms_of_service_create_at, disable_welcome_email=proto.disable_welcome_email, last_login=proto.last_login, ) def to_proto(self) -> "user_pb2.User": """Convert to a protobuf message.""" from mattermost_plugin.grpc import user_pb2 proto = user_pb2.User( id=self.id, username=self.username, email=self.email, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, email_verified=self.email_verified, nickname=self.nickname, first_name=self.first_name, last_name=self.last_name, position=self.position, roles=self.roles, locale=self.locale, is_bot=self.is_bot, bot_description=self.bot_description, auth_service=self.auth_service, last_password_update=self.last_password_update, last_picture_update=self.last_picture_update, failed_attempts=self.failed_attempts, mfa_active=self.mfa_active, last_activity_at=self.last_activity_at, bot_last_icon_update=self.bot_last_icon_update, terms_of_service_id=self.terms_of_service_id, terms_of_service_create_at=self.terms_of_service_create_at, disable_welcome_email=self.disable_welcome_email, last_login=self.last_login, ) # Set optional fields if self.auth_data is not None: proto.auth_data = self.auth_data if self.remote_id is not None: proto.remote_id = self.remote_id # Set map fields proto.timezone.update(self.timezone) proto.props.update(self.props) proto.notify_props.update(self.notify_props) return proto @dataclass(frozen=True) class UserStatus: """ Represents a user's online status. Attributes: user_id: The ID of the user. status: Status string (online, away, dnd, offline). manual: Whether status was manually set. last_activity_at: Unix timestamp of last activity. dnd_end_time: Unix timestamp when DND ends (0 if not in DND). """ user_id: str status: str manual: bool = False last_activity_at: int = 0 dnd_end_time: int = 0 @classmethod def from_proto(cls, proto: "api_user_team_pb2.Status") -> "UserStatus": """Create a UserStatus from a protobuf message.""" return cls( user_id=proto.user_id, status=proto.status, manual=proto.manual, last_activity_at=proto.last_activity_at, dnd_end_time=proto.dnd_end_time, ) @dataclass(frozen=True) class CustomStatus: """ Represents a user's custom status. Attributes: emoji: The emoji for the custom status. text: The text for the custom status. duration: Duration preset (e.g., "thirty_minutes", "one_hour"). expires_at: Unix timestamp when status expires. """ emoji: str = "" text: str = "" duration: str = "" expires_at: int = 0 @classmethod def from_proto(cls, proto: "api_user_team_pb2.CustomStatus") -> "CustomStatus": """Create a CustomStatus from a protobuf message.""" return cls( emoji=proto.emoji, text=proto.text, duration=proto.duration, expires_at=proto.expires_at, ) def to_proto(self) -> "api_user_team_pb2.CustomStatus": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_user_team_pb2 return api_user_team_pb2.CustomStatus( emoji=self.emoji, text=self.text, duration=self.duration, expires_at=self.expires_at, ) @dataclass(frozen=True) class UserAuth: """ Represents a user's authentication information. Attributes: auth_data: The authentication data (e.g., LDAP ID). auth_service: The authentication service name. """ auth_data: Optional[str] = None auth_service: str = "" @classmethod def from_proto(cls, proto: "api_user_team_pb2.UserAuth") -> "UserAuth": """Create a UserAuth from a protobuf message.""" return cls( auth_data=proto.auth_data if proto.HasField("auth_data") else None, auth_service=proto.auth_service, ) def to_proto(self) -> "api_user_team_pb2.UserAuth": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_user_team_pb2 proto = api_user_team_pb2.UserAuth( auth_service=self.auth_service, ) if self.auth_data is not None: proto.auth_data = self.auth_data return proto @dataclass(frozen=True) class Session: """ Represents a user session. Attributes: id: Unique session identifier. token: Session token. create_at: Unix timestamp when session was created. expires_at: Unix timestamp when session expires. last_activity_at: Unix timestamp of last activity. user_id: ID of the user who owns this session. device_id: Device ID if applicable. roles: Space-separated list of roles. is_oauth: Whether this is an OAuth session. props: Session properties. """ id: str token: str = "" create_at: int = 0 expires_at: int = 0 last_activity_at: int = 0 user_id: str = "" device_id: str = "" roles: str = "" is_oauth: bool = False expired_notify: bool = False props: Dict[str, str] = field(default_factory=dict) local: bool = False @classmethod def from_proto(cls, proto: "api_user_team_pb2.Session") -> "Session": """Create a Session from a protobuf message.""" return cls( id=proto.id, token=proto.token, create_at=proto.create_at, expires_at=proto.expires_at, last_activity_at=proto.last_activity_at, user_id=proto.user_id, device_id=proto.device_id, roles=proto.roles, is_oauth=proto.is_oauth, expired_notify=proto.expired_notify, props=dict(proto.props), local=proto.local, ) def to_proto(self) -> "api_user_team_pb2.Session": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_user_team_pb2 proto = api_user_team_pb2.Session( id=self.id, token=self.token, create_at=self.create_at, expires_at=self.expires_at, last_activity_at=self.last_activity_at, user_id=self.user_id, device_id=self.device_id, roles=self.roles, is_oauth=self.is_oauth, expired_notify=self.expired_notify, local=self.local, ) proto.props.update(self.props) return proto @dataclass(frozen=True) class UserAccessToken: """ Represents a user access token. Attributes: id: Unique token identifier. token: The actual token string. user_id: ID of the user who owns this token. description: Description of the token. is_active: Whether the token is active. """ id: str token: str = "" user_id: str = "" description: str = "" is_active: bool = True @classmethod def from_proto(cls, proto: "api_user_team_pb2.UserAccessToken") -> "UserAccessToken": """Create a UserAccessToken from a protobuf message.""" return cls( id=proto.id, token=proto.token, user_id=proto.user_id, description=proto.description, is_active=proto.is_active, ) def to_proto(self) -> "api_user_team_pb2.UserAccessToken": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_user_team_pb2 return api_user_team_pb2.UserAccessToken( id=self.id, token=self.token, user_id=self.user_id, description=self.description, is_active=self.is_active, ) # ============================================================================= # TEAM TYPES # ============================================================================= def _proto_team_type_to_str(proto_type: int) -> str: """Convert protobuf TeamType enum to string.""" from mattermost_plugin.grpc import team_pb2 if proto_type == team_pb2.TEAM_TYPE_OPEN: return "O" elif proto_type == team_pb2.TEAM_TYPE_INVITE: return "I" return "O" # Default to open def _str_team_type_to_proto(type_str: str) -> int: """Convert string team type to protobuf enum.""" from mattermost_plugin.grpc import team_pb2 if type_str == "I": return team_pb2.TEAM_TYPE_INVITE return team_pb2.TEAM_TYPE_OPEN @dataclass(frozen=True) class Team: """ Represents a Mattermost team. Attributes: id: Unique identifier for the team. create_at: Unix timestamp (milliseconds) when team was created. update_at: Unix timestamp (milliseconds) when team was last updated. delete_at: Unix timestamp (milliseconds) when team was deleted (0 if not deleted). display_name: Display name of the team. name: URL-safe name of the team. description: Team description. email: Team email address. type: Team type (O=open, I=invite only). company_name: Company name. allowed_domains: Comma-separated list of allowed email domains. invite_id: Invite ID for the team. allow_open_invite: Whether open invites are allowed. scheme_id: ID of the permissions scheme. group_constrained: Whether team is group-constrained. policy_id: Data retention policy ID. """ id: str display_name: str = "" name: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 description: str = "" email: str = "" type: str = "O" company_name: str = "" allowed_domains: str = "" invite_id: str = "" allow_open_invite: bool = False last_team_icon_update: int = 0 scheme_id: Optional[str] = None group_constrained: Optional[bool] = None policy_id: Optional[str] = None cloud_limits_archived: bool = False @classmethod def from_proto(cls, proto: "team_pb2.Team") -> "Team": """Create a Team from a protobuf message.""" return cls( id=proto.id, display_name=proto.display_name, name=proto.name, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, description=proto.description, email=proto.email, type=_proto_team_type_to_str(proto.type), company_name=proto.company_name, allowed_domains=proto.allowed_domains, invite_id=proto.invite_id, allow_open_invite=proto.allow_open_invite, last_team_icon_update=proto.last_team_icon_update, scheme_id=proto.scheme_id if proto.HasField("scheme_id") else None, group_constrained=proto.group_constrained if proto.HasField("group_constrained") else None, policy_id=proto.policy_id if proto.HasField("policy_id") else None, cloud_limits_archived=proto.cloud_limits_archived, ) def to_proto(self) -> "team_pb2.Team": """Convert to a protobuf message.""" from mattermost_plugin.grpc import team_pb2 proto = team_pb2.Team( id=self.id, display_name=self.display_name, name=self.name, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, description=self.description, email=self.email, type=_str_team_type_to_proto(self.type), company_name=self.company_name, allowed_domains=self.allowed_domains, invite_id=self.invite_id, allow_open_invite=self.allow_open_invite, last_team_icon_update=self.last_team_icon_update, cloud_limits_archived=self.cloud_limits_archived, ) if self.scheme_id is not None: proto.scheme_id = self.scheme_id if self.group_constrained is not None: proto.group_constrained = self.group_constrained if self.policy_id is not None: proto.policy_id = self.policy_id return proto @dataclass(frozen=True) class TeamMember: """ Represents a team membership. Attributes: team_id: ID of the team. user_id: ID of the user. roles: Space-separated list of roles. delete_at: Unix timestamp when membership was deleted (0 if not deleted). scheme_guest: Whether user is a guest through scheme. scheme_user: Whether user is a member through scheme. scheme_admin: Whether user is an admin through scheme. create_at: Unix timestamp when membership was created. """ team_id: str user_id: str roles: str = "" delete_at: int = 0 scheme_guest: bool = False scheme_user: bool = False scheme_admin: bool = False create_at: int = 0 @classmethod def from_proto(cls, proto: "team_pb2.TeamMember") -> "TeamMember": """Create a TeamMember from a protobuf message.""" return cls( team_id=proto.team_id, user_id=proto.user_id, roles=proto.roles, delete_at=proto.delete_at, scheme_guest=proto.scheme_guest, scheme_user=proto.scheme_user, scheme_admin=proto.scheme_admin, create_at=proto.create_at, ) @dataclass(frozen=True) class TeamMemberWithError: """ Represents a team membership result with potential error. Used for graceful batch operations where some may fail. Attributes: user_id: ID of the user. member: The team member if successful, None otherwise. error: Error message if failed, None otherwise. """ user_id: str member: Optional[TeamMember] = None error: Optional[str] = None @classmethod def from_proto(cls, proto: "api_user_team_pb2.TeamMemberWithError") -> "TeamMemberWithError": """Create a TeamMemberWithError from a protobuf message.""" member = None if proto.HasField("member"): member = TeamMember.from_proto(proto.member) error = None if proto.HasField("error") and proto.error.id: error = proto.error.message return cls( user_id=proto.user_id, member=member, error=error, ) @dataclass(frozen=True) class TeamUnread: """ Represents unread counts for a team. Attributes: team_id: ID of the team. msg_count: Total unread message count. mention_count: Mention count. mention_count_root: Root post mention count. msg_count_root: Root post message count. thread_count: Unread thread count. thread_mention_count: Thread mention count. """ team_id: str msg_count: int = 0 mention_count: int = 0 mention_count_root: int = 0 msg_count_root: int = 0 thread_count: int = 0 thread_mention_count: int = 0 @classmethod def from_proto(cls, proto: "team_pb2.TeamUnread") -> "TeamUnread": """Create a TeamUnread from a protobuf message.""" return cls( team_id=proto.team_id, msg_count=proto.msg_count, mention_count=proto.mention_count, mention_count_root=proto.mention_count_root, msg_count_root=proto.msg_count_root, thread_count=proto.thread_count, thread_mention_count=proto.thread_mention_count, ) @dataclass(frozen=True) class TeamStats: """ Represents team statistics. Attributes: team_id: ID of the team. total_member_count: Total number of members. active_member_count: Number of active members. """ team_id: str total_member_count: int = 0 active_member_count: int = 0 @classmethod def from_proto(cls, proto: "api_user_team_pb2.TeamStats") -> "TeamStats": """Create a TeamStats from a protobuf message.""" return cls( team_id=proto.team_id, total_member_count=proto.total_member_count, active_member_count=proto.active_member_count, ) # ============================================================================= # CHANNEL TYPES # ============================================================================= def _proto_channel_type_to_str(proto_type: int) -> str: """Convert protobuf ChannelType enum to string.""" from mattermost_plugin.grpc import channel_pb2 if proto_type == channel_pb2.CHANNEL_TYPE_OPEN: return "O" elif proto_type == channel_pb2.CHANNEL_TYPE_PRIVATE: return "P" elif proto_type == channel_pb2.CHANNEL_TYPE_DIRECT: return "D" elif proto_type == channel_pb2.CHANNEL_TYPE_GROUP: return "G" return "O" # Default to open def _str_channel_type_to_proto(type_str: str) -> int: """Convert string channel type to protobuf enum.""" from mattermost_plugin.grpc import channel_pb2 if type_str == "P": return channel_pb2.CHANNEL_TYPE_PRIVATE elif type_str == "D": return channel_pb2.CHANNEL_TYPE_DIRECT elif type_str == "G": return channel_pb2.CHANNEL_TYPE_GROUP return channel_pb2.CHANNEL_TYPE_OPEN @dataclass(frozen=True) class Channel: """ Represents a Mattermost channel. Attributes: id: Unique identifier for the channel. create_at: Unix timestamp (milliseconds) when channel was created. update_at: Unix timestamp (milliseconds) when channel was last updated. delete_at: Unix timestamp (milliseconds) when channel was deleted (0 if not deleted). team_id: ID of the team this channel belongs to. type: Channel type (O=open, P=private, D=direct, G=group). display_name: Display name of the channel. name: URL-safe name of the channel. header: Channel header text. purpose: Channel purpose text. last_post_at: Unix timestamp of last post. total_msg_count: Total message count. creator_id: ID of the user who created the channel. scheme_id: ID of the permissions scheme. group_constrained: Whether channel is group-constrained. """ id: str team_id: str = "" display_name: str = "" name: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 type: str = "O" header: str = "" purpose: str = "" last_post_at: int = 0 total_msg_count: int = 0 extra_update_at: int = 0 creator_id: str = "" scheme_id: Optional[str] = None props: Dict[str, object] = field(default_factory=dict) group_constrained: Optional[bool] = None auto_translation: bool = False shared: Optional[bool] = None total_msg_count_root: int = 0 policy_id: Optional[str] = None last_root_post_at: int = 0 policy_enforced: bool = False policy_is_active: bool = False default_category_name: str = "" @classmethod def from_proto(cls, proto: "channel_pb2.Channel") -> "Channel": """Create a Channel from a protobuf message.""" from google.protobuf.json_format import MessageToDict props = {} if proto.HasField("props"): props = MessageToDict(proto.props) return cls( id=proto.id, team_id=proto.team_id, display_name=proto.display_name, name=proto.name, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, type=_proto_channel_type_to_str(proto.type), header=proto.header, purpose=proto.purpose, last_post_at=proto.last_post_at, total_msg_count=proto.total_msg_count, extra_update_at=proto.extra_update_at, creator_id=proto.creator_id, scheme_id=proto.scheme_id if proto.HasField("scheme_id") else None, props=props, group_constrained=proto.group_constrained if proto.HasField("group_constrained") else None, auto_translation=proto.auto_translation, shared=proto.shared if proto.HasField("shared") else None, total_msg_count_root=proto.total_msg_count_root, policy_id=proto.policy_id if proto.HasField("policy_id") else None, last_root_post_at=proto.last_root_post_at, policy_enforced=proto.policy_enforced, policy_is_active=proto.policy_is_active, default_category_name=proto.default_category_name, ) def to_proto(self) -> "channel_pb2.Channel": """Convert to a protobuf message.""" from google.protobuf.json_format import ParseDict from google.protobuf.struct_pb2 import Struct from mattermost_plugin.grpc import channel_pb2 proto = channel_pb2.Channel( id=self.id, team_id=self.team_id, display_name=self.display_name, name=self.name, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, type=_str_channel_type_to_proto(self.type), header=self.header, purpose=self.purpose, last_post_at=self.last_post_at, total_msg_count=self.total_msg_count, extra_update_at=self.extra_update_at, creator_id=self.creator_id, auto_translation=self.auto_translation, total_msg_count_root=self.total_msg_count_root, last_root_post_at=self.last_root_post_at, policy_enforced=self.policy_enforced, policy_is_active=self.policy_is_active, default_category_name=self.default_category_name, ) if self.scheme_id is not None: proto.scheme_id = self.scheme_id if self.group_constrained is not None: proto.group_constrained = self.group_constrained if self.shared is not None: proto.shared = self.shared if self.policy_id is not None: proto.policy_id = self.policy_id if self.props: props_struct = Struct() ParseDict(self.props, props_struct) proto.props.CopyFrom(props_struct) return proto @dataclass(frozen=True) class ChannelMember: """ Represents a channel membership. Attributes: channel_id: ID of the channel. user_id: ID of the user. roles: Space-separated list of roles. last_viewed_at: Unix timestamp when user last viewed channel. msg_count: Message count at last view. mention_count: Unread mention count. mention_count_root: Root post mention count. msg_count_root: Root post message count. notify_props: Notification preferences. last_update_at: Unix timestamp of last update. scheme_guest: Whether user is a guest through scheme. scheme_user: Whether user is a member through scheme. scheme_admin: Whether user is an admin through scheme. urgent_mention_count: Urgent mention count. """ channel_id: str user_id: str roles: str = "" last_viewed_at: int = 0 msg_count: int = 0 mention_count: int = 0 mention_count_root: int = 0 msg_count_root: int = 0 notify_props: Dict[str, str] = field(default_factory=dict) last_update_at: int = 0 scheme_guest: bool = False scheme_user: bool = False scheme_admin: bool = False urgent_mention_count: int = 0 @classmethod def from_proto(cls, proto: "api_channel_post_pb2.ChannelMember") -> "ChannelMember": """Create a ChannelMember from a protobuf message.""" return cls( channel_id=proto.channel_id, user_id=proto.user_id, roles=proto.roles, last_viewed_at=proto.last_viewed_at, msg_count=proto.msg_count, mention_count=proto.mention_count, mention_count_root=proto.mention_count_root, msg_count_root=proto.msg_count_root, notify_props=dict(proto.notify_props), last_update_at=proto.last_update_at, scheme_guest=proto.scheme_guest, scheme_user=proto.scheme_user, scheme_admin=proto.scheme_admin, urgent_mention_count=proto.urgent_mention_count, ) @dataclass(frozen=True) class ChannelStats: """ Represents channel statistics. Attributes: channel_id: ID of the channel. member_count: Number of members. guest_count: Number of guests. pinnedpost_count: Number of pinned posts. files_count: Number of files. """ channel_id: str member_count: int = 0 guest_count: int = 0 pinnedpost_count: int = 0 files_count: int = 0 @classmethod def from_proto(cls, proto: "api_channel_post_pb2.ChannelStats") -> "ChannelStats": """Create a ChannelStats from a protobuf message.""" return cls( channel_id=proto.channel_id, member_count=proto.member_count, guest_count=proto.guest_count, pinnedpost_count=proto.pinnedpost_count, files_count=proto.files_count, ) @dataclass(frozen=True) class SidebarCategoryWithChannels: """ Represents a sidebar category with its channels. Attributes: id: Category ID. user_id: User ID who owns this category. team_id: Team ID this category belongs to. display_name: Display name of the category. type: Category type. sorting: Sorting preference. muted: Whether category is muted. collapsed: Whether category is collapsed. channel_ids: List of channel IDs in this category. """ id: str = "" user_id: str = "" team_id: str = "" display_name: str = "" type: str = "" sorting: int = 0 muted: bool = False collapsed: bool = False channel_ids: List[str] = field(default_factory=list) @classmethod def from_proto(cls, proto: "api_channel_post_pb2.SidebarCategoryWithChannels") -> "SidebarCategoryWithChannels": """Create a SidebarCategoryWithChannels from a protobuf message.""" return cls( id=proto.id, user_id=proto.user_id, team_id=proto.team_id, display_name=proto.display_name, type=proto.type, sorting=proto.sorting, muted=proto.muted, collapsed=proto.collapsed, channel_ids=list(proto.channel_ids), ) def to_proto(self) -> "api_channel_post_pb2.SidebarCategoryWithChannels": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_channel_post_pb2 return api_channel_post_pb2.SidebarCategoryWithChannels( id=self.id, user_id=self.user_id, team_id=self.team_id, display_name=self.display_name, type=self.type, sorting=self.sorting, muted=self.muted, collapsed=self.collapsed, channel_ids=self.channel_ids, ) @dataclass(frozen=True) class OrderedSidebarCategories: """ Represents ordered sidebar categories for a user in a team. Attributes: categories: List of sidebar categories with channels. order: Order of category IDs. """ categories: List[SidebarCategoryWithChannels] = field(default_factory=list) order: List[str] = field(default_factory=list) @classmethod def from_proto(cls, proto: "api_channel_post_pb2.OrderedSidebarCategories") -> "OrderedSidebarCategories": """Create an OrderedSidebarCategories from a protobuf message.""" return cls( categories=[SidebarCategoryWithChannels.from_proto(c) for c in proto.categories], order=list(proto.order), ) # ============================================================================= # VIEW RESTRICTIONS # ============================================================================= @dataclass(frozen=True) class ViewUsersRestrictions: """ Represents restrictions on which users can be viewed. Attributes: teams: List of team IDs to restrict to. channels: List of channel IDs to restrict to. """ teams: List[str] = field(default_factory=list) channels: List[str] = field(default_factory=list) def to_proto(self) -> "user_pb2.ViewUsersRestrictions": """Convert to a protobuf message.""" from mattermost_plugin.grpc import user_pb2 return user_pb2.ViewUsersRestrictions( teams=self.teams, channels=self.channels, ) # ============================================================================= # POST TYPES # ============================================================================= @dataclass(frozen=True) class Reaction: """ Represents a user's emoji reaction to a post. Attributes: user_id: ID of the user who reacted. post_id: ID of the post. emoji_name: Name of the emoji (without colons). create_at: Unix timestamp when reaction was created. update_at: Unix timestamp when reaction was updated. delete_at: Unix timestamp when reaction was deleted (0 if not deleted). remote_id: Remote cluster ID if from shared channel. channel_id: ID of the channel the post is in. """ user_id: str post_id: str emoji_name: str create_at: int = 0 update_at: int = 0 delete_at: int = 0 remote_id: Optional[str] = None channel_id: Optional[str] = None @classmethod def from_proto(cls, proto: "post_pb2.Reaction") -> "Reaction": """Create a Reaction from a protobuf message.""" return cls( user_id=proto.user_id, post_id=proto.post_id, emoji_name=proto.emoji_name, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, remote_id=proto.remote_id if proto.HasField("remote_id") else None, channel_id=proto.channel_id if proto.HasField("channel_id") else None, ) def to_proto(self) -> "post_pb2.Reaction": """Convert to a protobuf message.""" from mattermost_plugin.grpc import post_pb2 proto = post_pb2.Reaction( user_id=self.user_id, post_id=self.post_id, emoji_name=self.emoji_name, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, ) if self.remote_id is not None: proto.remote_id = self.remote_id if self.channel_id is not None: proto.channel_id = self.channel_id return proto @dataclass(frozen=True) class Post: """ Represents a message/post in Mattermost. Attributes: id: Unique identifier for the post (26-char ID). create_at: Timestamp when the post was created (milliseconds since epoch). update_at: Timestamp when the post was last updated. edit_at: Timestamp when the post was last edited (0 if never edited). delete_at: Timestamp when the post was deleted (0 if not deleted). is_pinned: Whether the post is pinned to the channel. user_id: ID of the user who created the post. channel_id: ID of the channel this post belongs to. root_id: ID of the root post if this is a reply (empty for root posts). original_id: Original post ID (used for cross-posting). message: The message content. message_source: Original message before server modifications. type: Post type (empty for normal posts, "system_*" for system messages). props: Post-specific properties. hashtags: Extracted hashtags from the message. file_ids: IDs of attached files. pending_post_id: Client-generated ID for deduplication. has_reactions: Whether this post has emoji reactions. remote_id: Remote cluster ID if from shared channel. reply_count: Number of replies (for root posts only). last_reply_at: Timestamp of the last reply. is_following: Whether the current user is following this thread. """ id: str channel_id: str user_id: str = "" message: str = "" create_at: int = 0 update_at: int = 0 edit_at: int = 0 delete_at: int = 0 is_pinned: bool = False root_id: str = "" original_id: str = "" message_source: str = "" type: str = "" props: Dict[str, object] = field(default_factory=dict) hashtags: str = "" file_ids: List[str] = field(default_factory=list) pending_post_id: str = "" has_reactions: bool = False remote_id: Optional[str] = None reply_count: int = 0 last_reply_at: int = 0 is_following: Optional[bool] = None @classmethod def from_proto(cls, proto: "post_pb2.Post") -> "Post": """Create a Post from a protobuf message.""" from google.protobuf.json_format import MessageToDict props = {} if proto.HasField("props"): props = MessageToDict(proto.props) return cls( id=proto.id, channel_id=proto.channel_id, user_id=proto.user_id, message=proto.message, create_at=proto.create_at, update_at=proto.update_at, edit_at=proto.edit_at, delete_at=proto.delete_at, is_pinned=proto.is_pinned, root_id=proto.root_id, original_id=proto.original_id, message_source=proto.message_source, type=proto.type, props=props, hashtags=proto.hashtags, file_ids=list(proto.file_ids), pending_post_id=proto.pending_post_id, has_reactions=proto.has_reactions, remote_id=proto.remote_id if proto.HasField("remote_id") else None, reply_count=proto.reply_count, last_reply_at=proto.last_reply_at, is_following=proto.is_following if proto.HasField("is_following") else None, ) def to_proto(self) -> "post_pb2.Post": """Convert to a protobuf message.""" from google.protobuf.json_format import ParseDict from google.protobuf.struct_pb2 import Struct from mattermost_plugin.grpc import post_pb2 proto = post_pb2.Post( id=self.id, channel_id=self.channel_id, user_id=self.user_id, message=self.message, create_at=self.create_at, update_at=self.update_at, edit_at=self.edit_at, delete_at=self.delete_at, is_pinned=self.is_pinned, root_id=self.root_id, original_id=self.original_id, message_source=self.message_source, type=self.type, hashtags=self.hashtags, pending_post_id=self.pending_post_id, has_reactions=self.has_reactions, reply_count=self.reply_count, last_reply_at=self.last_reply_at, ) # Set file_ids proto.file_ids.extend(self.file_ids) # Set optional fields if self.remote_id is not None: proto.remote_id = self.remote_id if self.is_following is not None: proto.is_following = self.is_following # Set props if self.props: props_struct = Struct() ParseDict(self.props, props_struct) proto.props.CopyFrom(props_struct) return proto @dataclass(frozen=True) class PostList: """ Represents a list of posts with ordering information. Attributes: order: Ordered list of post IDs. posts: Map of post ID to Post. next_post_id: The ID to use for fetching the next page. prev_post_id: The ID to use for fetching the previous page. has_next: Whether there are more posts to fetch. """ order: List[str] = field(default_factory=list) posts: Dict[str, Post] = field(default_factory=dict) next_post_id: str = "" prev_post_id: str = "" has_next: bool = False @classmethod def from_proto(cls, proto: "post_pb2.PostList") -> "PostList": """Create a PostList from a protobuf message.""" posts = {post_id: Post.from_proto(post) for post_id, post in proto.posts.items()} return cls( order=list(proto.order), posts=posts, next_post_id=proto.next_post_id, prev_post_id=proto.prev_post_id, has_next=proto.has_next, ) # ============================================================================= # FILE TYPES # ============================================================================= @dataclass(frozen=True) class FileInfo: """ Represents metadata about an uploaded file. Attributes: id: Unique identifier for the file (26-char ID). creator_id: ID of the user who uploaded the file. post_id: ID of the post this file is attached to. channel_id: ID of the channel this file is in. create_at: Timestamp when the file was created. update_at: Timestamp when the file was last updated. delete_at: Timestamp when the file was deleted (0 if not deleted). name: Original filename as uploaded. extension: File extension without the leading dot. size: File size in bytes. mime_type: MIME type of the file. width: Image width in pixels (0 for non-image files). height: Image height in pixels (0 for non-image files). has_preview_image: Whether a preview image was generated. mini_preview: Mini preview data (small thumbnail for images). remote_id: Remote cluster ID if from shared channel. archived: Whether the file content has been archived. """ id: str creator_id: str = "" post_id: str = "" channel_id: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 name: str = "" extension: str = "" size: int = 0 mime_type: str = "" width: int = 0 height: int = 0 has_preview_image: bool = False mini_preview: Optional[bytes] = None remote_id: Optional[str] = None archived: bool = False @classmethod def from_proto(cls, proto: "file_pb2.FileInfo") -> "FileInfo": """Create a FileInfo from a protobuf message.""" return cls( id=proto.id, creator_id=proto.creator_id, post_id=proto.post_id, channel_id=proto.channel_id, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, name=proto.name, extension=proto.extension, size=proto.size, mime_type=proto.mime_type, width=proto.width, height=proto.height, has_preview_image=proto.has_preview_image, mini_preview=proto.mini_preview if proto.HasField("mini_preview") else None, remote_id=proto.remote_id if proto.HasField("remote_id") else None, archived=proto.archived, ) @dataclass(frozen=True) class UploadSession: """ Represents an upload session for resumable uploads. Attributes: id: Unique identifier for the upload session. type: Upload type ("attachment" or "import"). create_at: Timestamp when the session was created. user_id: ID of the user who created the session. channel_id: ID of the channel (for attachment uploads). filename: Name of the file being uploaded. path: Server-side path where file will be stored. file_size: Total size of the file being uploaded. file_offset: Current offset (bytes already uploaded). remote_id: Remote cluster ID. req_file_id: Requested file ID (for deduplication). """ id: str = "" type: str = "attachment" create_at: int = 0 user_id: str = "" channel_id: str = "" filename: str = "" path: str = "" file_size: int = 0 file_offset: int = 0 remote_id: str = "" req_file_id: str = "" @classmethod def from_proto(cls, proto: "api_file_bot_pb2.UploadSession") -> "UploadSession": """Create an UploadSession from a protobuf message.""" return cls( id=proto.id, type=proto.type, create_at=proto.create_at, user_id=proto.user_id, channel_id=proto.channel_id, filename=proto.filename, path=proto.path, file_size=proto.file_size, file_offset=proto.file_offset, remote_id=proto.remote_id, req_file_id=proto.req_file_id, ) def to_proto(self) -> "api_file_bot_pb2.UploadSession": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_file_bot_pb2 return api_file_bot_pb2.UploadSession( id=self.id, type=self.type, create_at=self.create_at, user_id=self.user_id, channel_id=self.channel_id, filename=self.filename, path=self.path, file_size=self.file_size, file_offset=self.file_offset, remote_id=self.remote_id, req_file_id=self.req_file_id, ) # ============================================================================= # KV STORE TYPES # ============================================================================= @dataclass(frozen=True) class PluginKVSetOptions: """ Options for KVSetWithOptions. Attributes: atomic: Whether the operation should be atomic (compare-and-set). old_value: The expected current value for atomic operations. expire_in_seconds: Number of seconds until the key expires (0 = no expiry). """ atomic: bool = False old_value: Optional[bytes] = None expire_in_seconds: int = 0 def to_proto(self) -> "api_kv_config_pb2.PluginKVSetOptions": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_kv_config_pb2 return api_kv_config_pb2.PluginKVSetOptions( atomic=self.atomic, old_value=self.old_value or b"", expire_in_seconds=self.expire_in_seconds, ) # ============================================================================= # BOT TYPES # ============================================================================= @dataclass(frozen=True) class Bot: """ Represents a bot account. Attributes: user_id: The user ID of the bot. username: Bot username. display_name: Display name of the bot. description: Description of the bot. owner_id: ID of the user who owns the bot. create_at: Timestamp when the bot was created. update_at: Timestamp when the bot was last updated. delete_at: Timestamp when the bot was deleted (0 if not deleted). last_icon_update: Timestamp of the last icon update. """ user_id: str = "" username: str = "" display_name: str = "" description: str = "" owner_id: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 last_icon_update: int = 0 @classmethod def from_proto(cls, proto: "api_file_bot_pb2.Bot") -> "Bot": """Create a Bot from a protobuf message.""" return cls( user_id=proto.user_id, username=proto.username, display_name=proto.display_name, description=proto.description, owner_id=proto.owner_id, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, last_icon_update=proto.last_icon_update, ) def to_proto(self) -> "api_file_bot_pb2.Bot": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_file_bot_pb2 return api_file_bot_pb2.Bot( user_id=self.user_id, username=self.username, display_name=self.display_name, description=self.description, owner_id=self.owner_id, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, last_icon_update=self.last_icon_update, ) @dataclass(frozen=True) class BotPatch: """ Patch data for updating a bot. Attributes: username: New username (optional). display_name: New display name (optional). description: New description (optional). """ username: Optional[str] = None display_name: Optional[str] = None description: Optional[str] = None def to_proto(self) -> "api_file_bot_pb2.BotPatch": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_file_bot_pb2 proto = api_file_bot_pb2.BotPatch() if self.username is not None: proto.username = self.username if self.display_name is not None: proto.display_name = self.display_name if self.description is not None: proto.description = self.description return proto @dataclass(frozen=True) class BotGetOptions: """ Options for getting bots. Attributes: owner_id: Filter to bots owned by this user. include_deleted: Include deleted bots. only_orphaned: Only include orphaned bots (owner deleted). page: Page number (0-indexed). per_page: Results per page. """ owner_id: str = "" include_deleted: bool = False only_orphaned: bool = False page: int = 0 per_page: int = 60 def to_proto(self) -> "api_file_bot_pb2.BotGetOptions": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_file_bot_pb2 return api_file_bot_pb2.BotGetOptions( owner_id=self.owner_id, include_deleted=self.include_deleted, only_orphaned=self.only_orphaned, page=self.page, per_page=self.per_page, ) # ============================================================================= # COMMAND TYPES # ============================================================================= @dataclass(frozen=True) class Command: """ Represents a slash command. Attributes: id: Unique identifier for the command. token: Command token. create_at: Timestamp when the command was created. update_at: Timestamp when the command was last updated. delete_at: Timestamp when the command was deleted (0 if not deleted). creator_id: ID of the user who created the command. team_id: ID of the team this command belongs to. trigger: The trigger word that invokes this command. method: HTTP method ("P" for POST, "G" for GET). username: Username override for posts from this command. icon_url: Icon URL for posts from this command. auto_complete: Whether the command appears in autocomplete. auto_complete_desc: Description shown in autocomplete. auto_complete_hint: Hint shown in autocomplete. display_name: Display name for the command. description: Description of the command. url: URL to call when command is triggered. plugin_id: ID of the plugin that registered this command. """ id: str = "" token: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 creator_id: str = "" team_id: str = "" trigger: str = "" method: str = "P" username: str = "" icon_url: str = "" auto_complete: bool = False auto_complete_desc: str = "" auto_complete_hint: str = "" display_name: str = "" description: str = "" url: str = "" plugin_id: str = "" @classmethod def from_proto(cls, proto: "api_remaining_pb2.Command") -> "Command": """Create a Command from a protobuf message.""" return cls( id=proto.id, token=proto.token, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, creator_id=proto.creator_id, team_id=proto.team_id, trigger=proto.trigger, method=proto.method, username=proto.username, icon_url=proto.icon_url, auto_complete=proto.auto_complete, auto_complete_desc=proto.auto_complete_desc, auto_complete_hint=proto.auto_complete_hint, display_name=proto.display_name, description=proto.description, url=proto.url, plugin_id=proto.plugin_id, ) def to_proto(self) -> "api_remaining_pb2.Command": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.Command( id=self.id, token=self.token, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, creator_id=self.creator_id, team_id=self.team_id, trigger=self.trigger, method=self.method, username=self.username, icon_url=self.icon_url, auto_complete=self.auto_complete, auto_complete_desc=self.auto_complete_desc, auto_complete_hint=self.auto_complete_hint, display_name=self.display_name, description=self.description, url=self.url, plugin_id=self.plugin_id, ) @dataclass(frozen=True) class CommandArgs: """ Arguments for executing a slash command. Attributes: user_id: ID of the user executing the command. channel_id: ID of the channel where command was executed. team_id: ID of the team. root_id: Root post ID for threaded replies. parent_id: Parent post ID. trigger_id: Trigger ID for interactive dialogs. command: The full command string. site_url: Base URL of the Mattermost site. """ user_id: str = "" channel_id: str = "" team_id: str = "" root_id: str = "" parent_id: str = "" trigger_id: str = "" command: str = "" site_url: str = "" def to_proto(self) -> "api_remaining_pb2.CommandArgs": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.CommandArgs( user_id=self.user_id, channel_id=self.channel_id, team_id=self.team_id, root_id=self.root_id, parent_id=self.parent_id, trigger_id=self.trigger_id, command=self.command, site_url=self.site_url, ) @dataclass(frozen=True) class CommandResponse: """ Response from executing a slash command. Attributes: response_type: "in_channel" or "ephemeral". text: Response text. username: Username override. channel_id: Channel to post to. icon_url: Icon URL override. type: Response type. props: Additional properties. goto_location: URL to redirect to. trigger_id: Trigger ID for follow-up actions. skip_slack_parsing: Skip Slack-compatible parsing. """ response_type: str = "" text: str = "" username: str = "" channel_id: str = "" icon_url: str = "" type: str = "" props: Dict[str, object] = field(default_factory=dict) goto_location: str = "" trigger_id: str = "" skip_slack_parsing: bool = False @classmethod def from_proto(cls, proto: "api_remaining_pb2.CommandResponse") -> "CommandResponse": """Create a CommandResponse from a protobuf message.""" from google.protobuf.json_format import MessageToDict props = {} if proto.HasField("props"): props = MessageToDict(proto.props) return cls( response_type=proto.response_type, text=proto.text, username=proto.username, channel_id=proto.channel_id, icon_url=proto.icon_url, type=proto.type, props=props, goto_location=proto.goto_location, trigger_id=proto.trigger_id, skip_slack_parsing=proto.skip_slack_parsing, ) # ============================================================================= # PREFERENCE TYPES # ============================================================================= @dataclass(frozen=True) class Preference: """ Represents a user preference. Attributes: user_id: ID of the user. category: Category of the preference. name: Name of the preference within the category. value: Value of the preference. """ user_id: str = "" category: str = "" name: str = "" value: str = "" @classmethod def from_proto(cls, proto: "api_remaining_pb2.Preference") -> "Preference": """Create a Preference from a protobuf message.""" return cls( user_id=proto.user_id, category=proto.category, name=proto.name, value=proto.value, ) def to_proto(self) -> "api_remaining_pb2.Preference": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.Preference( user_id=self.user_id, category=self.category, name=self.name, value=self.value, ) # ============================================================================= # OAUTH TYPES # ============================================================================= @dataclass(frozen=True) class OAuthApp: """ Represents an OAuth application. Attributes: id: Unique identifier for the OAuth app. creator_id: ID of the user who created the app. create_at: Timestamp when the app was created. update_at: Timestamp when the app was last updated. client_secret: The client secret. name: Name of the app. description: Description of the app. icon_url: URL of the app's icon. callback_urls: Comma-separated list of callback URLs. homepage: Homepage URL. is_trusted: Whether the app is trusted. mattermost_app_id: Associated Mattermost app ID. """ id: str = "" creator_id: str = "" create_at: int = 0 update_at: int = 0 client_secret: str = "" name: str = "" description: str = "" icon_url: str = "" callback_urls: str = "" homepage: str = "" is_trusted: bool = False mattermost_app_id: str = "" @classmethod def from_proto(cls, proto: "api_remaining_pb2.OAuthApp") -> "OAuthApp": """Create an OAuthApp from a protobuf message.""" return cls( id=proto.id, creator_id=proto.creator_id, create_at=proto.create_at, update_at=proto.update_at, client_secret=proto.client_secret, name=proto.name, description=proto.description, icon_url=proto.icon_url, callback_urls=proto.callback_urls, homepage=proto.homepage, is_trusted=proto.is_trusted, mattermost_app_id=proto.mattermostAppId, ) def to_proto(self) -> "api_remaining_pb2.OAuthApp": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.OAuthApp( id=self.id, creator_id=self.creator_id, create_at=self.create_at, update_at=self.update_at, client_secret=self.client_secret, name=self.name, description=self.description, icon_url=self.icon_url, callback_urls=self.callback_urls, homepage=self.homepage, is_trusted=self.is_trusted, mattermostAppId=self.mattermost_app_id, ) # ============================================================================= # GROUP TYPES # ============================================================================= @dataclass(frozen=True) class Group: """ Represents a user group. Attributes: id: Unique identifier for the group. name: Unique name of the group. display_name: Display name of the group. description: Description of the group. source: Source of the group ("ldap", "custom"). remote_id: Remote ID (for LDAP groups). create_at: Timestamp when the group was created. update_at: Timestamp when the group was last updated. delete_at: Timestamp when the group was deleted (0 if not deleted). allow_reference: Whether the group can be @mentioned. """ id: str = "" name: str = "" display_name: str = "" description: str = "" source: str = "" remote_id: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 allow_reference: bool = False @classmethod def from_proto(cls, proto: "api_remaining_pb2.Group") -> "Group": """Create a Group from a protobuf message.""" return cls( id=proto.id, name=proto.name, display_name=proto.display_name, description=proto.description, source=proto.source, remote_id=proto.remote_id, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, allow_reference=proto.allow_reference, ) def to_proto(self) -> "api_remaining_pb2.Group": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.Group( id=self.id, name=self.name, display_name=self.display_name, description=self.description, source=self.source, remote_id=self.remote_id, create_at=self.create_at, update_at=self.update_at, delete_at=self.delete_at, allow_reference=self.allow_reference, ) @dataclass(frozen=True) class GroupMember: """ Represents membership in a group. Attributes: group_id: ID of the group. user_id: ID of the user. create_at: Timestamp when the membership was created. delete_at: Timestamp when the membership was deleted (0 if not deleted). """ group_id: str = "" user_id: str = "" create_at: int = 0 delete_at: int = 0 @classmethod def from_proto(cls, proto: "api_remaining_pb2.GroupMember") -> "GroupMember": """Create a GroupMember from a protobuf message.""" return cls( group_id=proto.group_id, user_id=proto.user_id, create_at=proto.create_at, delete_at=proto.delete_at, ) @dataclass(frozen=True) class GroupSyncable: """ Represents a group's sync to a team or channel. Attributes: group_id: ID of the group. syncable_id: ID of the team or channel. syncable_type: "team" or "channel". auto_add: Whether to auto-add group members. scheme_admin: Whether group members are scheme admins. create_at: Timestamp when the syncable was created. delete_at: Timestamp when the syncable was deleted (0 if not deleted). update_at: Timestamp when the syncable was last updated. """ group_id: str = "" syncable_id: str = "" syncable_type: str = "" auto_add: bool = False scheme_admin: bool = False create_at: int = 0 delete_at: int = 0 update_at: int = 0 @classmethod def from_proto(cls, proto: "api_remaining_pb2.GroupSyncable") -> "GroupSyncable": """Create a GroupSyncable from a protobuf message.""" return cls( group_id=proto.group_id, syncable_id=proto.syncable_id, syncable_type=proto.syncable_type, auto_add=proto.auto_add, scheme_admin=proto.scheme_admin, create_at=proto.create_at, delete_at=proto.delete_at, update_at=proto.update_at, ) def to_proto(self) -> "api_remaining_pb2.GroupSyncable": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.GroupSyncable( group_id=self.group_id, syncable_id=self.syncable_id, syncable_type=self.syncable_type, auto_add=self.auto_add, scheme_admin=self.scheme_admin, create_at=self.create_at, delete_at=self.delete_at, update_at=self.update_at, ) # ============================================================================= # SHARED CHANNEL TYPES # ============================================================================= @dataclass(frozen=True) class SharedChannel: """ Represents a shared channel. Attributes: channel_id: ID of the channel. team_id: ID of the team. home: Whether this is the home instance. read_only: Whether the channel is read-only. share_name: Name for the share. share_display_name: Display name for the share. share_purpose: Purpose for the share. share_header: Header for the share. creator_id: ID of the user who shared the channel. create_at: Timestamp when the channel was shared. update_at: Timestamp when the share was last updated. remote_id: Remote cluster ID. type: Channel type. """ channel_id: str = "" team_id: str = "" home: str = "" read_only: bool = False share_name: str = "" share_display_name: str = "" share_purpose: str = "" share_header: str = "" creator_id: str = "" create_at: int = 0 update_at: int = 0 remote_id: str = "" type: str = "" @classmethod def from_proto(cls, proto: "api_remaining_pb2.SharedChannel") -> "SharedChannel": """Create a SharedChannel from a protobuf message.""" return cls( channel_id=proto.channel_id, team_id=proto.team_id, home=proto.home, read_only=proto.read_only, share_name=proto.share_name, share_display_name=proto.share_display_name, share_purpose=proto.share_purpose, share_header=proto.share_header, creator_id=proto.creator_id, create_at=proto.create_at, update_at=proto.update_at, remote_id=proto.remote_id, type=proto.type, ) def to_proto(self) -> "api_remaining_pb2.SharedChannel": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.SharedChannel( channel_id=self.channel_id, team_id=self.team_id, home=self.home, read_only=self.read_only, share_name=self.share_name, share_display_name=self.share_display_name, share_purpose=self.share_purpose, share_header=self.share_header, creator_id=self.creator_id, create_at=self.create_at, update_at=self.update_at, remote_id=self.remote_id, type=self.type, ) # ============================================================================= # EMOJI TYPES # ============================================================================= @dataclass(frozen=True) class Emoji: """ Represents a custom emoji. Attributes: id: Unique identifier for the emoji. create_at: Timestamp when the emoji was created. update_at: Timestamp when the emoji was last updated. delete_at: Timestamp when the emoji was deleted (0 if not deleted). creator_id: ID of the user who created the emoji. name: Name of the emoji (without colons). """ id: str = "" create_at: int = 0 update_at: int = 0 delete_at: int = 0 creator_id: str = "" name: str = "" @classmethod def from_proto(cls, proto: "api_channel_post_pb2.Emoji") -> "Emoji": """Create an Emoji from a protobuf message.""" return cls( id=proto.id, create_at=proto.create_at, update_at=proto.update_at, delete_at=proto.delete_at, creator_id=proto.creator_id, name=proto.name, ) # ============================================================================= # PLUGIN TYPES # ============================================================================= @dataclass(frozen=True) class PluginInfo: """ Represents plugin information. Attributes: manifest: Plugin manifest data. """ manifest: Dict[str, object] = field(default_factory=dict) @dataclass(frozen=True) class PluginStatus: """ Represents plugin status. Attributes: plugin_id: ID of the plugin. plugin_path: Path to the plugin. state: Plugin state. name: Plugin name. description: Plugin description. version: Plugin version. """ plugin_id: str = "" plugin_path: str = "" state: int = 0 name: str = "" description: str = "" version: str = "" # ============================================================================= # DIALOG TYPES # ============================================================================= @dataclass(frozen=True) class DialogElement: """ Represents an element in an interactive dialog. Attributes: display_name: Display name of the element. name: Internal name of the element. type: Element type (e.g., "text", "select"). subtype: Subtype (e.g., "email", "number"). default: Default value. placeholder: Placeholder text. help_text: Help text shown below the element. optional: Whether the element is optional. min_length: Minimum text length. max_length: Maximum text length. data_source: Data source for select elements. options: Options for select elements. """ display_name: str = "" name: str = "" type: str = "" subtype: str = "" default: str = "" placeholder: str = "" help_text: str = "" optional: bool = False min_length: int = 0 max_length: int = 0 data_source: str = "" options: List[Dict[str, str]] = field(default_factory=list) def to_proto(self) -> "api_remaining_pb2.DialogElement": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 proto_options = [ api_remaining_pb2.DialogElementOption(text=o.get("text", ""), value=o.get("value", "")) for o in self.options ] return api_remaining_pb2.DialogElement( display_name=self.display_name, name=self.name, type=self.type, subtype=self.subtype, default=self.default, placeholder=self.placeholder, help_text=self.help_text, optional=self.optional, min_length=self.min_length, max_length=self.max_length, data_source=self.data_source, options=proto_options, ) @dataclass(frozen=True) class Dialog: """ Represents an interactive dialog. Attributes: callback_id: Callback ID for identifying the dialog submission. title: Title of the dialog. introduction_text: Introduction text shown at the top. elements: List of dialog elements. submit_label: Label for the submit button. notify_on_cancel: Whether to notify on cancel. state: State to pass back on submission. icon_url: URL of the dialog icon. """ callback_id: str = "" title: str = "" introduction_text: str = "" elements: List[DialogElement] = field(default_factory=list) submit_label: str = "" notify_on_cancel: bool = False state: str = "" icon_url: str = "" def to_proto(self) -> "api_remaining_pb2.Dialog": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.Dialog( callback_id=self.callback_id, title=self.title, introduction_text=self.introduction_text, elements=[e.to_proto() for e in self.elements], submit_label=self.submit_label, notify_on_cancel=self.notify_on_cancel, state=self.state, icon_url=self.icon_url, ) @dataclass(frozen=True) class OpenDialogRequest: """ Request to open an interactive dialog. Attributes: trigger_id: Trigger ID from the initiating action. url: URL to submit the dialog to. dialog: The dialog to open. """ trigger_id: str = "" url: str = "" dialog: Optional[Dialog] = None def to_proto(self) -> "api_remaining_pb2.OpenDialogRequest": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 proto = api_remaining_pb2.OpenDialogRequest( trigger_id=self.trigger_id, url=self.url, ) if self.dialog: proto.dialog.CopyFrom(self.dialog.to_proto()) return proto # ============================================================================= # AUDIT TYPES # ============================================================================= @dataclass(frozen=True) class AuditRecord: """ Represents an audit log record. Attributes: id: Unique identifier for the record. create_at: Timestamp when the record was created. level: Log level (e.g., "info", "error"). api_path: API path that was called. event: Event name. status: Status of the event. user_id: ID of the user. session_id: ID of the session. client: Client information. ip_address: IP address of the client. meta: Additional metadata. """ id: str = "" create_at: int = 0 level: str = "" api_path: str = "" event: str = "" status: str = "" user_id: str = "" session_id: str = "" client: str = "" ip_address: str = "" meta: Dict[str, object] = field(default_factory=dict) def to_proto(self) -> "api_remaining_pb2.AuditRecord": """Convert to a protobuf message.""" from google.protobuf.json_format import ParseDict from google.protobuf.struct_pb2 import Struct from mattermost_plugin.grpc import api_remaining_pb2 proto = api_remaining_pb2.AuditRecord( id=self.id, create_at=self.create_at, level=self.level, api_path=self.api_path, event=self.event, status=self.status, user_id=self.user_id, session_id=self.session_id, client=self.client, ip_address=self.ip_address, ) if self.meta: meta_struct = Struct() ParseDict(self.meta, meta_struct) proto.meta.CopyFrom(meta_struct) return proto # ============================================================================= # PUSH NOTIFICATION TYPES # ============================================================================= @dataclass(frozen=True) class PushNotification: """ Represents a push notification. Attributes: ack_id: Acknowledgement ID. platform: Platform (e.g., "ios", "android"). server_id: Server ID. device_id: Device ID. post_id: Post ID. category: Notification category. sound: Notification sound. message: Notification message. badge: Badge count. team_id: Team ID. channel_id: Channel ID. root_id: Root post ID. sender_id: Sender user ID. sender_name: Sender name. channel_name: Channel name. type: Notification type. """ ack_id: str = "" platform: str = "" server_id: str = "" device_id: str = "" post_id: str = "" category: str = "" sound: str = "" message: str = "" badge: str = "" team_id: str = "" channel_id: str = "" root_id: str = "" sender_id: str = "" sender_name: str = "" channel_name: str = "" type: str = "" def to_proto(self) -> "api_remaining_pb2.PushNotification": """Convert to a protobuf message.""" from mattermost_plugin.grpc import api_remaining_pb2 return api_remaining_pb2.PushNotification( ack_id=self.ack_id, platform=self.platform, server_id=self.server_id, device_id=self.device_id, post_id=self.post_id, category=self.category, sound=self.sound, message=self.message, badge=self.badge, team_id=self.team_id, channel_id=self.channel_id, root_id=self.root_id, sender_id=self.sender_id, sender_name=self.sender_name, channel_name=self.channel_name, type=self.type, )