# pylint: disable=unused-argument
"""Django signal receivers for account app."""
import logging
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,
Charge,
DailyBillingRecord,
User,
UserProfile,
)
from smarter.apps.account.signals import broker_ready
from smarter.apps.account.utils import get_cached_default_account
from smarter.common.helpers.console_helpers import formatted_text
from smarter.lib import json
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
from smarter.lib.manifest.broker import AbstractBroker
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.info(
"%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.info(
"%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.info("%s Account updated: %s", model_prefix, account_json)
logger.info(
"%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=Charge)
def charge_post_save(sender: Charge, instance: Charge, created, **kwargs):
"""Signal receiver for created/saved of Charge model."""
charge_json = json.dumps(model_to_dict(instance))
logger.info(
"%s Charge post_save: %s, created: %s",
formatted_text(f"{module_prefix}.charge_post_save()"),
charge_json,
created,
)
[docs]
@receiver(post_save, sender=DailyBillingRecord)
def daily_billing_record_post_save(sender: DailyBillingRecord, instance: DailyBillingRecord, created, **kwargs):
"""Signal receiver for created/saved of DailyBillingRecord model."""
daily_billing_record_json = json.dumps(model_to_dict(instance))
logger.info(
"%s DailyBillingRecord: %s, created: %s",
formatted_text(f"{module_prefix}.daily_billing_record_post_save()"),
daily_billing_record_json,
created,
)
[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.info(
"%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,
)
[docs]
@receiver(broker_ready)
def broker_ready_receiver(sender, broker: AbstractBroker, **kwargs):
"""Signal receiver for broker_ready signal."""
logger.info(
"%s %s %s for %s is ready.",
formatted_text(f"{module_prefix}.broker_ready()"),
broker.kind,
str(broker),
broker.name,
)