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
|
from __future__ import annotations
import functools
from typing import TYPE_CHECKING, Any, cast
import strawberry
from asgiref.sync import async_to_sync
from django.contrib import auth
from django.contrib.auth.password_validation import validate_password
from django.core.exceptions import ValidationError
from strawberry_django.auth.utils import get_current_user
from strawberry_django.mutations import mutations, resolvers
from strawberry_django.mutations.fields import DjangoCreateMutation
from strawberry_django.optimizer import DjangoOptimizerExtension
from strawberry_django.resolvers import django_resolver
from strawberry_django.utils.requests import get_request
try:
# Django-channels is not always used/intalled,
# therefore it shouldn't be it a hard requirement.
from channels import auth as channels_auth
except ModuleNotFoundError:
channels_auth = None
if TYPE_CHECKING:
from django.contrib.auth.base_user import AbstractBaseUser
from strawberry.types import Info
@django_resolver
def resolve_login(info: Info, username: str, password: str) -> AbstractBaseUser:
request = get_request(info)
user = auth.authenticate(request, username=username, password=password)
if user is None:
raise ValidationError("Incorrect username/password")
try:
auth.login(request, user)
except AttributeError:
# ASGI in combo with websockets needs the channels login functionality.
# to ensure we're talking about channels, let's veriy that our
# request is actually channelsrequest
try:
scope = request.consumer.scope # type: ignore
async_to_sync(channels_auth.login)(scope, user) # type: ignore
# According to channels docs you must save the session
scope["session"].save()
except (AttributeError, NameError):
# When Django-channels is not installed,
# this code will be non-existing
pass
return user
@django_resolver
def resolve_logout(info: Info) -> bool:
user = get_current_user(info)
ret = user.is_authenticated
try:
request = get_request(info)
auth.logout(request)
except AttributeError:
try:
scope = request.consumer.scope # type: ignore
async_to_sync(channels_auth.logout)(scope) # type: ignore
except (AttributeError, NameError):
# When Django-channels is not installed,
# this code will be non-existing
pass
return ret
class DjangoRegisterMutation(DjangoCreateMutation):
def create(self, data: dict[str, Any], *, info: Info):
model = cast("type[AbstractBaseUser]", self.django_model)
assert model is not None
password = data.pop("password")
validate_password(password)
# Do not optimize anything while retrieving the object to update
with DjangoOptimizerExtension.disabled():
return resolvers.create(
info,
model,
data,
key_attr=self.key_attr,
full_clean=self.full_clean,
pre_save_hook=lambda obj: obj.set_password(password),
)
login = functools.partial(strawberry.mutation, resolver=resolve_login)
logout = functools.partial(strawberry.mutation, resolver=resolve_logout)
register = mutations.create if TYPE_CHECKING else DjangoRegisterMutation
|
