Skip to content

Mattermost

module/mattermost_bot.py

module.mattermost_bot

MattermostBot

MattermostBot

Bases: Session

MattermostBot

Parameters:

Name Type Description Default
token str

API token.

 required
base_url str

API base url.

 required
log_name str

Log name.

 'MattermostBot'
Note

The headers will update the Authorization in Bearer {self.token}.

Source code in module/mattermost_bot.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class MattermostBot(Session):
    ''' MattermostBot

    Args:
        token (str): API token.
        base_url (str): API base url.
        log_name (str): Log name.

    Note:
        The `headers` will update the `Authorization` in `Bearer {self.token}`.

    '''

    def __init__(self, token: str, base_url: str, log_name: str = 'MattermostBot') -> None:
        super().__init__()
        self.token = token
        self.base_url = base_url
        self.log = logging.getLogger(log_name)
        self.headers.update({'Authorization': f'Bearer {self.token}'})

    def log_rate_limit(self, headers: dict[str, Any]) -> None:
        ''' Get log info from headers

        Args:
            headers (dict): [requests.Response.headers][].

        '''
        self.log.info('X-Ratelimit-Limit: %s, X-Ratelimit-Remaining: %s, X-Ratelimit-Reset: %s',
                      headers.get('X-Ratelimit-Limit'),
                      headers.get('X-Ratelimit-Remaining'),
                      headers.get('X-Ratelimit-Reset'),
                      )

    def get_users(self, page: int, per_page: int = 200) -> Response:
        ''' Get users

        Args:
            page (int): Page.
            per_page (int): Numbers per page.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.get(f'{self.base_url}/users', params={'page': page, 'per_page': per_page})

    def get_users_loop(self, per_page: int = 200) -> Generator[dict[str, Any], None, None]:
        ''' Get users in loop

        Args:
            per_page (int): Numbers per page.

        Yields:
            Yield the user's info.

        '''
        page = 0
        num = 0
        for user in self.get_users(page=page, per_page=per_page).json():
            yield user
            num += 1

        while num == per_page:
            page += 1
            num = 0
            for user in self.get_users(page=page, per_page=per_page).json():
                yield user
                num += 1

    def get_users_stats(self) -> Response:
        ''' Get users stats

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.get(f'{self.base_url}/users/stats')

    def get_user_by_username(self, username: str) -> Response:
        ''' Get user by username

        Args:
            username (str): Username.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.get(f'{self.base_url}/users/username/{username}')

    def create_a_direct_message(self,
                                users: Union[list[str], tuple[str, str]]) -> Response:
        ''' Create a direct messge

        Args:
            users (list | tuple): Two uids in list or tuple.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.base_url}/channels/direct', json=users)

    def posts(self, channel_id: str, message: str) -> Response:
        ''' Posts message

        Args:
            channel_id (str): Channel id.
            message (str): Messages.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.base_url}/posts',
                         json={'channel_id': channel_id, 'message': message})

    def get_posts_from_channel(self, channel_id: str) -> Response:
        ''' Get post from channel

        Args:
            channel_id (str): Channel id.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.get(f'{self.base_url}/channels/{channel_id}/posts')

    def post_invite_by_email(self, team_id: str, emails: list[str]) -> Response:
        ''' Post an invite by email

        Args:
            team_id (str): Team id.
            emails (list): Email addresses.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.base_url}/teams/{team_id}/invite/email', json=emails)

    def post_invite_guests_by_email(self, team_id: str, emails: list[str],
                                    channels: list[str], message: Optional[str] = None) -> Response:
        ''' Post an invite to guest by email

        Args:
            team_id (str): Team id.
            emails (list): Email addresses.
            channels (list): Channel ids.
            message (str): Messages.

        Returns:
            Return the [requests.Response][] object.

        '''
        data: dict[str, Any] = {
            'emails': emails,
            'channels': channels,
        }
        if message:
            data['message'] = message.strip()

        return self.post(f'{self.base_url}/teams/{team_id}/invite-guests/email', json=data)

    def post_user_to_channel(self, channel_id: str, uid: str) -> Response:
        ''' Post user to channel

        Args:
            channel_id (str): Channel id.
            uid (str): User id.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.base_url}/channels/{channel_id}/members', json={'user_id': uid})

    def put_users_patch(self, uid: str, position: str) -> Response:
        ''' Update user

        Args:
            uid (str): User id.
            position (str): Position title.

        Returns:
            Return the [requests.Response][] object.

        '''
        data = {
            'position': position,
        }
        return self.put(f'{self.base_url}/users/{uid}/patch', json=data)

    def post_revoke_all_sessions_for_a_user(self, uid: str) -> Response:
        ''' Revoke all sessions for a user

        Args:
            uid (str): User id.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.post(f'{self.base_url}/users/{uid}/sessions/revoke/all')

    def del_deactivate_user(self, uid: str) -> Response:
        ''' Deactivate user account

        Args:
            uid (str): User id.

        Returns:
            Return the [requests.Response][] object.

        '''
        return self.delete(f'{self.base_url}/users/{uid}')

create_a_direct_message 

create_a_direct_message(
    users: Union[list[str], tuple[str, str]]
) -> Response

Create a direct messge

Parameters:

Name Type Description Default
users list | tuple

Two uids in list or tuple.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
104
105
106
107
108
109
110
111
112
113
114
115
def create_a_direct_message(self,
                            users: Union[list[str], tuple[str, str]]) -> Response:
    ''' Create a direct messge

    Args:
        users (list | tuple): Two uids in list or tuple.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.base_url}/channels/direct', json=users)

del_deactivate_user 

del_deactivate_user(uid: str) -> Response

Deactivate user account

Parameters:

Name Type Description Default
uid str

User id.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
220
221
222
223
224
225
226
227
228
229
230
def del_deactivate_user(self, uid: str) -> Response:
    ''' Deactivate user account

    Args:
        uid (str): User id.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.delete(f'{self.base_url}/users/{uid}')

get_posts_from_channel 

get_posts_from_channel(channel_id: str) -> Response

Get post from channel

Parameters:

Name Type Description Default
channel_id str

Channel id.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
131
132
133
134
135
136
137
138
139
140
141
def get_posts_from_channel(self, channel_id: str) -> Response:
    ''' Get post from channel

    Args:
        channel_id (str): Channel id.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.get(f'{self.base_url}/channels/{channel_id}/posts')

get_user_by_username 

get_user_by_username(username: str) -> Response

Get user by username

Parameters:

Name Type Description Default
username str

Username.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_user_by_username(self, username: str) -> Response:
    ''' Get user by username

    Args:
        username (str): Username.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.get(f'{self.base_url}/users/username/{username}')

get_users 

get_users(page: int, per_page: int = 200) -> Response

Get users

Parameters:

Name Type Description Default
page int

Page.

 required
per_page int

Numbers per page.

 200

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
47
48
49
50
51
52
53
54
55
56
57
58
def get_users(self, page: int, per_page: int = 200) -> Response:
    ''' Get users

    Args:
        page (int): Page.
        per_page (int): Numbers per page.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.get(f'{self.base_url}/users', params={'page': page, 'per_page': per_page})

get_users_loop 

get_users_loop(
    per_page: int = 200,
) -> Generator[dict[str, Any], None, None]

Get users in loop

Parameters:

Name Type Description Default
per_page int

Numbers per page.

 200

Yields:

Type Description
dict[str, Any]

Yield the user's info.
Source code in module/mattermost_bot.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def get_users_loop(self, per_page: int = 200) -> Generator[dict[str, Any], None, None]:
    ''' Get users in loop

    Args:
        per_page (int): Numbers per page.

    Yields:
        Yield the user's info.

    '''
    page = 0
    num = 0
    for user in self.get_users(page=page, per_page=per_page).json():
        yield user
        num += 1

    while num == per_page:
        page += 1
        num = 0
        for user in self.get_users(page=page, per_page=per_page).json():
            yield user
            num += 1

get_users_stats 

get_users_stats() -> Response

Get users stats

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
83
84
85
86
87
88
89
90
def get_users_stats(self) -> Response:
    ''' Get users stats

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.get(f'{self.base_url}/users/stats')

log_rate_limit 

log_rate_limit(headers: dict[str, Any]) -> None

Get log info from headers

Parameters:

Name Type Description Default
headers dict

requests.Response.headers.

 required
Source code in module/mattermost_bot.py 
34
35
36
37
38
39
40
41
42
43
44
45
def log_rate_limit(self, headers: dict[str, Any]) -> None:
    ''' Get log info from headers

    Args:
        headers (dict): [requests.Response.headers][].

    '''
    self.log.info('X-Ratelimit-Limit: %s, X-Ratelimit-Remaining: %s, X-Ratelimit-Reset: %s',
                  headers.get('X-Ratelimit-Limit'),
                  headers.get('X-Ratelimit-Remaining'),
                  headers.get('X-Ratelimit-Reset'),
                  )

post_invite_by_email 

post_invite_by_email(
    team_id: str, emails: list[str]
) -> Response

Post an invite by email

Parameters:

Name Type Description Default
team_id str

Team id.

 required
emails list

Email addresses.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
143
144
145
146
147
148
149
150
151
152
153
154
def post_invite_by_email(self, team_id: str, emails: list[str]) -> Response:
    ''' Post an invite by email

    Args:
        team_id (str): Team id.
        emails (list): Email addresses.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.base_url}/teams/{team_id}/invite/email', json=emails)

post_invite_guests_by_email 

post_invite_guests_by_email(
    team_id: str,
    emails: list[str],
    channels: list[str],
    message: Optional[str] = None,
) -> Response

Post an invite to guest by email

Parameters:

Name Type Description Default
team_id str

Team id.

 required
emails list

Email addresses.

 required
channels list

Channel ids.

 required
message str

Messages.

 None

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def post_invite_guests_by_email(self, team_id: str, emails: list[str],
                                channels: list[str], message: Optional[str] = None) -> Response:
    ''' Post an invite to guest by email

    Args:
        team_id (str): Team id.
        emails (list): Email addresses.
        channels (list): Channel ids.
        message (str): Messages.

    Returns:
        Return the [requests.Response][] object.

    '''
    data: dict[str, Any] = {
        'emails': emails,
        'channels': channels,
    }
    if message:
        data['message'] = message.strip()

    return self.post(f'{self.base_url}/teams/{team_id}/invite-guests/email', json=data)

post_revoke_all_sessions_for_a_user 

post_revoke_all_sessions_for_a_user(uid: str) -> Response

Revoke all sessions for a user

Parameters:

Name Type Description Default
uid str

User id.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
208
209
210
211
212
213
214
215
216
217
218
def post_revoke_all_sessions_for_a_user(self, uid: str) -> Response:
    ''' Revoke all sessions for a user

    Args:
        uid (str): User id.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.base_url}/users/{uid}/sessions/revoke/all')

post_user_to_channel 

post_user_to_channel(channel_id: str, uid: str) -> Response

Post user to channel

Parameters:

Name Type Description Default
channel_id str

Channel id.

 required
uid str

User id.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
179
180
181
182
183
184
185
186
187
188
189
190
def post_user_to_channel(self, channel_id: str, uid: str) -> Response:
    ''' Post user to channel

    Args:
        channel_id (str): Channel id.
        uid (str): User id.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.base_url}/channels/{channel_id}/members', json={'user_id': uid})

posts 

posts(channel_id: str, message: str) -> Response

Posts message

Parameters:

Name Type Description Default
channel_id str

Channel id.

 required
message str

Messages.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
def posts(self, channel_id: str, message: str) -> Response:
    ''' Posts message

    Args:
        channel_id (str): Channel id.
        message (str): Messages.

    Returns:
        Return the [requests.Response][] object.

    '''
    return self.post(f'{self.base_url}/posts',
                     json={'channel_id': channel_id, 'message': message})

put_users_patch 

put_users_patch(uid: str, position: str) -> Response

Update user

Parameters:

Name Type Description Default
uid str

User id.

 required
position str

Position title.

 required

Returns:

Type Description
Response

Return the requests.Response object.
Source code in module/mattermost_bot.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def put_users_patch(self, uid: str, position: str) -> Response:
    ''' Update user

    Args:
        uid (str): User id.
        position (str): Position title.

    Returns:
        Return the [requests.Response][] object.

    '''
    data = {
        'position': position,
    }
    return self.put(f'{self.base_url}/users/{uid}/patch', json=data)

MattermostTools

Bases: MattermostBot

MattermostTools for more implement in operation

Parameters:

Name Type Description Default
token str

API token.

 required
base_url str

The API base url.

 required
Source code in module/mattermost_bot.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
class MattermostTools(MattermostBot):
    ''' MattermostTools for more implement in operation

    Args:
        token (str): API token.
        base_url (str): The API base url.

    '''

    def __init__(self, token: str, base_url: str) -> None:
        super().__init__(token=token, base_url=base_url)

    @staticmethod
    def find_possible_mid(uid: str, mail: Optional[str] = None) -> str:
        ''' Find any possible mattermost user id

        Args:
            uid (str): User id.
            mail (str): User email address.

        Returns:
            Return the user's mattermost id or `''`.

        '''
        mml = MattermostLink(uid)
        if not mml:
            return ''

        if mml.raw and 'data' in mml.raw and 'user_id' in mml.raw['data']:
            return str(mml.raw['data']['user_id'])

        if mail is None:
            oauth = OAuthDB().find_one({'owner': uid}, {'_id': 1})
            if oauth:
                mail = oauth['_id']

        if mail:
            mm_user = MattermostUsersDB().find_one(
                {'email': mail.strip()}, {'_id': 1})
            if mm_user:
                return str(mm_user['_id'])

        return ''

    @staticmethod
    def find_user_name(mid: str) -> str:
        ''' Find user_name by mid

        Args:
            mid (str): Mattermost user id.

        Returns:
            Return the user's mattermost user name or `''`.

        '''
        mm_user = MattermostUsersDB().find_one({'_id': mid}, {'username': 1})
        if mm_user:
            return str(mm_user['username'])

        mattermost_link = MattermostLinkDB().find_one(
            {'data.user_id': mid}, {'data.user_name': 1})
        if mattermost_link and 'data' in mattermost_link:
            return str(mattermost_link['data']['user_name'])

        return ''

find_possible_mid staticmethod 

find_possible_mid(
    uid: str, mail: Optional[str] = None
) -> str

Find any possible mattermost user id

Parameters:

Name Type Description Default
uid str

User id.

 required
mail str

User email address.

 None

Returns:

Type Description
str

Return the user's mattermost id or ''.
Source code in module/mattermost_bot.py 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@staticmethod
def find_possible_mid(uid: str, mail: Optional[str] = None) -> str:
    ''' Find any possible mattermost user id

    Args:
        uid (str): User id.
        mail (str): User email address.

    Returns:
        Return the user's mattermost id or `''`.

    '''
    mml = MattermostLink(uid)
    if not mml:
        return ''

    if mml.raw and 'data' in mml.raw and 'user_id' in mml.raw['data']:
        return str(mml.raw['data']['user_id'])

    if mail is None:
        oauth = OAuthDB().find_one({'owner': uid}, {'_id': 1})
        if oauth:
            mail = oauth['_id']

    if mail:
        mm_user = MattermostUsersDB().find_one(
            {'email': mail.strip()}, {'_id': 1})
        if mm_user:
            return str(mm_user['_id'])

    return ''

find_user_name staticmethod 

find_user_name(mid: str) -> str

Find user_name by mid

Parameters:

Name Type Description Default
mid str

Mattermost user id.

 required

Returns:

Type Description
str

Return the user's mattermost user name or ''.
Source code in module/mattermost_bot.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
@staticmethod
def find_user_name(mid: str) -> str:
    ''' Find user_name by mid

    Args:
        mid (str): Mattermost user id.

    Returns:
        Return the user's mattermost user name or `''`.

    '''
    mm_user = MattermostUsersDB().find_one({'_id': mid}, {'username': 1})
    if mm_user:
        return str(mm_user['username'])

    mattermost_link = MattermostLinkDB().find_one(
        {'data.user_id': mid}, {'data.user_name': 1})
    if mattermost_link and 'data' in mattermost_link:
        return str(mattermost_link['data']['user_name'])

    return ''

module/mattermost_link.py

MattermostLink

MattermostLink

Parameters:

Name Type Description Default
uid str

User id.

 required
Source code in module/mattermost_link.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class MattermostLink:
    ''' MattermostLink

    Args:
        uid (str): User id.

    '''

    def __init__(self, uid: str) -> None:
        self.uid = uid
        self.raw = MattermostLinkDB().find_one({'_id': uid})

        if not self.raw and uid:
            mml_db = MattermostLinkDB()
            mml_db.insert_one({'_id': uid, 'code': uuid4().hex})
            self.raw = mml_db.find_one({'_id': uid})

    @classmethod
    def verify_save(cls, data: dict[str, Any]) -> bool:
        ''' verify and save data

        Args:
            data (dict): Check the `token`, and verify the user's code.

        Returns:
            `true` or `false`.

        '''
        if 'token' in data and data['token'] == setting.MATTERMOST_SLASH_VOLUNTEER:
            if 'text' in data and data['text']:
                texts = data['text'].split(' ')
                if len(texts) < 2:
                    return False

                cmd, pwd = texts

                if cmd == 'verify':
                    uid, code = pwd.split('.')
                    mml = cls(uid=uid)
                    if mml.raw and code == mml.raw['code']:
                        MattermostLinkDB().find_one_and_update(
                            {'_id': uid}, {'$set': {'data': data, 'create_at': time()}})
                        return True

        return False

    @staticmethod
    def reset(uid: str) -> None:
        ''' Reset the link

        Args:
            uid (str): User id.

        '''
        MattermostLinkDB().delete_one({'_id': uid})

reset staticmethod 

reset(uid: str) -> None

Reset the link

Parameters:

Name Type Description Default
uid str

User id.

 required
Source code in module/mattermost_link.py 
56
57
58
59
60
61
62
63
64
@staticmethod
def reset(uid: str) -> None:
    ''' Reset the link

    Args:
        uid (str): User id.

    '''
    MattermostLinkDB().delete_one({'_id': uid})

verify_save classmethod 

verify_save(data: dict[str, Any]) -> bool

verify and save data

Parameters:

Name Type Description Default
data dict

Check the token, and verify the user's code.

 required

Returns:

Type Description
bool

true or false.
Source code in module/mattermost_link.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@classmethod
def verify_save(cls, data: dict[str, Any]) -> bool:
    ''' verify and save data

    Args:
        data (dict): Check the `token`, and verify the user's code.

    Returns:
        `true` or `false`.

    '''
    if 'token' in data and data['token'] == setting.MATTERMOST_SLASH_VOLUNTEER:
        if 'text' in data and data['text']:
            texts = data['text'].split(' ')
            if len(texts) < 2:
                return False

            cmd, pwd = texts

            if cmd == 'verify':
                uid, code = pwd.split('.')
                mml = cls(uid=uid)
                if mml.raw and code == mml.raw['code']:
                    MattermostLinkDB().find_one_and_update(
                        {'_id': uid}, {'$set': {'data': data, 'create_at': time()}})
                    return True

    return False