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
|
import random
import string
from asgiref.sync import sync_to_async
from django.conf import settings
from django.core.mail import send_mail
from django.db import models
from django.template.loader import render_to_string
from core.models import Config
from stator.models import State, StateField, StateGraph, StatorModel
class PasswordResetStates(StateGraph):
new = State(try_interval=300)
sent = State()
new.transitions_to(sent)
@classmethod
async def handle_new(cls, instance: "PasswordReset"):
"""
Sends the password reset email.
"""
reset = await instance.afetch_full()
if reset.new_account:
await sync_to_async(send_mail)(
subject=f"{Config.system.site_name}: Confirm new account",
message=render_to_string(
"emails/account_new.txt",
{
"reset": reset,
"config": Config.system,
"settings": settings,
},
),
from_email=settings.SERVER_EMAIL,
recipient_list=[reset.user.email],
)
else:
await sync_to_async(send_mail)(
subject=f"{Config.system.site_name}: Reset password",
message=render_to_string(
"emails/password_reset.txt",
{
"reset": reset,
"config": Config.system,
"settings": settings,
},
),
from_email=settings.SERVER_EMAIL,
recipient_list=[reset.user.email],
)
return cls.sent
class PasswordReset(StatorModel):
"""
A password reset for a user (this is also how we create accounts)
"""
state = StateField(PasswordResetStates)
user = models.ForeignKey(
"users.user",
on_delete=models.CASCADE,
related_name="password_resets",
)
token = models.CharField(max_length=500, unique=True)
new_account = models.BooleanField()
created = models.DateTimeField(auto_now_add=True)
updated = models.DateTimeField(auto_now=True)
@classmethod
def create_for_user(cls, user):
return cls.objects.create(
user=user,
token="".join(random.choice(string.ascii_lowercase) for i in range(42)),
new_account=not user.password,
)
### Async helpers ###
async def afetch_full(self):
"""
Returns a version of the object with all relations pre-loaded
"""
return await PasswordReset.objects.select_related(
"user",
).aget(pk=self.pk)
|
