-
-
Notifications
You must be signed in to change notification settings - Fork 501
Expand file tree
/
Copy pathfastapi_users.py
More file actions
223 lines (200 loc) · 8.76 KB
/
fastapi_users.py
File metadata and controls
223 lines (200 loc) · 8.76 KB
1
2
3
4
5
6
7
8
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
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
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
180
181
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
210
211
212
213
214
215
216
217
218
219
220
221
222
223
from collections.abc import Sequence
from typing import Generic, Literal
from fastapi import APIRouter
from fastapi_users import models, schemas
from fastapi_users.authentication import AuthenticationBackend, Authenticator
from fastapi_users.jwt import SecretType
from fastapi_users.manager import UserManagerDependency
from fastapi_users.router import (
get_auth_router,
get_register_router,
get_reset_password_router,
get_users_router,
get_verify_router,
)
try:
from httpx_oauth.oauth2 import BaseOAuth2
from fastapi_users.router.oauth import (
CSRF_TOKEN_COOKIE_NAME,
get_oauth_associate_router,
get_oauth_router,
)
except ModuleNotFoundError: # pragma: no cover
BaseOAuth2 = type # type: ignore
CSRF_TOKEN_COOKIE_NAME = "" # type: ignore
class FastAPIUsers(Generic[models.UP, models.ID]):
"""
Main object that ties together the component for users authentication.
:param get_user_manager: Dependency callable getter to inject the
user manager class instance.
:param auth_backends: List of authentication backends.
:attribute current_user: Dependency callable getter to inject authenticated user
with a specific set of parameters.
"""
authenticator: Authenticator[models.UP, models.ID]
def __init__(
self,
get_user_manager: UserManagerDependency[models.UP, models.ID],
auth_backends: Sequence[AuthenticationBackend[models.UP, models.ID]],
):
self.authenticator = Authenticator(auth_backends, get_user_manager)
self.get_user_manager = get_user_manager
self.current_user = self.authenticator.current_user
def get_register_router(
self, user_schema: type[schemas.U], user_create_schema: type[schemas.UC]
) -> APIRouter:
"""
Return a router with a register route.
:param user_schema: Pydantic schema of a public user.
:param user_create_schema: Pydantic schema for creating a user.
"""
return get_register_router(
self.get_user_manager, user_schema, user_create_schema
)
def get_verify_router(self, user_schema: type[schemas.U]) -> APIRouter:
"""
Return a router with e-mail verification routes.
:param user_schema: Pydantic schema of a public user.
"""
return get_verify_router(self.get_user_manager, user_schema)
def get_reset_password_router(self) -> APIRouter:
"""Return a reset password process router."""
return get_reset_password_router(self.get_user_manager)
def get_auth_router(
self,
backend: AuthenticationBackend[models.UP, models.ID],
requires_verification: bool = False,
) -> APIRouter:
"""
Return an auth router for a given authentication backend.
:param backend: The authentication backend instance.
:param requires_verification: Whether the authentication
require the user to be verified or not. Defaults to False.
"""
return get_auth_router(
backend,
self.get_user_manager,
self.authenticator,
requires_verification,
)
def get_oauth_router(
self,
oauth_client: BaseOAuth2,
backend: AuthenticationBackend[models.UP, models.ID],
state_secret: SecretType,
redirect_url: str | None = None,
associate_by_email: bool = False,
is_verified_by_default: bool = False,
*,
csrf_token_cookie_name: str = CSRF_TOKEN_COOKIE_NAME,
csrf_token_cookie_path: str = "/",
csrf_token_cookie_domain: str | None = None,
csrf_token_cookie_secure: bool = True,
csrf_token_cookie_httponly: bool = True,
csrf_token_cookie_samesite: Literal["lax", "strict", "none"] = "lax",
) -> APIRouter:
"""
Return an OAuth router for a given OAuth client and authentication backend.
:param oauth_client: The HTTPX OAuth client instance.
:param backend: The authentication backend instance.
:param state_secret: Secret used to encode the state JWT.
:param redirect_url: Optional arbitrary redirect URL for the OAuth2 flow.
If not given, the URL to the callback endpoint will be generated.
:param associate_by_email: If True, any existing user with the same
e-mail address will be associated to this user. Defaults to False.
:param is_verified_by_default: If True, the `is_verified` flag will be
set to `True` on newly created user. Make sure the OAuth Provider you're
using does verify the email address before enabling this flag.
:param csrf_token_cookie_name: Name of the cookie.
:param csrf_token_cookie_path: Cookie path.
:param csrf_token_cookie_domain: Cookie domain.
:param csrf_token_cookie_secure: Whether to only send the cookie to the
server via SSL request.
:param csrf_token_cookie_httponly: Whether to prevent access to the cookie
via JavaScript.
:param csrf_token_cookie_samesite: A string that specifies the samesite
strategy for the cookie. Valid values are lax, strict and none. Defaults to lax.
"""
return get_oauth_router(
oauth_client,
backend,
self.get_user_manager,
state_secret,
redirect_url,
associate_by_email,
is_verified_by_default,
csrf_token_cookie_name=csrf_token_cookie_name,
csrf_token_cookie_path=csrf_token_cookie_path,
csrf_token_cookie_domain=csrf_token_cookie_domain,
csrf_token_cookie_secure=csrf_token_cookie_secure,
csrf_token_cookie_httponly=csrf_token_cookie_httponly,
csrf_token_cookie_samesite=csrf_token_cookie_samesite,
)
def get_oauth_associate_router(
self,
oauth_client: BaseOAuth2,
user_schema: type[schemas.U],
state_secret: SecretType,
redirect_url: str | None = None,
requires_verification: bool = False,
csrf_token_cookie_name: str = CSRF_TOKEN_COOKIE_NAME,
csrf_token_cookie_path: str = "/",
csrf_token_cookie_domain: str | None = None,
csrf_token_cookie_secure: bool = True,
csrf_token_cookie_httponly: bool = True,
csrf_token_cookie_samesite: Literal["lax", "strict", "none"] = "lax",
) -> APIRouter:
"""
Return an OAuth association router for a given OAuth client.
:param oauth_client: The HTTPX OAuth client instance.
:param user_schema: Pydantic schema of a public user.
:param state_secret: Secret used to encode the state JWT.
:param redirect_url: Optional arbitrary redirect URL for the OAuth2 flow.
If not given, the URL to the callback endpoint will be generated.
:param requires_verification: Whether the endpoints
require the users to be verified or not. Defaults to False.
:param csrf_token_cookie_name: Name of the cookie.
:param csrf_token_cookie_path: Cookie path.
:param csrf_token_cookie_domain: Cookie domain.
:param csrf_token_cookie_secure: Whether to only send the cookie to the
server via SSL request.
:param csrf_token_cookie_httponly: Whether to prevent access to the cookie
via JavaScript.
:param csrf_token_cookie_samesite: A string that specifies the samesite
strategy for the cookie. Valid values are lax, strict and none. Defaults to lax.
"""
return get_oauth_associate_router(
oauth_client,
self.authenticator,
self.get_user_manager,
user_schema,
state_secret,
redirect_url,
requires_verification,
csrf_token_cookie_name=csrf_token_cookie_name,
csrf_token_cookie_path=csrf_token_cookie_path,
csrf_token_cookie_domain=csrf_token_cookie_domain,
csrf_token_cookie_secure=csrf_token_cookie_secure,
csrf_token_cookie_httponly=csrf_token_cookie_httponly,
csrf_token_cookie_samesite=csrf_token_cookie_samesite,
)
def get_users_router(
self,
user_schema: type[schemas.U],
user_update_schema: type[schemas.UU],
requires_verification: bool = False,
) -> APIRouter:
"""
Return a router with routes to manage users.
:param user_schema: Pydantic schema of a public user.
:param user_update_schema: Pydantic schema for updating a user.
:param requires_verification: Whether the endpoints
require the users to be verified or not. Defaults to False.
"""
return get_users_router(
self.get_user_manager,
user_schema,
user_update_schema,
self.authenticator,
requires_verification,
)