Egor fureunoir Gorbunov
9898212855
Fixes: 1) Fixed missing `exc_info` flag in critical and error logs to provide richer error context; 2) Addressed redundant code and unused imports in the logging logic for cleaner execution. Extra: Refactored `LOGGING` configuration for improved readability and runtime adaptability; optimized related log-handling logic throughout the codebase.
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
import logging
|
|
import traceback
|
|
|
|
from django.contrib import auth
|
|
from django.contrib.auth.base_user import BaseUserManager
|
|
from django.contrib.auth.hashers import make_password
|
|
|
|
from core.models import Address, Order
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class UserManager(BaseUserManager):
|
|
use_in_migrations = True
|
|
|
|
@staticmethod
|
|
def handle_unregistered_entities(user):
|
|
try:
|
|
orders = set()
|
|
mark_business = False
|
|
for order in Order.objects.filter(attributes__icontains=user.email):
|
|
if not order.user:
|
|
order.user = user
|
|
order.save()
|
|
orders.add(order.uuid)
|
|
if type(order.attributes) is not dict:
|
|
order.attributes = {}
|
|
if order.attributes.get("is_business"):
|
|
mark_business = True
|
|
if user.phone_number:
|
|
for order in Order.objects.filter(attributes__icontains=user.phone_number):
|
|
if not order.user:
|
|
order.user = user
|
|
order.save()
|
|
orders.add(order.uuid)
|
|
for address in Address.objects.filter(billing_address_order__in=orders):
|
|
if not address.user:
|
|
address.user = user
|
|
address.save()
|
|
for address in Address.objects.filter(shipping_address_order__in=orders):
|
|
if not address.user:
|
|
address.user = user
|
|
address.save()
|
|
if type(user.attributes) is not dict:
|
|
user.attributes = {}
|
|
if mark_business:
|
|
user.attributes.update({"is_business": True})
|
|
user.save()
|
|
return user
|
|
except Exception as e:
|
|
logger.error(e)
|
|
logger.error(traceback.format_exc())
|
|
|
|
# noinspection PyUnusedLocal
|
|
def _create_user(self, email, password, **extra_fields):
|
|
email = self.normalize_email(email)
|
|
# noinspection PyShadowingNames
|
|
user = self.model(email=email, **extra_fields)
|
|
user.password = make_password(password)
|
|
user.save(using=self._db)
|
|
return self.handle_unregistered_entities(user)
|
|
|
|
# noinspection PyUnusedLocal
|
|
def create_user(self, email=None, password=None, **extra_fields):
|
|
extra_fields.setdefault("is_staff", False)
|
|
extra_fields.setdefault("is_superuser", False)
|
|
return self._create_user(email, password, **extra_fields)
|
|
|
|
# noinspection PyUnusedLocal
|
|
def create_superuser(self, email=None, password=None, **extra_fields):
|
|
extra_fields.setdefault("is_staff", True)
|
|
extra_fields.setdefault("is_superuser", True)
|
|
if not extra_fields.get("is_staff"):
|
|
raise ValueError("Superuser must have is_staff=True.")
|
|
if not extra_fields.get("is_superuser"):
|
|
raise ValueError("Superuser must have is_superuser=True.")
|
|
# noinspection PyShadowingNames
|
|
user = self._create_user(email, password, **extra_fields)
|
|
user.is_active = True
|
|
user.is_verified = True
|
|
user.save()
|
|
return user
|
|
|
|
# noinspection PyUnusedLocal
|
|
def with_perm(self, perm, is_active=True, include_superusers=True, backend=None, obj=None):
|
|
if backend is None:
|
|
# noinspection PyCallingNonCallable
|
|
backends = auth._get_backends(return_tuples=True)
|
|
if len(backends) == 1:
|
|
backend, _ = backends[0]
|
|
else:
|
|
raise ValueError(
|
|
"You have multiple authentication backends configured and "
|
|
"therefore must provide the `backend` argument."
|
|
)
|
|
elif not isinstance(backend, str):
|
|
raise TypeError(f"backend must be a dotted import path string (got {backend}).")
|
|
else:
|
|
backend = auth.load_backend(backend)
|
|
if hasattr(backend, "with_perm"):
|
|
return backend.with_perm(
|
|
perm,
|
|
is_active=is_active,
|
|
include_superusers=include_superusers,
|
|
obj=obj,
|
|
)
|
|
return self.none()
|