# pylint: disable=unused-argument
"""Django signal receivers for account app."""
from django.contrib.auth.signals import user_logged_in
from django.core import serializers
from django.db.models.signals import post_delete, post_save
from django.dispatch import receiver
from django.forms.models import model_to_dict
from smarter.apps.account.models import (
Account,
User,
UserProfile,
)
from smarter.apps.account.utils import get_cached_default_account
from smarter.common.helpers.console_helpers import formatted_text
from smarter.lib import json, logging
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
from .manifest.transformers.secret import SecretTransformer
from .models import Secret
from .signals import (
secret_accessed,
secret_created,
secret_deleted,
secret_inializing,
secret_ready,
secret_saved,
secret_updated,
)
[docs]
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.RECEIVER_LOGGING)
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
module_prefix = f"{__name__}"
[docs]
@receiver(user_logged_in)
def user_logged_in_receiver(sender, request, user: User, **kwargs):
"""
Signal receiver for user login.
Verify that a UserProfile record exists for the user.
If not, create one with the default account.
"""
logger.info("%s User logged in: %s", formatted_text(f"{module_prefix}.user_logged_in()"), user)
try:
UserProfile.objects.get(user=user)
except UserProfile.DoesNotExist:
account = get_cached_default_account()
UserProfile.objects.create(name=user.username, user=user, account=account)
logger.info("Created UserProfile for user: %s with default account: %s", user, account)
except UserProfile.MultipleObjectsReturned:
# this is fine. the same user can have multiple UserProfiles if they belong to multiple accounts
pass
[docs]
@receiver(post_save, sender=User)
def user_post_save(sender: User, instance: User, created, **kwargs):
"""
Signal receiver for created/saved of User model.
Assumed to be called on all logins since Django's
default behavior is to update the last_login field on
each login, which triggers a save.
"""
from smarter.apps.dashboard.context_processors import cache_invalidations
logger.debug(
"%s User post_save: %s, created: %s",
formatted_text(f"{module_prefix}.user_post_save()"),
instance,
created,
)
try:
user_profile = UserProfile.get_cached_object(user=instance)
cache_invalidations(user_profile=user_profile)
except UserProfile.DoesNotExist:
pass
except UserProfile.MultipleObjectsReturned:
# this is fine. the same user can have multiple UserProfiles if they belong to multiple accounts
pass
[docs]
@receiver(post_delete, sender=User)
def user_post_delete(sender: User, instance: User, **kwargs):
"""Signal receiver for deleted of User model."""
logger.info(
"%s User post_delete: %s, id: %s",
formatted_text(f"{module_prefix}.user_post_delete()"),
instance,
instance.id, # type: ignore
)
[docs]
@receiver(post_save, sender=UserProfile)
def user_profile_post_save(sender: UserProfile, instance: UserProfile, created, **kwargs):
"""Signal receiver for created/saved of UserProfile model."""
logger.debug(
"%s UserProfile post_save: %s, created: %s",
formatted_text(f"{module_prefix}.user_profile_post_save()"),
instance,
created,
)
[docs]
@receiver(post_delete, sender=UserProfile)
def user_profile_post_delete(sender: UserProfile, instance: UserProfile, **kwargs):
"""Signal receiver for deleted of UserProfile model."""
logger.info(
"%s UserProfile: %s, id: %s",
formatted_text(f"{module_prefix}.user_profile_post_delete()"),
instance,
instance.id, # type: ignore
)
[docs]
@receiver(post_save, sender=Account)
def account_post_save(sender: Account, instance: Account, created, **kwargs):
"""Signal receiver for created/saved of Account model."""
model_prefix = formatted_text(f"{module_prefix}.account_post_save()")
account_json = json.dumps(model_to_dict(instance))
if created:
logger.info("%s Account created: %s", model_prefix, account_json)
else:
logger.debug("%s Account updated: %s", model_prefix, account_json)
logger.debug(
"%s invalidating cache for Account: %s", formatted_text(f"{module_prefix}.account_post_save()"), instance
)
[docs]
@receiver(post_delete, sender=Account)
def account_post_delete(sender: Account, instance: Account, **kwargs):
"""Signal receiver for deleted of Account model."""
logger.info(
"%s Account post_delete: %s, id: %s",
formatted_text(f"{module_prefix}.account_post_delete()"),
instance,
instance.id, # type: ignore
)
[docs]
@receiver(post_save, sender=Secret)
def secret_post_save(sender: Secret, instance: Secret, created, **kwargs):
"""Signal receiver for created/saved of Secret model."""
secret_json = json.dumps(model_to_dict(instance))
logger.debug(
"%s Secret: %s, id: %s created: %s, user_profile: %s",
formatted_text(f"{module_prefix}.secret_post_save()"),
secret_json,
instance.id, # type: ignore
created,
instance.user_profile,
)
[docs]
@receiver(post_delete, sender=Secret)
def secret_post_delete(sender: Secret, instance: Secret, **kwargs):
"""Signal receiver for deleted of Secret model."""
logger.info(
"%s Secret: %s, id: %s",
formatted_text(f"{module_prefix}.secret_post_delete()"),
instance,
instance.id, # type: ignore
)
[docs]
@receiver(secret_created)
def secret_created_receiver(sender, secret: Secret, **kwargs):
"""Signal receiver for secret_created signal."""
logger.info(
"%s.%s Secret: %s id: %s, user_profile: %s",
formatted_text(f"{module_prefix}.secret_created()"),
type(sender),
str(secret),
secret.id, # type: ignore
secret.user_profile,
)
[docs]
@receiver(secret_deleted)
def secret_deleted_receiver(sender, secret_id, secret_name, **kwargs):
"""Signal receiver for secret_deleted signal."""
logger.info(
"%s.%s Secret: %s, name: %s",
formatted_text(f"{module_prefix}.secret_deleted()"),
type(sender),
secret_id,
secret_name,
)
[docs]
@receiver(secret_ready)
def secret_ready_receiver(sender, secret: SecretTransformer, **kwargs):
"""Signal receiver for secret_ready signal."""
logger.info(
"%s.%s Secret: %s, id: %s, user_profile: %s",
formatted_text(f"{module_prefix}.secret_ready()"),
type(sender),
str(secret),
secret.id,
secret.user_profile,
)
[docs]
@receiver(secret_accessed)
def secret_accessed_receiver(sender, secret: Secret, user_profile: UserProfile, **kwargs):
"""Signal receiver for secret_accessed signal."""
logger.info(
"%s.%s Secret: %s, id: %s, user_profile: %s",
formatted_text(f"{module_prefix}.secret_accessed()"),
type(sender),
str(secret),
secret.id, # type: ignore
user_profile,
)
[docs]
@receiver(secret_inializing)
def secret_inializing_receiver(sender, secret_name: str, user_profile: UserProfile, **kwargs):
"""Signal receiver for secret_inializing signal."""
logger.info(
"%s.%s name: %s, user_profile: %s",
formatted_text(f"{module_prefix}.secret_inializing()"),
type(sender),
secret_name,
user_profile,
)
[docs]
@receiver(secret_saved)
def secret_saved_receiver(sender, secret: SecretTransformer, user_profile: UserProfile, **kwargs):
"""Signal receiver for secret_saved signal."""
if not secret.secret:
raise ValueError("secret.secret is None in secret_saved_receiver")
json_data = serializers.serialize("json", [secret.secret])
tags = list(secret.secret.tags_list) if secret and hasattr(secret.secret, "tags") else []
logger.info(
"%s.%s Secret: %s, id: %s, user_profile: %s, dump: %s, tags: %s",
formatted_text(f"{module_prefix}.secret_saved()"),
type(sender),
str(secret),
secret.id,
user_profile,
json_data,
tags,
)
[docs]
@receiver(secret_updated)
def secret_updated_receiver(sender, secret: SecretTransformer, user_profile: UserProfile, **kwargs):
"""Signal receiver for secret_updated signal."""
if not secret.secret:
raise ValueError("secret.secret is None in secret_updated_receiver")
json_data = serializers.serialize("json", [secret.secret])
tags = list(secret.secret.tags_list) if secret and hasattr(secret.secret, "tags") else []
logger.info(
"%s.%s secret_updated signal received. instance: %s, id: %s, user_profile: %s, dump: %s, tags: %s",
formatted_text(f"{module_prefix}.secret_updated()"),
type(sender),
str(secret),
secret.id,
user_profile,
json_data,
tags,
)