Skip to content

models/oauth_db.py

models.oauth_db

oauth database

OAuthDB

Bases: DBBase

OAuthDB Collection

Source code in models/oauth_db.py 
 9
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
class OAuthDB(DBBase):  # pylint: disable=abstract-method
    ''' OAuthDB Collection '''

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

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

        Indexs:
            - `owner`

        '''
        self.create_index([('owner', 1), ])

    def add_data(self, mail: str, data: dict[str, Any]) -> dict[str, Any]:
        ''' Add user data

        Args:
            mail (str): Email address as unique key to save.
            data (dict): The data from Google OAuth return.

        Returns:
            Return the inserted / updated data.

        '''
        return self.find_one_and_update(
            {'_id': mail},
            {'$set': {'data': data}},
            upsert=True)

    def add_token(self, mail: str, credentials: Credentials) -> None:
        ''' Add user oauth token

        Args:
            mail (str): Email address.
            credentials: [google.oauth2.credentials.Credentials][]

        '''
        oauth = self.find_one({'_id': mail}, {'token': 1})
        if oauth and 'token' in oauth:
            data = oauth['token']
        else:
            data = {}

        data['token'] = credentials.token
        if credentials.refresh_token:
            data['refresh_token'] = credentials.refresh_token

        data['token_uri'] = credentials.token_uri
        data['id_token'] = credentials.id_token
        data['scopes'] = credentials.scopes

        self.find_one_and_update(
            {'_id': mail},
            {'$set': {'token': data}},
            upsert=True)

    def setup_owner(self, mail: str, uid: str) -> None:
        ''' Setup user id into `owner` field

        Args:
            mail (str): Email address.
            uid (str): User id.

        '''
        self.find_one_and_update({'_id': mail}, {'$set': {'owner': uid}})

add_data 

add_data(mail: str, data: dict[str, Any]) -> dict[str, Any]

Add user data

Parameters:

Name Type Description Default
mail str

Email address as unique key to save.

 required
data dict

The data from Google OAuth return.

 required

Returns:

Type Description
dict[str, Any]

Return the inserted / updated data.
Source code in models/oauth_db.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def add_data(self, mail: str, data: dict[str, Any]) -> dict[str, Any]:
    ''' Add user data

    Args:
        mail (str): Email address as unique key to save.
        data (dict): The data from Google OAuth return.

    Returns:
        Return the inserted / updated data.

    '''
    return self.find_one_and_update(
        {'_id': mail},
        {'$set': {'data': data}},
        upsert=True)

add_token 

add_token(mail: str, credentials: Credentials) -> None

Add user oauth token

Parameters:

Name Type Description Default
mail str

Email address.

 required
credentials Credentials

google.oauth2.credentials.Credentials

 required
Source code in models/oauth_db.py 
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
def add_token(self, mail: str, credentials: Credentials) -> None:
    ''' Add user oauth token

    Args:
        mail (str): Email address.
        credentials: [google.oauth2.credentials.Credentials][]

    '''
    oauth = self.find_one({'_id': mail}, {'token': 1})
    if oauth and 'token' in oauth:
        data = oauth['token']
    else:
        data = {}

    data['token'] = credentials.token
    if credentials.refresh_token:
        data['refresh_token'] = credentials.refresh_token

    data['token_uri'] = credentials.token_uri
    data['id_token'] = credentials.id_token
    data['scopes'] = credentials.scopes

    self.find_one_and_update(
        {'_id': mail},
        {'$set': {'token': data}},
        upsert=True)

index 

index() -> None

To make collection's index

Indexs
  • owner
Source code in models/oauth_db.py 
15
16
17
18
19
20
21
22
def index(self) -> None:
    ''' To make collection's index

    Indexs:
        - `owner`

    '''
    self.create_index([('owner', 1), ])

setup_owner 

setup_owner(mail: str, uid: str) -> None

Setup user id into owner field

Parameters:

Name Type Description Default
mail str

Email address.

 required
uid str

User id.

 required
Source code in models/oauth_db.py 
67
68
69
70
71
72
73
74
75
def setup_owner(self, mail: str, uid: str) -> None:
    ''' Setup user id into `owner` field

    Args:
        mail (str): Email address.
        uid (str): User id.

    '''
    self.find_one_and_update({'_id': mail}, {'$set': {'owner': uid}})