upstream/mattermost
1
0
Fork 0
mirror of https://github.com/mattermost/mattermost.git synced 2026-02-03 20:40:00 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions 2
mattermost/python-sdk/build/lib/mattermost_plugin/_internal/wrappers.py

2459 lines
78 KiB
Python
Raw Permalink Normal View History

debug: add extensive logging to trace hook registration flow Go side: - Log hooks returned by Implemented() - Log each hook name -> ID mapping - Log OnActivate implementation status - Log OnActivate call flow Python side: - Log Implemented() return value - Log OnActivate gRPC receipt and handler invocation This is temporary debug logging to diagnose why OnActivate isn't being called for Python plugins. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-20 09:12:22 -05:00
# 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,
)
Reference in a new issue Copy permalink