newsreader/src/newsreader/accounts/models.py

74 lines
2.5 KiB
Python

from django.contrib.auth.models import AbstractUser
from django.contrib.auth.models import UserManager as DjangoUserManager
from django.db import models
from django.utils.translation import gettext as _
from django_celery_beat.models import PeriodicTask
class UserManager(DjangoUserManager):
def _create_user(self, email, password, **extra_fields):
"""
Create and save a user with the given username, email, and password.
"""
if not email:
raise ValueError("The given email must be set")
email = self.normalize_email(email)
user = self.model(email=email, **extra_fields)
user.set_password(password)
user.save(using=self._db)
return user
def create_user(self, email, password=None, **extra_fields):
extra_fields.setdefault("is_staff", False)
extra_fields.setdefault("is_superuser", False)
return self._create_user(email, password, **extra_fields)
def create_superuser(self, email, password, **extra_fields):
extra_fields.setdefault("is_staff", True)
extra_fields.setdefault("is_superuser", True)
if extra_fields.get("is_staff") is not True:
raise ValueError("Superuser must have is_staff=True.")
if extra_fields.get("is_superuser") is not True:
raise ValueError("Superuser must have is_superuser=True.")
return self._create_user(email, password, **extra_fields)
class User(AbstractUser):
email = models.EmailField(_("email address"), unique=True)
# reddit settings
reddit_refresh_token = models.CharField(max_length=255, blank=True, null=True)
reddit_access_token = models.CharField(max_length=255, blank=True, null=True)
# twitter settings
twitter_oauth_token = models.CharField(max_length=255, blank=True, null=True)
twitter_oauth_token_secret = models.CharField(max_length=255, blank=True, null=True)
# settings
auto_mark_read = models.BooleanField(
_("Auto read marking"),
default=True,
help_text=_(
"Wether posts should be marked as read after x amount of seconds of reading"
),
)
username = None
objects = UserManager()
USERNAME_FIELD = "email"
REQUIRED_FIELDS = []
def delete(self, *args, **kwargs):
tasks = PeriodicTask.objects.filter(name__contains=self.email)
tasks.delete()
return super().delete(*args, **kwargs)
@property
def has_twitter_auth(self):
return self.twitter_oauth_token and self.twitter_oauth_token_secret