Skip to content

models/waitlistdb.py

models.waitlistdb

WaitListDB

WaitListDB

Bases: DBBase

WaitList Collection

Source code in models/waitlistdb.py 
 10
 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
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) -> Generator[dict[str, Any], None, None]:
        ''' 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:
            data = self.find_one(query)
            if data:
                yield data
        else:
            yield from 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 
16
17
18
19
20
21
22
23
24
25
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 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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,
) -> Generator[dict[str, Any], None, None]

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
Generator[dict[str, Any], None, None]

If uid is specified, the return will be dict or None. Otherwise the return could be iterable.
Source code in models/waitlistdb.py 
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
def list_by(self, pid: str, tid: Optional[str] = None, uid: Optional[str] = None,
            _all: bool = False) -> Generator[dict[str, Any], None, None]:
    ''' 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:
        data = self.find_one(query)
        if data:
            yield data
    else:
        yield from 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 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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