Skip to content

module/sender.py

module.sender

Sender

SenderCampaign

SenderCampaign class

Source code in module/sender.py 
 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
class SenderCampaign:
    ''' SenderCampaign class '''

    @staticmethod
    def create(name: str, pid: str, tid: str, uid: str) -> dict[str, Any]:
        ''' Create new campaign

        Args:
            name (str): Campaign name.
            pid (str): Project id.
            tid (str): Team id.
            uid (str): User id.

        Returns:
            Return the added data.

        '''

        data = SenderCampaignDB.new(
            name=name.strip(), pid=pid, tid=tid, uid=uid)
        return SenderCampaignDB().add(data)

    @staticmethod
    def get(cid: str, pid: Optional[str] = None,
            tid: Optional[str] = None) -> Optional[dict[str, Any]]:
        ''' Get campaign

        Args:
            cid (str): Campaign id.
            pid (str): Project id.
            tid (str): Team id.

        Returns:
            Return the campaign data.

        '''
        query = {'_id': cid}

        if pid is not None:
            query['created.pid'] = pid

        if tid is not None:
            query['created.tid'] = tid

        return SenderCampaignDB().find_one(query)

    @staticmethod
    def get_list(pid: str, tid: str) -> Cursor[dict[str, Any]]:
        ''' Get list campaign

        Args:
            pid (str): Project id.
            tid (str): Team id.

        Returns:
            Return the [pymongo.cursor.Cursor][].

        '''
        return SenderCampaignDB().find({'created.pid': pid, 'created.tid': tid})

    @staticmethod
    def save_mail(cid: str, subject: str, content: str,
                  preheader: str, layout: Literal['1', '2']) -> dict[str, Any]:
        ''' Save mail data

        Args:
            cid (str): Campaign id.
            subject (str): Mail subject.
            content (str): Mail content.
            preheader (str): The preheader in mail.
            layout (str): In which layout. `1`: In Volunteer layout. `2`: In COSCUP layout.

        Returns:
            Return the added / updated data.

        '''
        return SenderCampaignDB().find_one_and_update(
            {'_id': cid},
            {'$set': {
                'mail.subject': subject,
                'mail.content': content,
                'mail.preheader': preheader,
                'mail.layout': layout,
            }},
            return_document=ReturnDocument.AFTER,
        )

    @staticmethod
    def save_receiver(cid: str, teams: list[str], team_w_tags: dict[str, list[str]],
                      users: Optional[dict[str, Any]] = None,
                      all_users: bool = False) -> dict[str, Any]:
        ''' Save receiver

        Args:
            cid (str): Campaign id.
            teams (list): List of `tid`.
            team_w_tags (dict): With team's internal tags. `{'<tid>': [<tag>, ...]}`.
            users (list): List of user/mail. `{'name': <username>, 'mail': <mail>}`.
                          **(not in completed implement)**
            all_users (bool): All volunteers in platform.

        '''
        update: dict[str, Any] = {'receiver.teams': teams}
        update['receiver.users'] = users if users else []
        update['receiver.all_users'] = all_users
        update['receiver.team_w_tags'] = team_w_tags

        return SenderCampaignDB().find_one_and_update(
            {'_id': cid},
            {'$set': update},
            return_document=ReturnDocument.AFTER,
        )

create staticmethod 

create(
    name: str, pid: str, tid: str, uid: str
) -> dict[str, Any]

Create new campaign

Parameters:

Name Type Description Default
name str

Campaign name.

 required
pid str

Project id.

 required
tid str

Team id.

 required
uid str

User id.

 required

Returns:

Type Description
dict[str, Any]

Return the added data.
Source code in module/sender.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@staticmethod
def create(name: str, pid: str, tid: str, uid: str) -> dict[str, Any]:
    ''' Create new campaign

    Args:
        name (str): Campaign name.
        pid (str): Project id.
        tid (str): Team id.
        uid (str): User id.

    Returns:
        Return the added data.

    '''

    data = SenderCampaignDB.new(
        name=name.strip(), pid=pid, tid=tid, uid=uid)
    return SenderCampaignDB().add(data)

get staticmethod 

get(
    cid: str,
    pid: Optional[str] = None,
    tid: Optional[str] = None,
) -> Optional[dict[str, Any]]

Get campaign

Parameters:

Name Type Description Default
cid str

Campaign id.

 required
pid str

Project id.

 None
tid str

Team id.

 None

Returns:

Type Description
Optional[dict[str, Any]]

Return the campaign data.
Source code in module/sender.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
def get(cid: str, pid: Optional[str] = None,
        tid: Optional[str] = None) -> Optional[dict[str, Any]]:
    ''' Get campaign

    Args:
        cid (str): Campaign id.
        pid (str): Project id.
        tid (str): Team id.

    Returns:
        Return the campaign data.

    '''
    query = {'_id': cid}

    if pid is not None:
        query['created.pid'] = pid

    if tid is not None:
        query['created.tid'] = tid

    return SenderCampaignDB().find_one(query)

get_list staticmethod 

get_list(pid: str, tid: str) -> Cursor[dict[str, Any]]

Get list campaign

Parameters:

Name Type Description Default
pid str

Project id.

 required
tid str

Team id.

 required

Returns:

Type Description
Cursor[dict[str, Any]]

Return the pymongo.cursor.Cursor.
Source code in module/sender.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
@staticmethod
def get_list(pid: str, tid: str) -> Cursor[dict[str, Any]]:
    ''' Get list campaign

    Args:
        pid (str): Project id.
        tid (str): Team id.

    Returns:
        Return the [pymongo.cursor.Cursor][].

    '''
    return SenderCampaignDB().find({'created.pid': pid, 'created.tid': tid})

save_mail staticmethod 

save_mail(
    cid: str,
    subject: str,
    content: str,
    preheader: str,
    layout: Literal["1", "2"],
) -> dict[str, Any]

Save mail data

Parameters:

Name Type Description Default
cid str

Campaign id.

 required
subject str

Mail subject.

 required
content str

Mail content.

 required
preheader str

The preheader in mail.

 required
layout str

In which layout. 1: In Volunteer layout. 2: In COSCUP layout.

 required

Returns:

Type Description
dict[str, Any]

Return the added / updated data.
Source code in module/sender.py 
 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
@staticmethod
def save_mail(cid: str, subject: str, content: str,
              preheader: str, layout: Literal['1', '2']) -> dict[str, Any]:
    ''' Save mail data

    Args:
        cid (str): Campaign id.
        subject (str): Mail subject.
        content (str): Mail content.
        preheader (str): The preheader in mail.
        layout (str): In which layout. `1`: In Volunteer layout. `2`: In COSCUP layout.

    Returns:
        Return the added / updated data.

    '''
    return SenderCampaignDB().find_one_and_update(
        {'_id': cid},
        {'$set': {
            'mail.subject': subject,
            'mail.content': content,
            'mail.preheader': preheader,
            'mail.layout': layout,
        }},
        return_document=ReturnDocument.AFTER,
    )

save_receiver staticmethod 

save_receiver(
    cid: str,
    teams: list[str],
    team_w_tags: dict[str, list[str]],
    users: Optional[dict[str, Any]] = None,
    all_users: bool = False,
) -> dict[str, Any]

Save receiver

Parameters:

Name Type Description Default
cid str

Campaign id.

 required
teams list

List of tid.

 required
team_w_tags dict

With team's internal tags. {'<tid>': [<tag>, ...]}.

 required
users list

List of user/mail. {'name': <username>, 'mail': <mail>}. (not in completed implement)

 None
all_users bool

All volunteers in platform.

 False
Source code in module/sender.py 
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
@staticmethod
def save_receiver(cid: str, teams: list[str], team_w_tags: dict[str, list[str]],
                  users: Optional[dict[str, Any]] = None,
                  all_users: bool = False) -> dict[str, Any]:
    ''' Save receiver

    Args:
        cid (str): Campaign id.
        teams (list): List of `tid`.
        team_w_tags (dict): With team's internal tags. `{'<tid>': [<tag>, ...]}`.
        users (list): List of user/mail. `{'name': <username>, 'mail': <mail>}`.
                      **(not in completed implement)**
        all_users (bool): All volunteers in platform.

    '''
    update: dict[str, Any] = {'receiver.teams': teams}
    update['receiver.users'] = users if users else []
    update['receiver.all_users'] = all_users
    update['receiver.team_w_tags'] = team_w_tags

    return SenderCampaignDB().find_one_and_update(
        {'_id': cid},
        {'$set': update},
        return_document=ReturnDocument.AFTER,
    )

SenderLogs

SenderLogs object

Source code in module/sender.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class SenderLogs:
    ''' SenderLogs object '''

    @staticmethod
    def save(cid: str, layout: str, desc: str, receivers: list[dict[str, Any]]) -> None:
        ''' save log

        :param str cid: cid
        :param str layout: layout
        :param str desc: desc
        :param list receivers: receivers

        '''
        SenderLogsDB().add(cid=cid, layout=layout, desc=desc, receivers=receivers)

    @staticmethod
    def get(cid: str) -> Generator[dict[str, Any], None, None]:
        ''' Get log

        :param str cid: cid

        '''
        for raw in SenderLogsDB().find({'cid': cid}, sort=(('_id', -1), )):
            yield raw

get staticmethod 

get(cid: str) -> Generator[dict[str, Any], None, None]

Get log

:param str cid: cid

Source code in module/sender.py 
229
230
231
232
233
234
235
236
237
@staticmethod
def get(cid: str) -> Generator[dict[str, Any], None, None]:
    ''' Get log

    :param str cid: cid

    '''
    for raw in SenderLogsDB().find({'cid': cid}, sort=(('_id', -1), )):
        yield raw

save staticmethod 

save(
    cid: str,
    layout: str,
    desc: str,
    receivers: list[dict[str, Any]],
) -> None

save log

:param str cid: cid :param str layout: layout :param str desc: desc :param list receivers: receivers

Source code in module/sender.py 
217
218
219
220
221
222
223
224
225
226
227
@staticmethod
def save(cid: str, layout: str, desc: str, receivers: list[dict[str, Any]]) -> None:
    ''' save log

    :param str cid: cid
    :param str layout: layout
    :param str desc: desc
    :param list receivers: receivers

    '''
    SenderLogsDB().add(cid=cid, layout=layout, desc=desc, receivers=receivers)

SenderMailer

Sender Mailer

Parameters:

Name Type Description Default
template_path str

Mail template path.

 required
subject str

Mail subject.

 required
source dict

Mail FROM, {'name': <name>, 'mail': <mail>}.

 None
Source code in module/sender.py 
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
class SenderMailer:
    ''' Sender Mailer

    Args:
        template_path (str): Mail template path.
        subject (str): Mail subject.
        source (dict): Mail `FROM`, `{'name': <name>, 'mail': <mail>}`.

    '''

    def __init__(self, template_path: str, subject: str,
                 content: dict[str, Any], source: Optional[dict[str, str]] = None) -> None:
        with open(template_path, 'r', encoding='UTF8') as files:
            body = SandboxedEnvironment().from_string(files.read()).render(**content)

            self.tpl = SandboxedEnvironment().from_string(body)
            self.subject = SandboxedEnvironment().from_string(subject)

            if source is None:
                source = setting.AWS_SES_FROM

            if 'text_body' in content and content['text_body']:
                self.text_body = SandboxedEnvironment(
                ).from_string(content['text_body'])

            self.awsses = AWSSES(aws_access_key_id=setting.AWS_ID,
                                 aws_secret_access_key=setting.AWS_KEY, source=source)

    def send(self, to_list: list[dict[str, str]],
             data: dict[str, Any], x_coscup: Optional[str] = None) -> Any:
        ''' Send mail

        Args:
            to_list (list): List of user/mail in `{'name': <name>, 'mail': <mail>}`.
            data (dict): Mail datas. More details: [module.awsses.AWSSES.raw_mail][].

        Returns:
            More details: [module.awsses.AWSSES.send_raw_email][].

        '''
        raw_mail = self.awsses.raw_mail(
            to_addresses=to_list,
            subject=self.subject.render(**data),
            body=self.tpl.render(**data),
            text_body=self.text_body.render(**data),
            x_coscup=x_coscup,
            list_unsubscribe=setting.AWS_LIST_UNSUBSCRIBE,
        )
        return self.awsses.send_raw_email(data=raw_mail)

send 

send(
    to_list: list[dict[str, str]],
    data: dict[str, Any],
    x_coscup: Optional[str] = None,
) -> Any

Send mail

Parameters:

Name Type Description Default
to_list list

List of user/mail in {'name': <name>, 'mail': <mail>}.

 required
data dict

Mail datas. More details: module.awsses.AWSSES.raw_mail.

 required

Returns:

Type Description
Any

More details: module.awsses.AWSSES.send_raw_email.
Source code in module/sender.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def send(self, to_list: list[dict[str, str]],
         data: dict[str, Any], x_coscup: Optional[str] = None) -> Any:
    ''' Send mail

    Args:
        to_list (list): List of user/mail in `{'name': <name>, 'mail': <mail>}`.
        data (dict): Mail datas. More details: [module.awsses.AWSSES.raw_mail][].

    Returns:
        More details: [module.awsses.AWSSES.send_raw_email][].

    '''
    raw_mail = self.awsses.raw_mail(
        to_addresses=to_list,
        subject=self.subject.render(**data),
        body=self.tpl.render(**data),
        text_body=self.text_body.render(**data),
        x_coscup=x_coscup,
        list_unsubscribe=setting.AWS_LIST_UNSUBSCRIBE,
    )
    return self.awsses.send_raw_email(data=raw_mail)

SenderMailerCOSCUP

Bases: SenderMailer

Sender using COSCUP template

Parameters:

Name Type Description Default
subject str

Mail subject.

 required
source dict

Mail FROM, {'name': <name>, 'mail': <mail>}.

 None
Source code in module/sender.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
class SenderMailerCOSCUP(SenderMailer):
    ''' Sender using COSCUP template

    Args:
        subject (str): Mail subject.
        source (dict): Mail `FROM`, `{'name': <name>, 'mail': <mail>}`.

    '''

    def __init__(self, subject: str, content: dict[str, Any],
                 source: Optional[dict[str, str]] = None):
        super().__init__(
            template_path='/app/templates/mail/coscup_base.html',
            subject=subject, content=content, source=source)

SenderMailerVolunteer

Bases: SenderMailer

Sender using volunteer template

Parameters:

Name Type Description Default
subject str

Mail subject.

 required
source dict

Mail FROM, {'name': <name>, 'mail': <mail>}.

 None
Source code in module/sender.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
class SenderMailerVolunteer(SenderMailer):
    ''' Sender using volunteer template

    Args:
        subject (str): Mail subject.
        source (dict): Mail `FROM`, `{'name': <name>, 'mail': <mail>}`.

    '''

    def __init__(self, subject: str, content: dict[str, Any],
                 source: Optional[dict[str, str]] = None) -> None:
        super().__init__(
            template_path='/app/templates/mail/sender_base.html',
            subject=subject, content=content, source=source)

SenderReceiver

SenderReceiver object

Source code in module/sender.py 
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
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
class SenderReceiver:
    ''' SenderReceiver object '''

    @staticmethod
    def replace(pid: str, cid: str, datas: list[dict[str, Any]]) -> int:
        ''' Replace

        Args:
            pid (str): Project id.
            cid (str): Campaign id.
            datas (list): List of receiver datas. `name`, `mail` are required
                or include `uid` for auto replace the `name`, `mail` value.

        '''
        sender_receiver_db = SenderReceiverDB()
        sender_receiver_db.remove_past(pid=pid, cid=cid)

        uids = []
        for data in datas:
            if 'uid' in data and data['uid']:
                uids.append(data['uid'])

        user_infos = User.get_info(uids=uids)
        user_info_uids = {}
        for uid, data in user_infos.items():
            user_info_uids[uid] = {
                'name': data['profile']['badge_name'],
                'mail': data['oauth']['email'],
            }

        save_datas = []
        for data in datas:
            if 'uid' in data and data['uid'] and data['uid'] in user_info_uids:
                _data = SenderReceiverDB.new(pid=pid, cid=cid,
                                             name=user_info_uids[data['uid']
                                                                 ]['name'],
                                             mail=user_info_uids[data['uid']
                                                                 ]['mail'],
                                             )
            else:
                _data = SenderReceiverDB.new(
                    pid=pid, cid=cid, name=data['name'], mail=data['mail'])

            _data['data'].update(data)
            save_datas.append(_data)

        sender_receiver_db.update_data(pid=pid, cid=cid, datas=save_datas)

        return len(save_datas)

    @staticmethod
    def update(pid: str, cid: str, datas: list[dict[str, Any]]) -> int:
        ''' Update

        Args:
            pid (str): Project id.
            cid (str): Campaign id.
            datas (list): List of receiver datas. `name`, `mail` are required
                or include `uid` for auto replace the `name`, `mail` value.

        '''
        uids = []
        for data in datas:
            if 'uid' in data and data['uid']:
                uids.append(data['uid'])

        user_infos = User.get_info(uids=uids)
        user_info_uids = {}
        for uid, data in user_infos.items():
            user_info_uids[uid] = {
                'name': data['profile']['badge_name'],
                'mail': data['oauth']['email'],
            }

        save_datas = []
        for data in datas:
            if 'uid' in data and data['uid'] and data['uid'] in user_info_uids:
                _data = SenderReceiverDB.new(pid=pid, cid=cid,
                                             name=user_info_uids[data['uid']
                                                                 ]['name'],
                                             mail=user_info_uids[data['uid']
                                                                 ]['mail'],
                                             )
            else:
                _data = SenderReceiverDB.new(
                    pid=pid, cid=cid, name=data['name'], mail=data['mail'])

            _data['data'].update(data)
            save_datas.append(_data)

        SenderReceiverDB().update_data(pid=pid, cid=cid, datas=save_datas)

        return len(save_datas)

    @staticmethod
    def remove(pid: str, cid: str) -> int:
        ''' Update

        Args:
            pid (str): Project id.
            cid (str): Campaign id.

        '''
        return SenderReceiverDB().remove_past(pid=pid, cid=cid)

    @staticmethod
    def get(pid: str, cid: str) -> tuple[list[str], list[list[str]]]:
        ''' Get

        Args:
            pid (str): Project id.
            cid (str): Campaign id.

        Returns:
            Return a tuple with all fields at first. The second is the all datas.

        '''
        datas = list(SenderReceiverDB().find({'pid': pid, 'cid': cid}))
        fields = ['name', 'mail']
        for data in datas:
            for k in data['data']:
                if k not in ('name', 'mail') and k not in fields:
                    fields.append(k)

        raws = []
        for data in datas:
            raw = []
            for field in fields:
                raw.append(data['data'].get(field, ''))

            raws.append(raw)

        return fields, raws

    @staticmethod
    def get_from_user(pid: str,
                      tids: Union[str, list[str]]) -> tuple[tuple[str, str], list[list[str]]]:
        ''' Get users from userdb by project, team

        Args:
            pid (str): Project id.
            tids (list): List of `tid`.

        Returns:
            Return a tuple with `('name', 'mail')` at first. The second is the all datas.

        '''
        _tids: list[str]

        if isinstance(tids, str):
            _tids = [tids, ]
        else:
            _tids = tids

        team_users = Team.get_users(pid=pid, tids=_tids)
        uids = []
        for user_ids in team_users.values():
            uids.extend(user_ids)

        user_infos = User.get_info(uids=uids)
        datas = []
        for value in user_infos.values():
            # append, plus more data here in the future
            datas.append({
                'name': value['profile']['badge_name'],
                'mail': value['oauth']['email'],
            })

        raws = []
        for data in datas:
            raw = []
            for field in ('name', 'mail'):
                raw.append(data[field])

            raws.append(raw)

        return (('name', 'mail'), raws)

    @staticmethod
    def get_all_users() -> tuple[tuple[str, str], list[tuple[str, str]]]:
        ''' Get all users

        Returns:
            Return a tuple with `('name', 'mail')` at first. The second is the all datas.

        '''
        uids = []
        for user in User.get_all_users():
            uids.append(user['_id'])

        user_infos = User.get_info(uids=uids)
        raws = []
        for value in user_infos.values():
            raws.append((
                value['profile']['badge_name'],
                value['oauth']['email'],
            ))

        return (('name', 'mail'), raws)

    @staticmethod
    def get_by_tags(pid: str, tid: str,
                    tags: list[str]) -> tuple[tuple[str, str], list[tuple[str, str]]]:
        ''' Get users by tags

        Returns:
            Return a tuple with `('name', 'mail')` at first. The second is the all datas.

        '''
        uids = Team.get_members_uid_by_tags(pid=pid, tid=tid, tags=tags)

        user_infos = User.get_info(uids=uids)
        raws = []
        for value in user_infos.values():
            raws.append((
                value['profile']['badge_name'],
                value['oauth']['email'],
            ))

        return (('name', 'mail'), raws)

get staticmethod 

get(
    pid: str, cid: str
) -> tuple[list[str], list[list[str]]]

Get

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id.

 required

Returns:

Type Description
tuple[list[str], list[list[str]]]

Return a tuple with all fields at first. The second is the all datas.
Source code in module/sender.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
@staticmethod
def get(pid: str, cid: str) -> tuple[list[str], list[list[str]]]:
    ''' Get

    Args:
        pid (str): Project id.
        cid (str): Campaign id.

    Returns:
        Return a tuple with all fields at first. The second is the all datas.

    '''
    datas = list(SenderReceiverDB().find({'pid': pid, 'cid': cid}))
    fields = ['name', 'mail']
    for data in datas:
        for k in data['data']:
            if k not in ('name', 'mail') and k not in fields:
                fields.append(k)

    raws = []
    for data in datas:
        raw = []
        for field in fields:
            raw.append(data['data'].get(field, ''))

        raws.append(raw)

    return fields, raws

get_all_users staticmethod 

get_all_users() -> (
    tuple[tuple[str, str], list[tuple[str, str]]]
)

Get all users

Returns:

Type Description
tuple[tuple[str, str], list[tuple[str, str]]]

Return a tuple with ('name', 'mail') at first. The second is the all datas.
Source code in module/sender.py 
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
@staticmethod
def get_all_users() -> tuple[tuple[str, str], list[tuple[str, str]]]:
    ''' Get all users

    Returns:
        Return a tuple with `('name', 'mail')` at first. The second is the all datas.

    '''
    uids = []
    for user in User.get_all_users():
        uids.append(user['_id'])

    user_infos = User.get_info(uids=uids)
    raws = []
    for value in user_infos.values():
        raws.append((
            value['profile']['badge_name'],
            value['oauth']['email'],
        ))

    return (('name', 'mail'), raws)

get_by_tags staticmethod 

get_by_tags(
    pid: str, tid: str, tags: list[str]
) -> tuple[tuple[str, str], list[tuple[str, str]]]

Get users by tags

Returns:

Type Description
tuple[tuple[str, str], list[tuple[str, str]]]

Return a tuple with ('name', 'mail') at first. The second is the all datas.
Source code in module/sender.py 
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
@staticmethod
def get_by_tags(pid: str, tid: str,
                tags: list[str]) -> tuple[tuple[str, str], list[tuple[str, str]]]:
    ''' Get users by tags

    Returns:
        Return a tuple with `('name', 'mail')` at first. The second is the all datas.

    '''
    uids = Team.get_members_uid_by_tags(pid=pid, tid=tid, tags=tags)

    user_infos = User.get_info(uids=uids)
    raws = []
    for value in user_infos.values():
        raws.append((
            value['profile']['badge_name'],
            value['oauth']['email'],
        ))

    return (('name', 'mail'), raws)

get_from_user staticmethod 

get_from_user(
    pid: str, tids: Union[str, list[str]]
) -> tuple[tuple[str, str], list[list[str]]]

Get users from userdb by project, team

Parameters:

Name Type Description Default
pid str

Project id.

 required
tids list

List of tid.

 required

Returns:

Type Description
tuple[tuple[str, str], list[list[str]]]

Return a tuple with ('name', 'mail') at first. The second is the all datas.
Source code in module/sender.py 
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
@staticmethod
def get_from_user(pid: str,
                  tids: Union[str, list[str]]) -> tuple[tuple[str, str], list[list[str]]]:
    ''' Get users from userdb by project, team

    Args:
        pid (str): Project id.
        tids (list): List of `tid`.

    Returns:
        Return a tuple with `('name', 'mail')` at first. The second is the all datas.

    '''
    _tids: list[str]

    if isinstance(tids, str):
        _tids = [tids, ]
    else:
        _tids = tids

    team_users = Team.get_users(pid=pid, tids=_tids)
    uids = []
    for user_ids in team_users.values():
        uids.extend(user_ids)

    user_infos = User.get_info(uids=uids)
    datas = []
    for value in user_infos.values():
        # append, plus more data here in the future
        datas.append({
            'name': value['profile']['badge_name'],
            'mail': value['oauth']['email'],
        })

    raws = []
    for data in datas:
        raw = []
        for field in ('name', 'mail'):
            raw.append(data[field])

        raws.append(raw)

    return (('name', 'mail'), raws)

remove staticmethod 

remove(pid: str, cid: str) -> int

Update

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id.

 required
Source code in module/sender.py 
351
352
353
354
355
356
357
358
359
360
@staticmethod
def remove(pid: str, cid: str) -> int:
    ''' Update

    Args:
        pid (str): Project id.
        cid (str): Campaign id.

    '''
    return SenderReceiverDB().remove_past(pid=pid, cid=cid)

replace staticmethod 

replace(
    pid: str, cid: str, datas: list[dict[str, Any]]
) -> int

Replace

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id.

 required
datas list

List of receiver datas. name, mail are required or include uid for auto replace the name, mail value.

 required
Source code in module/sender.py 
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
298
299
300
301
302
303
304
305
@staticmethod
def replace(pid: str, cid: str, datas: list[dict[str, Any]]) -> int:
    ''' Replace

    Args:
        pid (str): Project id.
        cid (str): Campaign id.
        datas (list): List of receiver datas. `name`, `mail` are required
            or include `uid` for auto replace the `name`, `mail` value.

    '''
    sender_receiver_db = SenderReceiverDB()
    sender_receiver_db.remove_past(pid=pid, cid=cid)

    uids = []
    for data in datas:
        if 'uid' in data and data['uid']:
            uids.append(data['uid'])

    user_infos = User.get_info(uids=uids)
    user_info_uids = {}
    for uid, data in user_infos.items():
        user_info_uids[uid] = {
            'name': data['profile']['badge_name'],
            'mail': data['oauth']['email'],
        }

    save_datas = []
    for data in datas:
        if 'uid' in data and data['uid'] and data['uid'] in user_info_uids:
            _data = SenderReceiverDB.new(pid=pid, cid=cid,
                                         name=user_info_uids[data['uid']
                                                             ]['name'],
                                         mail=user_info_uids[data['uid']
                                                             ]['mail'],
                                         )
        else:
            _data = SenderReceiverDB.new(
                pid=pid, cid=cid, name=data['name'], mail=data['mail'])

        _data['data'].update(data)
        save_datas.append(_data)

    sender_receiver_db.update_data(pid=pid, cid=cid, datas=save_datas)

    return len(save_datas)

update staticmethod 

update(
    pid: str, cid: str, datas: list[dict[str, Any]]
) -> int

Update

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id.

 required
datas list

List of receiver datas. name, mail are required or include uid for auto replace the name, mail value.

 required
Source code in module/sender.py 
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
@staticmethod
def update(pid: str, cid: str, datas: list[dict[str, Any]]) -> int:
    ''' Update

    Args:
        pid (str): Project id.
        cid (str): Campaign id.
        datas (list): List of receiver datas. `name`, `mail` are required
            or include `uid` for auto replace the `name`, `mail` value.

    '''
    uids = []
    for data in datas:
        if 'uid' in data and data['uid']:
            uids.append(data['uid'])

    user_infos = User.get_info(uids=uids)
    user_info_uids = {}
    for uid, data in user_infos.items():
        user_info_uids[uid] = {
            'name': data['profile']['badge_name'],
            'mail': data['oauth']['email'],
        }

    save_datas = []
    for data in datas:
        if 'uid' in data and data['uid'] and data['uid'] in user_info_uids:
            _data = SenderReceiverDB.new(pid=pid, cid=cid,
                                         name=user_info_uids[data['uid']
                                                             ]['name'],
                                         mail=user_info_uids[data['uid']
                                                             ]['mail'],
                                         )
        else:
            _data = SenderReceiverDB.new(
                pid=pid, cid=cid, name=data['name'], mail=data['mail'])

        _data['data'].update(data)
        save_datas.append(_data)

    SenderReceiverDB().update_data(pid=pid, cid=cid, datas=save_datas)

    return len(save_datas)

SenderSESLogs

SenderSESLogs

Source code in module/sender.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class SenderSESLogs:
    ''' SenderSESLogs '''

    @staticmethod
    def save(cid: str, name: str, mail: str, result: dict[str, Any]) -> None:
        ''' Save log

        Args:
            cid (str): Campaign id.
            name (str): User name.
            mail (str): User mail.
            result (str): The result from [SES.Client.send_email][], [SES.Client.send_raw_email][].

        '''
        SenderSESLogsDB().add(cid=cid, mail=mail, name=name, ses_result=result)

save staticmethod 

save(
    cid: str, name: str, mail: str, result: dict[str, Any]
) -> None

Save log

Parameters:

Name Type Description Default
cid str

Campaign id.

 required
name str

User name.

 required
mail str

User mail.

 required
result str

The result from SES.Client.send_email, SES.Client.send_raw_email.

 required
Source code in module/sender.py 
243
244
245
246
247
248
249
250
251
252
253
254
@staticmethod
def save(cid: str, name: str, mail: str, result: dict[str, Any]) -> None:
    ''' Save log

    Args:
        cid (str): Campaign id.
        name (str): User name.
        mail (str): User mail.
        result (str): The result from [SES.Client.send_email][], [SES.Client.send_raw_email][].

    '''
    SenderSESLogsDB().add(cid=cid, mail=mail, name=name, ses_result=result)