Core - Usage Examples

from django.db import models
from nextango.apps.core.models import SanityIntegratedModel

class MyDomainModel(SanityIntegratedModel):
    """UUID PK, timestamps, and Sanity sync come from the base class."""
    name = models.CharField(max_length=255)
    description = models.TextField(blank=True)

    class Meta:
        ordering = ['-created_at']

instance.name = 'Updated Name'
instance.save()  # django_revision bumps; sync_status becomes 'out_of_sync'

if instance.needs_sync_to_sanity():
    push_to_sanity(instance)
    instance.mark_synced_to_sanity(sanity_rev='rev-abc123')

instance.get_sync_lag()  # 0 after a successful sync

# Django-owned model (_is_sanity_sot = False), deleted from internal code:
result = instance.safe_delete(origin='django_internal', hard_delete=True)

if result.action == 'blocked':
    for blocker in result.dependency_check.blockers:
        print(blocker.model, blocker.relation, blocker.count)
elif result.action == 'hard_deleted':
    print('gone:', result.reason)

from nextango.apps.core.validators import InputSanitizer

clean_text = InputSanitizer.sanitize_text(request.data['notes'], max_length=500)
clean_code = InputSanitizer.sanitize_identifier(request.data['store_code'])
clean_email = InputSanitizer.sanitize_email(request.data['email'])

from django.core.exceptions import ValidationError
from nextango.apps.core.validators import FinancialValidator

try:
    amount = FinancialValidator.validate_payment_amount('149.99')
except ValidationError as e:
    return Response({'error': str(e)}, status=400)

from django.db import models
from nextango.apps.core.fields import EncryptedJSONField

class ProcessorConfig(models.Model):
    credentials = EncryptedJSONField()

config = ProcessorConfig()
config.credentials = {'api_key': 'secret', 'merchant_id': 'm_123'}
config.save()  # stored encrypted

config.refresh_from_db()
config.credentials['api_key']  # 'secret', decrypted transparently

from nextango.apps.core.utils.cache import cache_result

@cache_result(timeout=600, key_prefix='products', vary_on=['store_id'])
def get_products_for_display(store_id=None):
    return list(Product.objects.filter(is_active=True).values('id', 'name'))

from rest_framework_extensions.cache.decorators import cache_response
from nextango.apps.core.utils.cache import make_cache_key_func, invalidate_cache_pattern

class BrandViewSet(viewsets.ModelViewSet):
    @cache_response(timeout=3600, key_func=make_cache_key_func('brand'))
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

# In signals.py:
@receiver(post_save, sender=Brand)
def invalidate_brand_cache(sender, instance, **kwargs):
    invalidate_cache_pattern('brand:')

from nextango.apps.core.utils.performance import log_execution_time

@log_execution_time('ReportService.build_report')
def build_report(store_id):
    ...

from nextango.apps.core.utils.analytics import parse_date_range, parse_store_filter

def get(self, request):
    start_date, end_date = parse_date_range(request, default_days=30)
    store_q = parse_store_filter(request)
    qs = SaleTransaction.objects.filter(timestamp__date__range=(start_date, end_date))
    if store_q:
        qs = qs.filter(store_q)

from nextango.apps.core.exceptions import handle_view_exception, BusinessLogicException

class MyViewSet(viewsets.ModelViewSet):
    @handle_view_exception
    def create(self, request, *args, **kwargs):
        if not request.data.get('store'):
            raise BusinessLogicException('Store is required', error_code='STORE_REQUIRED')
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)  # returns 400 with field detail
        ...