Skip to content

Storage

Storage helper class for storing and retrieving client and resource owner information. See the examples on the sidebar to view this in action. 

from aioauth import storage

AuthorizationCodeStorage

Source code in aioauth/storage.py 
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
class AuthorizationCodeStorage:
    async def create_authorization_code(
        self,
        *,
        request: Request,
        client_id: str,
        scope: str,
        response_type: str,
        redirect_uri: str,
        code: str,
        code_challenge_method: Optional[CodeChallengeMethod] = None,
        code_challenge: Optional[str] = None,
        nonce: Optional[str] = None,
    ) -> AuthorizationCode:
        """Generates an authorization token and stores it in the database.

        Warning:
            Generated authorization token *must* be stored in the database.

        Note:
            This must is used by the response type
            `aioauth.respose_type.ResponseTypeAuthorizationCode`.

        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            scope: The scopes for the token.
            response_type: An `aioauth.types.ResponseType`.
            redirect_uri: The redirect URI.
            code_challenge_method: An `aioauth.types.CodeChallengeMethod`.
            code_challenge: Code challenge string.

        Returns:
            An `aioauth.models.AuthorizationCode` object.
        """
        raise NotImplementedError(
            "Method create_authorization_code must be implemented"
        )

    async def get_authorization_code(
        self,
        *,
        request: Request,
        client_id: str,
        code: str,
    ) -> Optional[AuthorizationCode]:
        """Gets existing authorization code from the database if it exists.

        Warning:
            If authorization code does not exists this function *must*
            return ``None`` to indicate to the validator that the
            requested authorization code does not exist or is invalid.

        Note:
            This method is used by the grant type
            `aioauth.grant_type.AuthorizationCodeGrantType`.

        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            code: An authorization code.

        Returns:
            An optional `aioauth.models.AuthorizationCode`.
        """
        raise NotImplementedError(
            "Method get_authorization_code must be implemented for AuthorizationCodeGrantType"
        )

    async def delete_authorization_code(
        self,
        *,
        request: Request,
        client_id: str,
        code: str,
    ) -> None:
        """Deletes authorization code from database.

        Note:
            This method is used by the grant type
            `aioauth.grant_type.AuthorizationCodeGrantType`.

        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            code: An authorization code.
        """
        raise NotImplementedError(
            "Method delete_authorization_code must be implemented for AuthorizationCodeGrantType"
        )

create_authorization_code(*, request, client_id, scope, response_type, redirect_uri, code, code_challenge_method=None, code_challenge=None, nonce=None) async

Generates an authorization token and stores it in the database.

Warning

Generated authorization token must be stored in the database.

Note

This must is used by the response type aioauth.respose_type.ResponseTypeAuthorizationCode.

Parameters:

Name Type Description Default
request Request

An aioauth.requests.Request.

 required
client_id str

A user client ID.

 required
scope str

The scopes for the token.

 required
response_type str

An aioauth.types.ResponseType.

 required
redirect_uri str

The redirect URI.

 required
code_challenge_method Optional[CodeChallengeMethod]

An aioauth.types.CodeChallengeMethod.

 None
code_challenge Optional[str]

Code challenge string.

 None

Returns:

Type Description
AuthorizationCode

An aioauth.models.AuthorizationCode object.
Source code in aioauth/storage.py 
 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
async def create_authorization_code(
    self,
    *,
    request: Request,
    client_id: str,
    scope: str,
    response_type: str,
    redirect_uri: str,
    code: str,
    code_challenge_method: Optional[CodeChallengeMethod] = None,
    code_challenge: Optional[str] = None,
    nonce: Optional[str] = None,
) -> AuthorizationCode:
    """Generates an authorization token and stores it in the database.

    Warning:
        Generated authorization token *must* be stored in the database.

    Note:
        This must is used by the response type
        `aioauth.respose_type.ResponseTypeAuthorizationCode`.

    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        scope: The scopes for the token.
        response_type: An `aioauth.types.ResponseType`.
        redirect_uri: The redirect URI.
        code_challenge_method: An `aioauth.types.CodeChallengeMethod`.
        code_challenge: Code challenge string.

    Returns:
        An `aioauth.models.AuthorizationCode` object.
    """
    raise NotImplementedError(
        "Method create_authorization_code must be implemented"
    )

delete_authorization_code(*, request, client_id, code) async

Deletes authorization code from database.

Note

This method is used by the grant type aioauth.grant_type.AuthorizationCodeGrantType.

Parameters:

Name Type Description Default
request Request

An aioauth.requests.Request.

 required
client_id str

A user client ID.

 required
code str

An authorization code.

 required
Source code in aioauth/storage.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def delete_authorization_code(
    self,
    *,
    request: Request,
    client_id: str,
    code: str,
) -> None:
    """Deletes authorization code from database.

    Note:
        This method is used by the grant type
        `aioauth.grant_type.AuthorizationCodeGrantType`.

    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        code: An authorization code.
    """
    raise NotImplementedError(
        "Method delete_authorization_code must be implemented for AuthorizationCodeGrantType"
    )

get_authorization_code(*, request, client_id, code) async

Gets existing authorization code from the database if it exists.

Warning

If authorization code does not exists this function must return None to indicate to the validator that the requested authorization code does not exist or is invalid.

Note

This method is used by the grant type aioauth.grant_type.AuthorizationCodeGrantType.

Parameters:

Name Type Description Default
request Request

An aioauth.requests.Request.

 required
client_id str

A user client ID.

 required
code str

An authorization code.

 required

Returns:

Type Description
Optional[AuthorizationCode]

An optional aioauth.models.AuthorizationCode.
Source code in aioauth/storage.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def get_authorization_code(
    self,
    *,
    request: Request,
    client_id: str,
    code: str,
) -> Optional[AuthorizationCode]:
    """Gets existing authorization code from the database if it exists.

    Warning:
        If authorization code does not exists this function *must*
        return ``None`` to indicate to the validator that the
        requested authorization code does not exist or is invalid.

    Note:
        This method is used by the grant type
        `aioauth.grant_type.AuthorizationCodeGrantType`.

    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        code: An authorization code.

    Returns:
        An optional `aioauth.models.AuthorizationCode`.
    """
    raise NotImplementedError(
        "Method get_authorization_code must be implemented for AuthorizationCodeGrantType"
    )

ClientStorage

Source code in aioauth/storage.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
class ClientStorage:
    async def get_client(
        self,
        *,
        request: Request,
        client_id: str,
        client_secret: Optional[str] = None,
    ) -> Optional[Client]:
        """Gets existing client from the database if it exists.

        Warning:
            If client does not exists in database this method *must*
            return `None` to indicate to the validator that the
            requested `client_id` does not exist or is invalid.

        Note:
            This method is used by all core grant types, as well as
            all core response types.

        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            client_secret: An optional user client secret.

        Returns:
            An optional `aioauth.models.Client` object.
        """
        raise NotImplementedError("Method get_client must be implemented")

get_client(*, request, client_id, client_secret=None) async

Gets existing client from the database if it exists.

Warning

If client does not exists in database this method must return None to indicate to the validator that the requested client_id does not exist or is invalid.

Note

This method is used by all core grant types, as well as all core response types.

Parameters:

Name Type Description Default
request Request

An aioauth.requests.Request.

 required
client_id str

A user client ID.

 required
client_secret Optional[str]

An optional user client secret.

 None

Returns:

Type Description
Optional[Client]

An optional aioauth.models.Client object.
Source code in aioauth/storage.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
async def get_client(
    self,
    *,
    request: Request,
    client_id: str,
    client_secret: Optional[str] = None,
) -> Optional[Client]:
    """Gets existing client from the database if it exists.

    Warning:
        If client does not exists in database this method *must*
        return `None` to indicate to the validator that the
        requested `client_id` does not exist or is invalid.

    Note:
        This method is used by all core grant types, as well as
        all core response types.

    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        client_secret: An optional user client secret.

    Returns:
        An optional `aioauth.models.Client` object.
    """
    raise NotImplementedError("Method get_client must be implemented")

IDTokenStorage

Source code in aioauth/storage.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
class IDTokenStorage:
    async def get_id_token(
        self,
        *,
        request: Request,
        client_id: str,
        scope: str,
        redirect_uri: str,
        response_type: Optional[str] = None,
        nonce: Optional[str] = None,
    ) -> str:
        """Returns an id_token.
        For more information see [OpenID Connect Core 1.0 incorporating errata set 1 section 2](https://openid.net/specs/openid-connect-core-1_0.html#IDToken).

        Note:
            Method is used by response type `aioauth.response_type.ResponseTypeIdToken`
            and `aioauth.oidc.core.grant_type.AuthorizationCodeGrantType`.
        """
        raise NotImplementedError("get_id_token must be implemented.")

get_id_token(*, request, client_id, scope, redirect_uri, response_type=None, nonce=None) async

Returns an id_token. For more information see OpenID Connect Core 1.0 incorporating errata set 1 section 2.

Note

Method is used by response type aioauth.response_type.ResponseTypeIdToken and aioauth.oidc.core.grant_type.AuthorizationCodeGrantType.

Source code in aioauth/storage.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
async def get_id_token(
    self,
    *,
    request: Request,
    client_id: str,
    scope: str,
    redirect_uri: str,
    response_type: Optional[str] = None,
    nonce: Optional[str] = None,
) -> str:
    """Returns an id_token.
    For more information see [OpenID Connect Core 1.0 incorporating errata set 1 section 2](https://openid.net/specs/openid-connect-core-1_0.html#IDToken).

    Note:
        Method is used by response type `aioauth.response_type.ResponseTypeIdToken`
        and `aioauth.oidc.core.grant_type.AuthorizationCodeGrantType`.
    """
    raise NotImplementedError("get_id_token must be implemented.")

TokenStorage

Source code in aioauth/storage.py 
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
class TokenStorage:
    async def create_token(
        self,
        *,
        request: Request,
        client_id: str,
        scope: str,
        access_token: str,
        refresh_token: Optional[str] = None,
    ) -> Token:
        """Generates a user token and stores it in the database.

        Used by:

        * `ResponseTypeToken`
        * `AuthorizationCodeGrantType`
        * `PasswordGrantType`
        * `ClientCredentialsGrantType`
        * `RefreshTokenGrantType`

        Warning:
            Generated token *must* be stored in the database.
        Note:
            Method is used by all core grant types, but only used for
            `aioauth.response_type.ResponseTypeToken`.
        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            scope: The scopes for the token.
        Returns:
            The new generated `aioauth.models.Token`.
        """
        raise NotImplementedError("Method create_token must be implemented")

    async def get_token(
        self,
        *,
        request: Request,
        client_id: str,
        token_type: Optional[TokenType] = None,
        access_token: Optional[str] = None,
        refresh_token: Optional[str] = None,
    ) -> Optional[Token]:
        """Gets existing token from the database.

        Note:
            Method is used by
            `aioauth.server.AuthorizationServer`,  and by the
            grant type `aioauth.grant_types.RefreshTokenGrantType`.
        Args:
            request: An `aioauth.requests.Request`.
            client_id: A user client ID.
            access_token: The user access token.
            refresh_token: The user refresh token.
        Returns:
            An optional `aioauth.models.Token` object.
        """
        raise NotImplementedError("Method get_token must be implemented")

    async def revoke_token(
        self,
        *,
        request: Request,
        client_id: str,
        refresh_token: Optional[str] = None,
        token_type: Optional[TokenType] = None,
        access_token: Optional[str] = None,
    ) -> None:
        """Revokes a token from the database."""
        raise NotImplementedError

create_token(*, request, client_id, scope, access_token, refresh_token=None) async

Generates a user token and stores it in the database.

Used by:

  • ResponseTypeToken
  • AuthorizationCodeGrantType
  • PasswordGrantType
  • ClientCredentialsGrantType
  • RefreshTokenGrantType
Warning

Generated token must be stored in the database.

Note: Method is used by all core grant types, but only used for aioauth.response_type.ResponseTypeToken. Args: request: An aioauth.requests.Request. client_id: A user client ID. scope: The scopes for the token. Returns: The new generated aioauth.models.Token.

Source code in aioauth/storage.py 
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
async def create_token(
    self,
    *,
    request: Request,
    client_id: str,
    scope: str,
    access_token: str,
    refresh_token: Optional[str] = None,
) -> Token:
    """Generates a user token and stores it in the database.

    Used by:

    * `ResponseTypeToken`
    * `AuthorizationCodeGrantType`
    * `PasswordGrantType`
    * `ClientCredentialsGrantType`
    * `RefreshTokenGrantType`

    Warning:
        Generated token *must* be stored in the database.
    Note:
        Method is used by all core grant types, but only used for
        `aioauth.response_type.ResponseTypeToken`.
    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        scope: The scopes for the token.
    Returns:
        The new generated `aioauth.models.Token`.
    """
    raise NotImplementedError("Method create_token must be implemented")

get_token(*, request, client_id, token_type=None, access_token=None, refresh_token=None) async

Gets existing token from the database.

Note

Method is used by aioauth.server.AuthorizationServer, and by the grant type aioauth.grant_types.RefreshTokenGrantType.

Args: request: An aioauth.requests.Request. client_id: A user client ID. access_token: The user access token. refresh_token: The user refresh token. Returns: An optional aioauth.models.Token object.

Source code in aioauth/storage.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
async def get_token(
    self,
    *,
    request: Request,
    client_id: str,
    token_type: Optional[TokenType] = None,
    access_token: Optional[str] = None,
    refresh_token: Optional[str] = None,
) -> Optional[Token]:
    """Gets existing token from the database.

    Note:
        Method is used by
        `aioauth.server.AuthorizationServer`,  and by the
        grant type `aioauth.grant_types.RefreshTokenGrantType`.
    Args:
        request: An `aioauth.requests.Request`.
        client_id: A user client ID.
        access_token: The user access token.
        refresh_token: The user refresh token.
    Returns:
        An optional `aioauth.models.Token` object.
    """
    raise NotImplementedError("Method get_token must be implemented")

revoke_token(*, request, client_id, refresh_token=None, token_type=None, access_token=None) async

Revokes a token from the database.

Source code in aioauth/storage.py 
77
78
79
80
81
82
83
84
85
86
87
async def revoke_token(
    self,
    *,
    request: Request,
    client_id: str,
    refresh_token: Optional[str] = None,
    token_type: Optional[TokenType] = None,
    access_token: Optional[str] = None,
) -> None:
    """Revokes a token from the database."""
    raise NotImplementedError

UserStorage

Source code in aioauth/storage.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
class UserStorage:
    async def get_user(self, request: Request) -> Optional[Any]:
        """Returns a user.

        Note:
            This method is used by the grant type
            `aioauth.grant_type.PasswordGrantType`.

        Args:
            request: An `aioauth.requests.Request`.

        Returns:
            Boolean indicating whether or not the user was authenticated
            successfully.
        """
        raise NotImplementedError("Method get_user must be implemented")

get_user(request) async

Returns a user.

Note

This method is used by the grant type aioauth.grant_type.PasswordGrantType.

Parameters:

Name Type Description Default
request Request

An aioauth.requests.Request.

 required

Returns:

Type Description
Optional[Any]

Boolean indicating whether or not the user was authenticated
Optional[Any]

successfully.
Source code in aioauth/storage.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def get_user(self, request: Request) -> Optional[Any]:
    """Returns a user.

    Note:
        This method is used by the grant type
        `aioauth.grant_type.PasswordGrantType`.

    Args:
        request: An `aioauth.requests.Request`.

    Returns:
        Boolean indicating whether or not the user was authenticated
        successfully.
    """
    raise NotImplementedError("Method get_user must be implemented")