Skip to content

module/tasks.py

module.tasks

Task

Tasks

Tasks class

Source code in module/tasks.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
 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
class Tasks:
    ''' Tasks class '''

    @classmethod
    def add(cls, pid: str, body: dict[str, Any]) -> dict[str, Any]:
        ''' add new task

        Args:
            pid (str): Project id.
            body (dict): The data to add.

                - `title`: Task title.
                - `cate`: Take category.
                - `desc`: Description.
                - `limit`: `int` User limit.
                - `starttime`: Start time in ISO8601.

        '''
        data = body

        if cls.get_with_pid(pid=pid, _id=data['_id']):
            data.pop('people', None)
            data.pop('created_by', None)
            data.pop('created_at', None)

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

    @staticmethod
    def delete(pid: str, _id: str) -> None:
        ''' Del task

        Args:
            pid (str): Project id.
            _id (str): task unique id.

        '''
        TasksDB().delete_one({'_id': _id, 'pid': pid})

    @staticmethod
    def get_by_pid(pid: str) -> Generator[dict[str, Any], None, None]:
        ''' Get by pid

        Args:
            pid (str): Project id.

        Yields:
            Yield return the data.

        '''
        for raw in TasksDB().find({'pid': pid}, sort=(('starttime', 1), )):
            yield raw

    @staticmethod
    def get_with_pid(pid: str, _id: str) -> Optional[dict[str, Any]]:
        ''' Get with pid

        Args:
            pid (str): Project id.
            _id (str): task unique id.

        Returns:
            Return the data.

        '''
        return TasksDB().find_one({'pid': pid, '_id': _id})

    @staticmethod
    def get_cate(pid: str) -> list[dict[str, Any]]:
        ''' Get cate

        Args:
            pid (str): Project id.

        Returns:
            Return the datas.

        '''
        cates = TasksDB().find({'pid': pid}).distinct('cate')
        return [cate for cate in cates if cate]

    @staticmethod
    def join(pid: str, task_id: str, uid: str) -> dict[str, Any]:
        ''' Join to

        Args:
            pid (str): Project id.
            task_id (str): Task id.
            uid (str): User id.

        Returns:
            Return the added data.

        '''
        return TasksDB().find_one_and_update(
            {'_id': task_id, 'pid': pid},
            {'$addToSet': {'people': uid}},
            return_document=ReturnDocument.AFTER,
        )

    @staticmethod
    def cancel(pid: str, task_id: str, uid: str) -> dict[str, Any]:
        ''' cancel join

        Args:
            pid (str): Project id.
            task_id (str): Task id.
            uid (str): User id.

        Returns:
            Return the updated data.

        '''
        return TasksDB().find_one_and_update(
            {'_id': task_id, 'pid': pid},
            {'$pull': {'people': uid}},
            return_document=ReturnDocument.AFTER,
        )

    @staticmethod
    def get_peoples_info(pid: str, task_id: str) -> Optional[dict[str, Any]]:
        ''' Get peoples info

        Args:
            pid (str): Project id.
            task_id (str): Task id.

        Returns:
            Return the datas.

        '''
        task = TasksDB().find_one(
            {'pid': pid, '_id': task_id}, {'people': 1})

        if task:
            uids = task['people']
            return User.get_info(uids=uids)

        return None

    @staticmethod
    def sitemap() -> list[str]:
        ''' list sitemap paths '''
        pid = set()
        paths: list[str] = []
        for task in TasksDB().find():
            pid.add(task['pid'])
            paths.append(f"/tasks/{task['pid']}/r/{task['_id']}")

        for year in pid:
            paths.append(f"/tasks/{year}")

        return paths

add classmethod 

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

add new task

Parameters:

Name Type Description Default
pid str

Project id.

 required
body dict

The data to add.

  • title: Task title.
  • cate: Take category.
  • desc: Description.
  • limit: int User limit.
  • starttime: Start time in ISO8601.
 required
Source code in module/tasks.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
@classmethod
def add(cls, pid: str, body: dict[str, Any]) -> dict[str, Any]:
    ''' add new task

    Args:
        pid (str): Project id.
        body (dict): The data to add.

            - `title`: Task title.
            - `cate`: Take category.
            - `desc`: Description.
            - `limit`: `int` User limit.
            - `starttime`: Start time in ISO8601.

    '''
    data = body

    if cls.get_with_pid(pid=pid, _id=data['_id']):
        data.pop('people', None)
        data.pop('created_by', None)
        data.pop('created_at', None)

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

cancel staticmethod 

cancel(pid: str, task_id: str, uid: str) -> dict[str, Any]

cancel join

Parameters:

Name Type Description Default
pid str

Project id.

 required
task_id str

Task id.

 required
uid str

User id.

 required

Returns:

Type Description
dict[str, Any]

Return the updated data.
Source code in module/tasks.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@staticmethod
def cancel(pid: str, task_id: str, uid: str) -> dict[str, Any]:
    ''' cancel join

    Args:
        pid (str): Project id.
        task_id (str): Task id.
        uid (str): User id.

    Returns:
        Return the updated data.

    '''
    return TasksDB().find_one_and_update(
        {'_id': task_id, 'pid': pid},
        {'$pull': {'people': uid}},
        return_document=ReturnDocument.AFTER,
    )

delete staticmethod 

delete(pid: str, _id: str) -> None

Del task

Parameters:

Name Type Description Default
pid str

Project id.

 required
_id str

task unique id.

 required
Source code in module/tasks.py 
43
44
45
46
47
48
49
50
51
52
@staticmethod
def delete(pid: str, _id: str) -> None:
    ''' Del task

    Args:
        pid (str): Project id.
        _id (str): task unique id.

    '''
    TasksDB().delete_one({'_id': _id, 'pid': pid})

get_by_pid staticmethod 

get_by_pid(
    pid: str,
) -> Generator[dict[str, Any], None, None]

Get by pid

Parameters:

Name Type Description Default
pid str

Project id.

 required

Yields:

Type Description
dict[str, Any]

Yield return the data.
Source code in module/tasks.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
@staticmethod
def get_by_pid(pid: str) -> Generator[dict[str, Any], None, None]:
    ''' Get by pid

    Args:
        pid (str): Project id.

    Yields:
        Yield return the data.

    '''
    for raw in TasksDB().find({'pid': pid}, sort=(('starttime', 1), )):
        yield raw

get_cate staticmethod 

get_cate(pid: str) -> list[dict[str, Any]]

Get cate

Parameters:

Name Type Description Default
pid str

Project id.

 required

Returns:

Type Description
list[dict[str, Any]]

Return the datas.
Source code in module/tasks.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
@staticmethod
def get_cate(pid: str) -> list[dict[str, Any]]:
    ''' Get cate

    Args:
        pid (str): Project id.

    Returns:
        Return the datas.

    '''
    cates = TasksDB().find({'pid': pid}).distinct('cate')
    return [cate for cate in cates if cate]

get_peoples_info staticmethod 

get_peoples_info(
    pid: str, task_id: str
) -> Optional[dict[str, Any]]

Get peoples info

Parameters:

Name Type Description Default
pid str

Project id.

 required
task_id str

Task id.

 required

Returns:

Type Description
Optional[dict[str, Any]]

Return the datas.
Source code in module/tasks.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@staticmethod
def get_peoples_info(pid: str, task_id: str) -> Optional[dict[str, Any]]:
    ''' Get peoples info

    Args:
        pid (str): Project id.
        task_id (str): Task id.

    Returns:
        Return the datas.

    '''
    task = TasksDB().find_one(
        {'pid': pid, '_id': task_id}, {'people': 1})

    if task:
        uids = task['people']
        return User.get_info(uids=uids)

    return None

get_with_pid staticmethod 

get_with_pid(
    pid: str, _id: str
) -> Optional[dict[str, Any]]

Get with pid

Parameters:

Name Type Description Default
pid str

Project id.

 required
_id str

task unique id.

 required

Returns:

Type Description
Optional[dict[str, Any]]

Return the data.
Source code in module/tasks.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
@staticmethod
def get_with_pid(pid: str, _id: str) -> Optional[dict[str, Any]]:
    ''' Get with pid

    Args:
        pid (str): Project id.
        _id (str): task unique id.

    Returns:
        Return the data.

    '''
    return TasksDB().find_one({'pid': pid, '_id': _id})

join staticmethod 

join(pid: str, task_id: str, uid: str) -> dict[str, Any]

Join to

Parameters:

Name Type Description Default
pid str

Project id.

 required
task_id str

Task id.

 required
uid str

User id.

 required

Returns:

Type Description
dict[str, Any]

Return the added data.
Source code in module/tasks.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@staticmethod
def join(pid: str, task_id: str, uid: str) -> dict[str, Any]:
    ''' Join to

    Args:
        pid (str): Project id.
        task_id (str): Task id.
        uid (str): User id.

    Returns:
        Return the added data.

    '''
    return TasksDB().find_one_and_update(
        {'_id': task_id, 'pid': pid},
        {'$addToSet': {'people': uid}},
        return_document=ReturnDocument.AFTER,
    )

sitemap staticmethod 

sitemap() -> list[str]

list sitemap paths

Source code in module/tasks.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
@staticmethod
def sitemap() -> list[str]:
    ''' list sitemap paths '''
    pid = set()
    paths: list[str] = []
    for task in TasksDB().find():
        pid.add(task['pid'])
        paths.append(f"/tasks/{task['pid']}/r/{task['_id']}")

    for year in pid:
        paths.append(f"/tasks/{year}")

    return paths

TasksStar

TasksStar object

Source code in module/tasks.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
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
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
class TasksStar:
    ''' TasksStar object '''

    @staticmethod
    def add(pid: str, uid: str) -> Optional[dict[str, Any]]:
        ''' add

        Args:
            pid (str): Project id.
            uid (str): User id.

        Returns:
            Return the data.

        '''
        data = TasksStarDB().find_one({'pid': pid, 'uid': uid})
        if not data:
            TasksStarDB().insert_one(TasksStarDB.new(pid=pid, uid=uid))
            data = TasksStarDB().find_one({'pid': pid, 'uid': uid})

        return data

    @staticmethod
    def delete(pid: str, uid: str) -> DeleteResult:
        ''' delete

        Args:
            pid (str): Project id.
            uid (str): User id.

        Returns:
            Return the delete result in [pymongo.results.DeleteResult][].

        '''
        return TasksStarDB().delete_one({'pid': pid, 'uid': uid})

    @staticmethod
    def status(pid: str, uid: str) -> dict[str, bool]:
        ''' ststus

        Args:
            pid (str): Project id.
            uid (str): User id.

        Returns:
            `{'add': <bool>}`

        '''
        if TasksStarDB().find_one({'pid': pid, 'uid': uid}):
            return {'add': True}

        return {'add': False}

    @staticmethod
    def toggle(pid: str, uid: str) -> dict[str, bool]:
        ''' toggle

        Args:
            pid (str): Project id.
            uid (str): User id.

        Returns:
            `{'add': <bool>}`

        '''
        data = TasksStarDB().find_one({'pid': pid, 'uid': uid})
        if data:
            TasksStarDB().delete_one({'pid': pid, 'uid': uid})
            return {'add': False}

        TasksStarDB().insert_one(TasksStarDB.new(pid=pid, uid=uid))
        return {'add': True}

    @staticmethod
    def get(pid: str) -> Generator[dict[str, Any], None, None]:
        ''' Get all star users

        Args:
            pid (str): Project id.

        Yields:
            Return the datas.

        '''
        for user in TasksStarDB().find({'pid': pid}, {'uid': 1}):
            yield user

add staticmethod 

add(pid: str, uid: str) -> Optional[dict[str, Any]]

add

Parameters:

Name Type Description Default
pid str

Project id.

 required
uid str

User id.

 required

Returns:

Type Description
Optional[dict[str, Any]]

Return the data.
Source code in module/tasks.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
@staticmethod
def add(pid: str, uid: str) -> Optional[dict[str, Any]]:
    ''' add

    Args:
        pid (str): Project id.
        uid (str): User id.

    Returns:
        Return the data.

    '''
    data = TasksStarDB().find_one({'pid': pid, 'uid': uid})
    if not data:
        TasksStarDB().insert_one(TasksStarDB.new(pid=pid, uid=uid))
        data = TasksStarDB().find_one({'pid': pid, 'uid': uid})

    return data

delete staticmethod 

delete(pid: str, uid: str) -> DeleteResult

delete

Parameters:

Name Type Description Default
pid str

Project id.

 required
uid str

User id.

 required

Returns:

Type Description
DeleteResult

Return the delete result in pymongo.results.DeleteResult.
Source code in module/tasks.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
@staticmethod
def delete(pid: str, uid: str) -> DeleteResult:
    ''' delete

    Args:
        pid (str): Project id.
        uid (str): User id.

    Returns:
        Return the delete result in [pymongo.results.DeleteResult][].

    '''
    return TasksStarDB().delete_one({'pid': pid, 'uid': uid})

get staticmethod 

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

Get all star users

Parameters:

Name Type Description Default
pid str

Project id.

 required

Yields:

Type Description
dict[str, Any]

Return the datas.
Source code in module/tasks.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
@staticmethod
def get(pid: str) -> Generator[dict[str, Any], None, None]:
    ''' Get all star users

    Args:
        pid (str): Project id.

    Yields:
        Return the datas.

    '''
    for user in TasksStarDB().find({'pid': pid}, {'uid': 1}):
        yield user

status staticmethod 

status(pid: str, uid: str) -> dict[str, bool]

ststus

Parameters:

Name Type Description Default
pid str

Project id.

 required
uid str

User id.

 required

Returns:

Type Description
dict[str, bool]

{'add': <bool>}
Source code in module/tasks.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
@staticmethod
def status(pid: str, uid: str) -> dict[str, bool]:
    ''' ststus

    Args:
        pid (str): Project id.
        uid (str): User id.

    Returns:
        `{'add': <bool>}`

    '''
    if TasksStarDB().find_one({'pid': pid, 'uid': uid}):
        return {'add': True}

    return {'add': False}

toggle staticmethod 

toggle(pid: str, uid: str) -> dict[str, bool]

toggle

Parameters:

Name Type Description Default
pid str

Project id.

 required
uid str

User id.

 required

Returns:

Type Description
dict[str, bool]

{'add': <bool>}
Source code in module/tasks.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@staticmethod
def toggle(pid: str, uid: str) -> dict[str, bool]:
    ''' toggle

    Args:
        pid (str): Project id.
        uid (str): User id.

    Returns:
        `{'add': <bool>}`

    '''
    data = TasksStarDB().find_one({'pid': pid, 'uid': uid})
    if data:
        TasksStarDB().delete_one({'pid': pid, 'uid': uid})
        return {'add': False}

    TasksStarDB().insert_one(TasksStarDB.new(pid=pid, uid=uid))
    return {'add': True}