Skip to content

module/usession.py

module.usession

USession

USession

USession Class

Source code in module/usession.py
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.

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

    @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.

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

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

        Args:
            uid (str): User id.

        Yields:
            Return the datas.

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

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

Return the datas.

Source code in module/usession.py
@staticmethod
def get_alive(uid: str) -> Generator[dict[str, Any], None, None]:
    ''' Get alive session

    Args:
        uid (str): User id.

    Yields:
        Return the datas.

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

get_no_ipinfo staticmethod

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

Get no ipinfo

Yields:

Type Description
Generator[dict[str, Any], None, None]

Return the data, ipinfo is not exist.

Source code in module/usession.py
@staticmethod
def get_no_ipinfo() -> Generator[dict[str, Any], None, None]:
    ''' Get no ipinfo

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

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

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

Return the recently datas.

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

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

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
@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
@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
@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}})