Skip to content

module/tasks.py

module.tasks

Task

Tasks

Tasks class

Source code in module/tasks.py
class Tasks:
    ''' Tasks class '''

    @classmethod
    def add(cls, pid: str, body: dict[str, Any],
            endtime: Optional[datetime] = None, task_id: Optional[str] = None) -> 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.

            endtime (datetime): End time in ISO8601.
            task_id (str): Task id

        '''
        data = TasksDB.new(pid=pid, body=body, endtime=endtime)

        if task_id is not None:
            if not cls.get_with_pid(pid=pid, _id=task_id):
                raise Exception(f'No task_id: {task_id}')

            data['_id'] = task_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

add classmethod

add(
    pid: str,
    body: dict[str, Any],
    endtime: Optional[datetime] = None,
    task_id: Optional[str] = None,
) -> 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
endtime datetime

End time in ISO8601.

None
task_id str

Task id

None
Source code in module/tasks.py
@classmethod
def add(cls, pid: str, body: dict[str, Any],
        endtime: Optional[datetime] = None, task_id: Optional[str] = None) -> 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.

        endtime (datetime): End time in ISO8601.
        task_id (str): Task id

    '''
    data = TasksDB.new(pid=pid, body=body, endtime=endtime)

    if task_id is not None:
        if not cls.get_with_pid(pid=pid, _id=task_id):
            raise Exception(f'No task_id: {task_id}')

        data['_id'] = task_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
@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
@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
Generator[dict[str, Any], None, None]

Yield return the data.

Source code in module/tasks.py
@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
@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
@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
@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
@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,
    )

TasksStar

TasksStar object

Source code in module/tasks.py
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
@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
@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
Generator[dict[str, Any], None, None]

Return the datas.

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