Skip to content

module/usession.py

module.usession

USession

USession

USession Class

Source code in module/usession.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
class USession:
    ''' USession Class '''
    @staticmethod
    def make_new(uid: str, header: dict[str, Any]) -> InsertOneResult:
        ''' make new session record

        Args:
            uid (str): User id.
            header (dict): User's request header.

        Returns:
            Return the [pymongo.results.InsertOneResult][] object.

        '''
        doc = {'uid': uid, 'header': header,
               'created_at': time(), 'alive': True}
        return USessionDB().add(doc)

    @staticmethod
    def get(sid: Optional[str]) -> Optional[dict[str, Any]]:
        ''' Get usession data

        Args:
            sid (str): usession id.

        Returns:
            Return the usession data.

        '''
        return USessionDB(token=sid).get()

    @staticmethod
    def get_no_ipinfo() -> Generator[dict[str, Any], None, None]:
        ''' Get no ipinfo

        Yields:
            Return the data, `ipinfo` is not exist.

        '''
        yield from USessionDB().find({'ipinfo': {'$exists': False}},
                                     {'header.X-Real-Ip': 1, 'header.X-Forwarded-For': 1})

    @staticmethod
    def update_ipinfo(sid: str, data: dict[str, Any]) -> None:
        ''' Update session ipinfo

        Args:
            sid (str): usession id.
            data (dict): The ipinfo response data.

        '''
        USessionDB().find_one_and_update(
            {'_id': sid}, {'$set': {'ipinfo': data}})

    @staticmethod
    def get_recently(uid: str, limit: int = 25) -> Generator[dict[str, Any], None, None]:
        ''' Get recently record

        Args:
            uid (str): User id.
            limit (int): Limit.

        Yields:
            Return the recently datas.

        '''
        yield from USessionDB(token='').find({'uid': uid},
                                             sort=(('created_at', -1), ),
                                             limit=limit)

    @staticmethod
    def get_alive(uid: str) -> Generator[dict[str, Any], None, None]:
        ''' Get alive session

        Args:
            uid (str): User id.

        Yields:
            Return the datas.

        '''
        yield from USessionDB(token='').find({'uid': uid, 'alive': True},
                                             sort=(('created_at', -1), ))

    @staticmethod
    def make_dead(sid: str, uid: Optional[str] = None) -> None:
        ''' Make session to dead

        Args:
            sid (str): usession id.
            uid (str): User id.

        '''
        query = {'_id': sid}
        if uid:
            query['uid'] = uid

        USessionDB(token='').find_one_and_update(
            query, {'$set': {'alive': False}})
        MC.get_client().delete(f'sid:{sid}')

    @staticmethod
    def clean(days: int = 3) -> UpdateResult:
        ''' Make expired

        :param int days: days

        Args:
            days (int): The long days to clean.

        Returns:
            Return the [pymongo.results.UpdateResult][] object.

        '''
        target = time() - 86400*days
        return USessionDB(token='').update_many(
            {'alive': True, 'created_at': {'$lte': target}},
            {'$set': {'alive': False}},
        )

clean staticmethod 

clean(days: int = 3) -> UpdateResult

Make expired

:param int days: days

Parameters:

Name Type Description Default
days int

The long days to clean.

 3

Returns:

Type Description
UpdateResult

Return the pymongo.results.UpdateResult object.
Source code in module/usession.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@staticmethod
def clean(days: int = 3) -> UpdateResult:
    ''' Make expired

    :param int days: days

    Args:
        days (int): The long days to clean.

    Returns:
        Return the [pymongo.results.UpdateResult][] object.

    '''
    target = time() - 86400*days
    return USessionDB(token='').update_many(
        {'alive': True, 'created_at': {'$lte': target}},
        {'$set': {'alive': False}},
    )

get staticmethod 

get(sid: Optional[str]) -> Optional[dict[str, Any]]

Get usession data

Parameters:

Name Type Description Default
sid str

usession id.

 required

Returns:

Type Description
Optional[dict[str, Any]]

Return the usession data.
Source code in module/usession.py 
29
30
31
32
33
34
35
36
37
38
39
40
@staticmethod
def get(sid: Optional[str]) -> Optional[dict[str, Any]]:
    ''' Get usession data

    Args:
        sid (str): usession id.

    Returns:
        Return the usession data.

    '''
    return USessionDB(token=sid).get()

get_alive staticmethod 

get_alive(
    uid: str,
) -> Generator[dict[str, Any], None, None]

Get alive session

Parameters:

Name Type Description Default
uid str

User id.

 required

Yields:

Type Description
dict[str, Any]

Return the datas.
Source code in module/usession.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
@staticmethod
def get_alive(uid: str) -> Generator[dict[str, Any], None, None]:
    ''' Get alive session

    Args:
        uid (str): User id.

    Yields:
        Return the datas.

    '''
    yield from USessionDB(token='').find({'uid': uid, 'alive': True},
                                         sort=(('created_at', -1), ))

get_no_ipinfo staticmethod 

get_no_ipinfo() -> Generator[dict[str, Any], None, None]

Get no ipinfo

Yields:

Type Description
dict[str, Any]

Return the data, ipinfo is not exist.
Source code in module/usession.py 
42
43
44
45
46
47
48
49
50
51
@staticmethod
def get_no_ipinfo() -> Generator[dict[str, Any], None, None]:
    ''' Get no ipinfo

    Yields:
        Return the data, `ipinfo` is not exist.

    '''
    yield from USessionDB().find({'ipinfo': {'$exists': False}},
                                 {'header.X-Real-Ip': 1, 'header.X-Forwarded-For': 1})

get_recently staticmethod 

get_recently(
    uid: str, limit: int = 25
) -> Generator[dict[str, Any], None, None]

Get recently record

Parameters:

Name Type Description Default
uid str

User id.

 required
limit int

Limit.

 25

Yields:

Type Description
dict[str, Any]

Return the recently datas.
Source code in module/usession.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def get_recently(uid: str, limit: int = 25) -> Generator[dict[str, Any], None, None]:
    ''' Get recently record

    Args:
        uid (str): User id.
        limit (int): Limit.

    Yields:
        Return the recently datas.

    '''
    yield from USessionDB(token='').find({'uid': uid},
                                         sort=(('created_at', -1), ),
                                         limit=limit)

make_dead staticmethod 

make_dead(sid: str, uid: Optional[str] = None) -> None

Make session to dead

Parameters:

Name Type Description Default
sid str

usession id.

 required
uid str

User id.

 None
Source code in module/usession.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@staticmethod
def make_dead(sid: str, uid: Optional[str] = None) -> None:
    ''' Make session to dead

    Args:
        sid (str): usession id.
        uid (str): User id.

    '''
    query = {'_id': sid}
    if uid:
        query['uid'] = uid

    USessionDB(token='').find_one_and_update(
        query, {'$set': {'alive': False}})
    MC.get_client().delete(f'sid:{sid}')

make_new staticmethod 

make_new(
    uid: str, header: dict[str, Any]
) -> InsertOneResult

make new session record

Parameters:

Name Type Description Default
uid str

User id.

 required
header dict

User's request header.

 required

Returns:

Type Description
InsertOneResult

Return the pymongo.results.InsertOneResult object.
Source code in module/usession.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
@staticmethod
def make_new(uid: str, header: dict[str, Any]) -> InsertOneResult:
    ''' make new session record

    Args:
        uid (str): User id.
        header (dict): User's request header.

    Returns:
        Return the [pymongo.results.InsertOneResult][] object.

    '''
    doc = {'uid': uid, 'header': header,
           'created_at': time(), 'alive': True}
    return USessionDB().add(doc)

update_ipinfo staticmethod 

update_ipinfo(sid: str, data: dict[str, Any]) -> None

Update session ipinfo

Parameters:

Name Type Description Default
sid str

usession id.

 required
data dict

The ipinfo response data.

 required
Source code in module/usession.py 
53
54
55
56
57
58
59
60
61
62
63
@staticmethod
def update_ipinfo(sid: str, data: dict[str, Any]) -> None:
    ''' Update session ipinfo

    Args:
        sid (str): usession id.
        data (dict): The ipinfo response data.

    '''
    USessionDB().find_one_and_update(
        {'_id': sid}, {'$set': {'ipinfo': data}})