Skip to content

models/waitlistdb.py

models.waitlistdb

WaitListDB

WaitListDB

Bases: DBBase

WaitList Collection

Source code in models/waitlistdb.py
class WaitListDB(DBBase):
    ''' WaitList Collection '''

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

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

        Indexs:
            - `pid`
            - `uid`

        '''
        self.create_index([('pid', 1), ])
        self.create_index([('uid', 1), ])

    def join_to(self, pid: str, tid: str, uid: str, note: Optional[str] = None) -> dict[str, Any]:
        ''' Join to

        Args:
            pid (str): Project id.
            tid (str): Team id.
            uid (str): User id.
            note (str): Optional. The note from user.

        Returns:
            Return the inserted / updated data.

        '''
        return self.find_one_and_update(
            {'pid': pid, 'tid': tid, 'uid': uid, 'result': {'$exists': False}},
            {'$set': {'note': note}},
            upsert=True,
            return_document=ReturnDocument.AFTER,
        )

    def is_in_wait(self, pid: str, tid: str, uid: str) -> int:
        ''' Is in waitting list

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

        Returns:
            How many requests are in waiting.

        '''
        return self.count_documents({
            'pid': pid, 'tid': tid, 'uid': uid, 'result': {'$exists': False}})

    def list_by(self, pid: str, tid: Optional[str] = None, uid: Optional[str] = None,
                _all: bool = False) -> Union[Optional[dict[str, Any]], Cursor[dict[str, Any]]]:
        ''' List by

        Args:
            pid (str): Project id.
            tid (str): Team id.
            uid (str): User id.
            _all (bool): To return all include waiting requests.

        Returns:
            If `uid` is specified, the return will be `dict` or `None`.
                Otherwise the return could be iterable.

        '''
        query: dict[str, Any] = {'pid': pid}
        if tid:
            query['tid'] = tid

        if uid:
            query['uid'] = uid

        if not _all:
            query['result'] = {'$exists': False}

        if 'uid' in query:
            return self.find_one(query)

        return self.find(query)

    def make_result(self, _id: str, pid: str, uid: str,
                    result: Literal['approval', 'deny']) -> Optional[dict[str, Any]]:
        ''' Make result

        Args:
            _id (str): waitlist id.
            pid (str): Project id.
            uid (str): User id.
            result (Literal): In `approval` or `deny`.

        Returns:
            Return the inserted / updated data.

        '''
        if result in ('approval', 'deny'):
            return self.find_one_and_update(
                {'_id': ObjectId(_id), 'pid': pid, 'uid': uid},
                {'$set': {'result': result}},
                return_document=ReturnDocument.AFTER,
            )

        return None

index

index() -> None

To make collection's index

Indexs
  • pid
  • uid
Source code in models/waitlistdb.py
def index(self) -> None:
    ''' To make collection's index

    Indexs:
        - `pid`
        - `uid`

    '''
    self.create_index([('pid', 1), ])
    self.create_index([('uid', 1), ])

is_in_wait

is_in_wait(pid: str, tid: str, uid: str) -> int

Is in waitting list

Parameters:

Name Type Description Default
pid str

Project id.

required
tid str

Team id.

required
uid str

User id.

required

Returns:

Type Description
int

How many requests are in waiting.

Source code in models/waitlistdb.py
def is_in_wait(self, pid: str, tid: str, uid: str) -> int:
    ''' Is in waitting list

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

    Returns:
        How many requests are in waiting.

    '''
    return self.count_documents({
        'pid': pid, 'tid': tid, 'uid': uid, 'result': {'$exists': False}})

join_to

join_to(
    pid: str, tid: str, uid: str, note: Optional[str] = None
) -> dict[str, Any]

Join to

Parameters:

Name Type Description Default
pid str

Project id.

required
tid str

Team id.

required
uid str

User id.

required
note str

Optional. The note from user.

None

Returns:

Type Description
dict[str, Any]

Return the inserted / updated data.

Source code in models/waitlistdb.py
def join_to(self, pid: str, tid: str, uid: str, note: Optional[str] = None) -> dict[str, Any]:
    ''' Join to

    Args:
        pid (str): Project id.
        tid (str): Team id.
        uid (str): User id.
        note (str): Optional. The note from user.

    Returns:
        Return the inserted / updated data.

    '''
    return self.find_one_and_update(
        {'pid': pid, 'tid': tid, 'uid': uid, 'result': {'$exists': False}},
        {'$set': {'note': note}},
        upsert=True,
        return_document=ReturnDocument.AFTER,
    )

list_by

list_by(
    pid: str,
    tid: Optional[str] = None,
    uid: Optional[str] = None,
    _all: bool = False,
) -> Union[
    Optional[dict[str, Any]], Cursor[dict[str, Any]]
]

List by

Parameters:

Name Type Description Default
pid str

Project id.

required
tid str

Team id.

None
uid str

User id.

None
_all bool

To return all include waiting requests.

False

Returns:

Type Description
Union[Optional[dict[str, Any]], Cursor[dict[str, Any]]]

If uid is specified, the return will be dict or None. Otherwise the return could be iterable.

Source code in models/waitlistdb.py
def list_by(self, pid: str, tid: Optional[str] = None, uid: Optional[str] = None,
            _all: bool = False) -> Union[Optional[dict[str, Any]], Cursor[dict[str, Any]]]:
    ''' List by

    Args:
        pid (str): Project id.
        tid (str): Team id.
        uid (str): User id.
        _all (bool): To return all include waiting requests.

    Returns:
        If `uid` is specified, the return will be `dict` or `None`.
            Otherwise the return could be iterable.

    '''
    query: dict[str, Any] = {'pid': pid}
    if tid:
        query['tid'] = tid

    if uid:
        query['uid'] = uid

    if not _all:
        query['result'] = {'$exists': False}

    if 'uid' in query:
        return self.find_one(query)

    return self.find(query)

make_result

make_result(
    _id: str,
    pid: str,
    uid: str,
    result: Literal["approval", "deny"],
) -> Optional[dict[str, Any]]

Make result

Parameters:

Name Type Description Default
_id str

waitlist id.

required
pid str

Project id.

required
uid str

User id.

required
result Literal

In approval or deny.

required

Returns:

Type Description
Optional[dict[str, Any]]

Return the inserted / updated data.

Source code in models/waitlistdb.py
def make_result(self, _id: str, pid: str, uid: str,
                result: Literal['approval', 'deny']) -> Optional[dict[str, Any]]:
    ''' Make result

    Args:
        _id (str): waitlist id.
        pid (str): Project id.
        uid (str): User id.
        result (Literal): In `approval` or `deny`.

    Returns:
        Return the inserted / updated data.

    '''
    if result in ('approval', 'deny'):
        return self.find_one_and_update(
            {'_id': ObjectId(_id), 'pid': pid, 'uid': uid},
            {'$set': {'result': result}},
            return_document=ReturnDocument.AFTER,
        )

    return None