Skip to content

models/senderdb.py

models.senderdb

Sender DB

SenderCampaignDB

Bases: DBBase

SenderCampaign Collection

Source code in models/senderdb.py 
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
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
class SenderCampaignDB(DBBase):
    ''' SenderCampaign Collection '''

    def __init__(self) -> None:
        super().__init__('sender_campaign')

    @staticmethod
    def new(name: str, pid: str, tid: str, uid: str) -> dict[str, Any]:
        ''' new a default struct

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

        Returns:
            Return a default struct.

        Struct:
            - ``_id``: Unique id, or `cid`.
            - ``name``: Campaign name.
            - ``created``:
                - ``pid``: Project id.
                - ``tid``: Team id.
                - ``uid``: User id.
                - ``at``: Created at.
            - ``receiver``:
                - ``teams``: List of team id(`tid`)
                - ``users``: List of user id(`uid)
                - ``team_w_tags``: List of the tag id created by team.
                - ``all_users``: `bool` Is send to all platform's users.

            - ``mail``:
                - ``subject``: Subject.
                - ``content``: Content, and Markdown format supported.

        TODO:
            Need refactor in pydantic.

        '''
        return {
            '_id': uuid4().hex,
            'name': name,
            'created': {
                'pid': pid,
                'tid': tid,
                'uid': uid,
                'at': time(),
            },
            'receiver': {
                'teams': [],
                'users': [],
                'team_w_tags': {},
                'all_users': False,
            },
            'mail': {
                'subject': '',
                'content': '',
                'preheader': '',
                'layout': '1',
            },
        }

    def add(self, data: dict[str, Any]) -> dict[str, Any]:
        ''' Save data

        Args:
            data (dict): The data to insert / update.

        Returns:
            Return the inserted / updated data.

        '''
        return self.find_one_and_update(
            {'_id': data['_id']},
            {'$set': data},
            upsert=True,
            return_document=ReturnDocument.AFTER,
        )

add 

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

Save data

Parameters:

Name Type Description Default
data dict

The data to insert / update.

 required

Returns:

Type Description
dict[str, Any]

Return the inserted / updated data.
Source code in models/senderdb.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def add(self, data: dict[str, Any]) -> dict[str, Any]:
    ''' Save data

    Args:
        data (dict): The data to insert / update.

    Returns:
        Return the inserted / updated data.

    '''
    return self.find_one_and_update(
        {'_id': data['_id']},
        {'$set': data},
        upsert=True,
        return_document=ReturnDocument.AFTER,
    )

new staticmethod 

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

new a default struct

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 a default struct.
Struct
  • _id: Unique id, or cid.
  • name: Campaign name.
  • created:
    • pid: Project id.
    • tid: Team id.
    • uid: User id.
    • at: Created at.

  • receiver:

    • teams: List of team id(tid)
    • users: List of user id(`uid)
    • team_w_tags: List of the tag id created by team.
    • all_users: bool Is send to all platform's users.

  • mail:

    • subject: Subject.
    • content: Content, and Markdown format supported.
TODO

Need refactor in pydantic.

Source code in models/senderdb.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
@staticmethod
def new(name: str, pid: str, tid: str, uid: str) -> dict[str, Any]:
    ''' new a default struct

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

    Returns:
        Return a default struct.

    Struct:
        - ``_id``: Unique id, or `cid`.
        - ``name``: Campaign name.
        - ``created``:
            - ``pid``: Project id.
            - ``tid``: Team id.
            - ``uid``: User id.
            - ``at``: Created at.
        - ``receiver``:
            - ``teams``: List of team id(`tid`)
            - ``users``: List of user id(`uid)
            - ``team_w_tags``: List of the tag id created by team.
            - ``all_users``: `bool` Is send to all platform's users.

        - ``mail``:
            - ``subject``: Subject.
            - ``content``: Content, and Markdown format supported.

    TODO:
        Need refactor in pydantic.

    '''
    return {
        '_id': uuid4().hex,
        'name': name,
        'created': {
            'pid': pid,
            'tid': tid,
            'uid': uid,
            'at': time(),
        },
        'receiver': {
            'teams': [],
            'users': [],
            'team_w_tags': {},
            'all_users': False,
        },
        'mail': {
            'subject': '',
            'content': '',
            'preheader': '',
            'layout': '1',
        },
    }

SenderLogsDB

Bases: DBBase

SenderLogsDB Collection

Source code in models/senderdb.py 
 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
class SenderLogsDB(DBBase):
    ''' SenderLogsDB Collection '''

    def __init__(self) -> None:
        super().__init__('sender_logs')

    def add(self, cid: str, layout: str, desc: str, receivers: list[dict[str, Any]]) -> None:
        ''' Save log

        Args:
            cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
            layout (str): In `1`, `2`.
            desc (str): Description.
            receivers (list): List of user info or replace data. Required fields in `name`, `mail`.

        '''

        data = {
            'cid': cid,
            'layout': layout,
            'desc': desc,
            'receivers': receivers,
            'create_at': time(),
        }

        self.insert_one(data)

add 

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

Save log

Parameters:

Name Type Description Default
cid str

Campaign id, the id is from models.senderdb.SenderCampaignDB.

 required
layout str

In 1, 2.

 required
desc str

Description.

 required
receivers list

List of user info or replace data. Required fields in name, mail.

 required
Source code in models/senderdb.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def add(self, cid: str, layout: str, desc: str, receivers: list[dict[str, Any]]) -> None:
    ''' Save log

    Args:
        cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
        layout (str): In `1`, `2`.
        desc (str): Description.
        receivers (list): List of user info or replace data. Required fields in `name`, `mail`.

    '''

    data = {
        'cid': cid,
        'layout': layout,
        'desc': desc,
        'receivers': receivers,
        'create_at': time(),
    }

    self.insert_one(data)

SenderReceiverDB

Bases: DBBase

SenderReceiver Collection

For User to upload the list of receivers in CSV.

Source code in models/senderdb.py 
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
231
232
233
234
235
236
class SenderReceiverDB(DBBase):
    ''' SenderReceiver Collection

    For User to upload the list of receivers in CSV.

    '''

    def __init__(self) -> None:
        super().__init__('sender_receiver')

    def index(self) -> None:
        ''' To make collection's index

        Indexs:
            - `pid`
            - `data.mail`

        '''
        self.create_index([('pid', 1), ])
        self.create_index([('data.mail', 1), ])

    @staticmethod
    def new(pid: str, cid: str, name: str, mail: str) -> dict[str, Any]:
        ''' new a struct

        Args:
            pid (str): Project id.
            cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
            mail (str): User's mail.
            name (str): User name.

        Returns:
            Return a default struct.

        Struct:
            - ``_id``: Random id in `ObjectID`.
            - ``cid``: Campaign id
            - ``pid``: Project id
            - ``data``: `dict`
                - ``mail``: User mail.
                - ``name``: User name.
                - ``...``: *(... and any other fields.)*

        TODO:
            Need refactor in pydantic.

        '''
        return {
            'pid': pid,
            'cid': cid,
            'data': {
                'name': name,
                'mail': mail,
            },
        }

    def remove_past(self, pid: str, cid: str) -> int:
        ''' Remove past data

        Args:
            pid (str): Project id.
            cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].

        '''
        return self.delete_many({'pid': pid, 'cid': cid}).deleted_count

    def update_data(self, pid: str, cid: str, datas: list[dict[str, Any]]) -> None:
        ''' Update datas

        Args:
            pid (str): Project id.
            cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
            datas (list): List of data to inserted / updated. Must have `mail`, `name`.

        '''
        for data in datas:
            _data = {}
            for k in data['data']:
                _data[f'data.{k}'] = data['data'][k]

            for key, value in _data.copy().items():
                _data[key] = value.strip()

            self.find_one_and_update(
                {'pid': pid, 'cid': cid, 'data.mail': data['data']['mail']},
                {'$set': _data},
                upsert=True,
            )

index 

index() -> None

To make collection's index

Indexs
  • pid
  • data.mail
Source code in models/senderdb.py 
159
160
161
162
163
164
165
166
167
168
def index(self) -> None:
    ''' To make collection's index

    Indexs:
        - `pid`
        - `data.mail`

    '''
    self.create_index([('pid', 1), ])
    self.create_index([('data.mail', 1), ])

new staticmethod 

new(
    pid: str, cid: str, name: str, mail: str
) -> dict[str, Any]

new a struct

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id, the id is from models.senderdb.SenderCampaignDB.

 required
mail str

User's mail.

 required
name str

User name.

 required

Returns:

Type Description
dict[str, Any]

Return a default struct.
Struct
  • _id: Random id in ObjectID.
  • cid: Campaign id
  • pid: Project id
  • data: dict
    • mail: User mail.
    • name: User name.
    • ...: (... and any other fields.)
TODO

Need refactor in pydantic.

Source code in models/senderdb.py 
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
@staticmethod
def new(pid: str, cid: str, name: str, mail: str) -> dict[str, Any]:
    ''' new a struct

    Args:
        pid (str): Project id.
        cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
        mail (str): User's mail.
        name (str): User name.

    Returns:
        Return a default struct.

    Struct:
        - ``_id``: Random id in `ObjectID`.
        - ``cid``: Campaign id
        - ``pid``: Project id
        - ``data``: `dict`
            - ``mail``: User mail.
            - ``name``: User name.
            - ``...``: *(... and any other fields.)*

    TODO:
        Need refactor in pydantic.

    '''
    return {
        'pid': pid,
        'cid': cid,
        'data': {
            'name': name,
            'mail': mail,
        },
    }

remove_past 

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

Remove past data

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id, the id is from models.senderdb.SenderCampaignDB.

 required
Source code in models/senderdb.py 
205
206
207
208
209
210
211
212
213
def remove_past(self, pid: str, cid: str) -> int:
    ''' Remove past data

    Args:
        pid (str): Project id.
        cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].

    '''
    return self.delete_many({'pid': pid, 'cid': cid}).deleted_count

update_data 

update_data(
    pid: str, cid: str, datas: list[dict[str, Any]]
) -> None

Update datas

Parameters:

Name Type Description Default
pid str

Project id.

 required
cid str

Campaign id, the id is from models.senderdb.SenderCampaignDB.

 required
datas list

List of data to inserted / updated. Must have mail, name.

 required
Source code in models/senderdb.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def update_data(self, pid: str, cid: str, datas: list[dict[str, Any]]) -> None:
    ''' Update datas

    Args:
        pid (str): Project id.
        cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
        datas (list): List of data to inserted / updated. Must have `mail`, `name`.

    '''
    for data in datas:
        _data = {}
        for k in data['data']:
            _data[f'data.{k}'] = data['data'][k]

        for key, value in _data.copy().items():
            _data[key] = value.strip()

        self.find_one_and_update(
            {'pid': pid, 'cid': cid, 'data.mail': data['data']['mail']},
            {'$set': _data},
            upsert=True,
        )

SenderSESLogsDB

Bases: DBBase

SenderSESLogsDB Collection

Source code in models/senderdb.py 
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
class SenderSESLogsDB(DBBase):
    ''' SenderSESLogsDB Collection '''

    def __init__(self) -> None:
        super().__init__('sender_ses_logs')

    def add(self, cid: str, mail: str, name: str, ses_result: dict[str, Any]) -> None:
        ''' Save log

        Args:
            cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
            mail (str): User mail.
            name (str): User name.
            ses_result (str): The result from the return of
                              [SES.Client.send_email][] / [SES.Client.send_raw_email][].

        '''
        data = {
            'cid': cid,
            'mail': mail,
            'name': name,
            'result': ses_result,
            'create_at': time(),
        }

        self.insert_one(data)

add 

add(
    cid: str,
    mail: str,
    name: str,
    ses_result: dict[str, Any],
) -> None

Save log

Parameters:

Name Type Description Default
cid str

Campaign id, the id is from models.senderdb.SenderCampaignDB.

 required
mail str

User mail.

 required
name str

User name.

 required
ses_result str

The result from the return of SES.Client.send_email / SES.Client.send_raw_email.

 required
Source code in models/senderdb.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def add(self, cid: str, mail: str, name: str, ses_result: dict[str, Any]) -> None:
    ''' Save log

    Args:
        cid (str): Campaign id, the `id` is from [models.senderdb.SenderCampaignDB][].
        mail (str): User mail.
        name (str): User name.
        ses_result (str): The result from the return of
                          [SES.Client.send_email][] / [SES.Client.send_raw_email][].

    '''
    data = {
        'cid': cid,
        'mail': mail,
        'name': name,
        'result': ses_result,
        'create_at': time(),
    }

    self.insert_one(data)