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
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)

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
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)

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
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
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
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
Generator[dict[str, Any], None, None]

Yield the user's info.

Source code in module/mattermost_bot.py
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
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
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
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
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_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
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
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
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
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
@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
@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
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
@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
@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