جنگو - هیوادار

پر شدن فیلد اتوماتیک توی فرم

توی یک فرم یک فیلد داریم بنام full_name که بر اساس first_name و last_name توی view باید ساخته بشه. برای اینکه بشه توی فرم و view ازش استفاده کرد توی fields مربوط به فرم اونو آوردیم.

چون این فیلد اتوماتیک ساخته میشه نمیخایم توی وب نشون داده بشه بنابراین expert_form.fields['full_name'].widget = forms.HiddenInput() ولی مشکل اینجاست که قبل از اینکه فرم post بشه ارور میگیریم که این فیلد باید پر بشه از طرفی Hidden هست.

راه حل این شد که باید توی __init__ فرم این رو از حالت required خارج کنیم.

def expert_add(request):
    if request.method == "POST":
        expert_form = ExpertForm(request.POST)
    else:
        expert_form = ExpertForm()

    expert_form.fields['prefix'].widget = forms.HiddenInput()
    expert_form.fields['sur_name'].widget = forms.HiddenInput()
    expert_form.fields['full_name'].widget = forms.HiddenInput()

    if request.method == "POST":
        if expert_form.is_valid():
            expert_instance = expert_form.save(commit=False)

            if expert_form.cleaned_data['is_male']:
                expert_instance.prefix = "آقای"
            else:
                expert_instance.prefix = "خانم"

            expert_instance.full_name = expert_form.cleaned_data['first_name'] + " " + expert_form.cleaned_data['last_name']

            expert_instance.save()
            messages.success(request, "...")
            return redirect('/man/expert/')

    return render(request, "man/expert/expert-add.html", {"form": expert_form})
	
	
class ExpertForm(ModelForm):
    class Meta:
        model = Expert
        fields = ["sur_name", "full_name", "prefix", "first_name", "last_name", "branch", "office", "is_male"]

    def clean_branch(self):
        if not self.cleaned_data["branch"]:
            raise ValidationError("این فیلد اجباری است.")
        return self.cleaned_data["branch"]

    def clean_office(self):
        if not self.cleaned_data["office"]:
            raise ValidationError("این فیلد اجباری است")
        return self.cleaned_data["office"]

    def clean_first_name(self):
        if not self.cleaned_data["first_name"]:
            raise ValidationError("این فیلد اجباری است")
        return self.cleaned_data["first_name"]

    def clean_last_name(self):
        if not self.cleaned_data["last_name"]:
            raise ValidationError("این فیلد اجباری است")
        return self.cleaned_data["last_name"]

    def clean(self):
        cleaned_data = super().clean()

        if cleaned_data.get("office") and cleaned_data.get("branch"):
            if cleaned_data.get("office").branch != cleaned_data.get("branch"):
                raise ValidationError("ا...")

        return cleaned_data

    def __init__(self, *args, **kwargs):
        super(ExpertForm, self).__init__(*args, **kwargs)
        self.fields['full_name'].required = False

def expert_add(request):

if request.method == "POST":

expert_form = ExpertForm(request.POST)

else:

expert_form = ExpertForm()

expert_form.fields['prefix'].widget = forms.HiddenInput()

expert_form.fields['sur_name'].widget = forms.HiddenInput()

expert_form.fields['full_name'].widget = forms.HiddenInput()

if request.method == "POST":

if expert_form.is_valid():

expert_instance = expert_form.save(commit=False)

if expert_form.cleaned_data['is_male']:

expert_instance.prefix = "آقای"

else:

expert_instance.prefix = "خانم"

expert_instance.full_name = expert_form.cleaned_data['first_name'] + " " + expert_form.cleaned_data['last_name']

expert_instance.save()

messages.success(request, "...")

return redirect('/man/expert/')

return render(request, "man/expert/expert-add.html", {"form": expert_form})

class ExpertForm(ModelForm):

class Meta:

model = Expert

fields = ["sur_name", "full_name", "prefix", "first_name", "last_name", "branch", "office", "is_male"]

def clean_branch(self):

if not self.cleaned_data["branch"]:

raise ValidationError("این فیلد اجباری است.")

return self.cleaned_data["branch"]

def clean_office(self):

if not self.cleaned_data["office"]:

raise ValidationError("این فیلد اجباری است")

return self.cleaned_data["office"]

def clean_first_name(self):

if not self.cleaned_data["first_name"]:

raise ValidationError("این فیلد اجباری است")

return self.cleaned_data["first_name"]

def clean_last_name(self):

if not self.cleaned_data["last_name"]:

raise ValidationError("این فیلد اجباری است")

return self.cleaned_data["last_name"]

def clean(self):

cleaned_data = super().clean()

if cleaned_data.get("office") and cleaned_data.get("branch"):

if cleaned_data.get("office").branch != cleaned_data.get("branch"):

raise ValidationError("ا...")

return cleaned_data

def __init__(self, *args, **kwargs):

super(ExpertForm, self).__init__(*args, **kwargs)

self.fields['full_name'].required = False

بهترین روش استفاده از فرم ها توی جنگو

فرم در جنگو قابلیت های زیادی داره و من توی پروژه های مختلف هر بار چالش هایی داشتم که به مرور متوجه شدم که این قابلیت ها خیلی میتونه کمک کنه.

1.فرض کنید که توی یک فرم از یک مدل ما میخایم که یک فیلد رو نشون کاربر نده و اون رو پنهان کنه برای اون کار باید از HiddenInput استفاده کرد link_form.fields['provider'].widget = forms.HiddenInput()

2.فرض کنید به همون فیلد فرم میخایم یه مقدار اولیه بدیم که دیگه نیاز نباشه کاربر فیلدی رو پر کنه. link_form.initial = {'provider': svc}

3.هر فیلد توی یک فرم یک queryset داره که متونید اون رو تغییر بدید و مثلا یه فیلدش رو کم کنید یا sort کنید. self.fields['provider'].queryset.order_by('name')

4.یکی از نکات مشکلسازی که وجود داره. اگر میخاید فیلدهای یک فرم رو توی view تغییراتی بهش بدید (از اون چیزی که کاربر وارد میکنه) باید اون فیلد رو save کنید و اون instance ساخته شده رو تغییر بدید و بعدش commit کنید. link_instance = link_form.save(commit=False)

5.اگر بخایم مقدار وارد شده توسط کاربر رو بگیریم باید از cleaned_data استفاده کنیم. link_form.cleaned_data['provider'].name

6.توی خود کلاس Form میتونید برای هر فیلد محدودیت هایی ایجاد کنید و در صورت تخطی از اون محدودیت ها ValidationError برگردونید.

Form و View ای که تمامی این موارد رو توش داره رو در کدهای زیر میبینید. هر بار نیاز به زدن چنین فرمهایی بودیم بد نیست یه سری به اینا بزنیم.

@login_required
@permission_required('user.can_call_service_view_custom')
def link_outgoing_add(request, service_id):
    svc = Service.objects.get(pk=service_id)

    if request.method == "POST":
        link_form = LinkADDForm(request.POST)
    else:
        link_form = LinkADDForm()

    link_form.fields['provider'].widget = forms.HiddenInput()
    link_form.initial = {'provider': svc}

    link_form.fields['consumer'].queryset = link_form.fields['consumer'].queryset.exclude(pk=svc.pk)
    link_form.fields['interface'].queryset = link_form.fields['interface'].queryset.exclude(pk=svc.pk)
    link_form.fields['interface2'].queryset = link_form.fields['interface2'].queryset.exclude(pk=svc.pk)

    if request.method == 'POST':
        if link_form.is_valid():
            link_instance = link_form.save(commit=False)

            link_instance.provider_name = convert_string(link_form.cleaned_data['provider'].name)
            link_instance.consumer_name = convert_string(link_form.cleaned_data['consumer'].name)

            if link_form.cleaned_data.get('interface'):
                link_instance.interface_name = convert_string(link_form.cleaned_data['interface'].name)
            if link_form.cleaned_data.get('interface2'):
                link_instance.interface2_name = convert_string(link_form.cleaned_data['interface2'].name)
            if link_form.cleaned_data.get('protocol'):
                link_instance.protocol_name = convert_string(link_form.cleaned_data['protocol'].name_fixed)

            link_instance.save()

            return redirect('man:link_outgoing', service_id=service_id)

    return render(request, 'man/link/link-outgoing-add.html', context={'link_form': link_form})

@login_required

@permission_required('user.can_call_service_view_custom')

def link_outgoing_add(request, service_id):

svc = Service.objects.get(pk=service_id)

if request.method == "POST":

link_form = LinkADDForm(request.POST)

else:

link_form = LinkADDForm()

link_form.fields['provider'].widget = forms.HiddenInput()

link_form.initial = {'provider': svc}

link_form.fields['consumer'].queryset = link_form.fields['consumer'].queryset.exclude(pk=svc.pk)

link_form.fields['interface'].queryset = link_form.fields['interface'].queryset.exclude(pk=svc.pk)

link_form.fields['interface2'].queryset = link_form.fields['interface2'].queryset.exclude(pk=svc.pk)

if request.method == 'POST':

if link_form.is_valid():

link_instance = link_form.save(commit=False)

link_instance.provider_name = convert_string(link_form.cleaned_data['provider'].name)

link_instance.consumer_name = convert_string(link_form.cleaned_data['consumer'].name)

if link_form.cleaned_data.get('interface'):

link_instance.interface_name = convert_string(link_form.cleaned_data['interface'].name)

if link_form.cleaned_data.get('interface2'):

link_instance.interface2_name = convert_string(link_form.cleaned_data['interface2'].name)

if link_form.cleaned_data.get('protocol'):

link_instance.protocol_name = convert_string(link_form.cleaned_data['protocol'].name_fixed)

link_instance.save()

return redirect('man:link_outgoing', service_id=service_id)

return render(request, 'man/link/link-outgoing-add.html', context={'link_form': link_form})

و فرم مربوطه

class LinkADDForm(forms.ModelForm):
    class Meta:
        model = Link
        fields = ("provider", "consumer", "interface", "interface2", "protocol", "lt", "description", "api", "frequency", "ESBVerified",
                  "ProviderVerified", "ConsumerVerified")

    def __init__(self, *args, **kwargs):
        super(LinkADDForm, self).__init__(*args, **kwargs)

        self.fields['provider'].queryset = self.fields['provider'].queryset.order_by('name')
        self.fields['consumer'].queryset = self.fields['consumer'].queryset.order_by('name')
        self.fields['interface'].queryset = self.fields['interface'].queryset.order_by('name')
        self.fields['interface2'].queryset = self.fields['interface2'].queryset.order_by('name')

    def clean(self):
        cleaned_data = super().clean()
        interface = cleaned_data.get('interface')
        interface2 = cleaned_data.get('interface2')

        if interface and interface2 and interface == interface2:
            raise ValidationError('واسط ها یکی هستند')

        return cleaned_data

    def clean_interface(self):
        interface = self.cleaned_data['interface']
        provider = self.cleaned_data['provider']
        consumer = self.cleaned_data['consumer']

        if interface == provider:
            raise ValidationError('واسط با مبدا یکی است')

        if interface == consumer:
            raise ValidationError('واسط با مقصد یکی است')

        return interface

    def clean_interface2(self):
        interface2 = self.cleaned_data['interface2']
        provider = self.cleaned_data['provider']
        consumer = self.cleaned_data['consumer']

        if interface2 == provider:
            raise ValidationError('واسط دوم با مبدا یکی است')

        if interface2 == consumer:
            raise ValidationError('واسط دوم با مقصد یکی است')

        return interface2

class LinkADDForm(forms.ModelForm):

class Meta:

model = Link

fields = ("provider", "consumer", "interface", "interface2", "protocol", "lt", "description", "api", "frequency", "ESBVerified",

"ProviderVerified", "ConsumerVerified")

def __init__(self, *args, **kwargs):

super(LinkADDForm, self).__init__(*args, **kwargs)

self.fields['provider'].queryset = self.fields['provider'].queryset.order_by('name')

self.fields['consumer'].queryset = self.fields['consumer'].queryset.order_by('name')

self.fields['interface'].queryset = self.fields['interface'].queryset.order_by('name')

self.fields['interface2'].queryset = self.fields['interface2'].queryset.order_by('name')

def clean(self):

cleaned_data = super().clean()

interface = cleaned_data.get('interface')

interface2 = cleaned_data.get('interface2')

if interface and interface2 and interface == interface2:

raise ValidationError('واسط ها یکی هستند')

return cleaned_data

def clean_interface(self):

interface = self.cleaned_data['interface']

provider = self.cleaned_data['provider']

consumer = self.cleaned_data['consumer']

if interface == provider:

raise ValidationError('واسط با مبدا یکی است')

if interface == consumer:

raise ValidationError('واسط با مقصد یکی است')

return interface

def clean_interface2(self):

interface2 = self.cleaned_data['interface2']

provider = self.cleaned_data['provider']

consumer = self.cleaned_data['consumer']

if interface2 == provider:

raise ValidationError('واسط دوم با مبدا یکی است')

if interface2 == consumer:

raise ValidationError('واسط دوم با مقصد یکی است')

return interface2

دسترسی به پارامترهای URL از توی Template Django

اگر نیاز شد که به متغیرهای توی URL دسترسی پیدا کنیم، میتونیم ببینیم که توی request چی پیدا میشه. طبیعیه که هر چیزی که توی یک فرم توی مرورگر میبینیم باید توی request پیدا بشه. برای این مورد از request.resolver_match استفاده میکنیم مثل:

<td class="bg-primary h5">
                            <a href="/man/link/p/{{request.resolver_match.kwargs.service_id}}/a/"><i class="mdi mdi-plus text-white"></i></a>
                        </td>

</td>

چند فرم و یک view در جنگو

توی یه اپلیکیشن یه فایل HTML داریم که سه تا فرم توشه ولی فقط با یه فرم داره رندر میشه. مشکل اینجاست که اینجوری پر از باگه. حتی ممکنه فرم اول رو سابمیت کنی کد فرم بعدی هم اجرا شه در حالی که اصلا لازم نیست هم از نظر performance ای یه کد اضافی داره اجرا میشه هم ممکنه دیتای فرم بعدی رو خراب کنه. برای اینکار به سه تا input فرم ها هر کدوم یه اسم یکتا دادم مثلا:

form1
.
.
.
<input type="submit" class="btn btn-lg btn-success" name="document_form_submit" value="تایید">
.
.
.
form2
.
.
.
<input type="submit" class="btn btn-lg btn-success" name="feedback_form_submit" value="ثبت">
.
.
.
form3
.
.
.
<button type="submit" class="btn btn-primary" name="comment_form_submit">ثبت</button>

form1

form2

form3

حالا توی فرم چک میکنیم که این فیلد توی request هست یا نه:

def documents_document_detail(request, doc_id):
    d = Document.objects.get(pk=doc_id)
    feedback_form = FeedbackForm()

    context = {
        'feedback_comment_form': FeedbackCommentForm(),
        'doc': d,
        'id': doc_id,
        'service': d.system,
        'form': DocumentForm(model_to_dict(d), None),
        'doc_attachments': d.attachments.filter(doc_content=0).all(),
        'report_attachments': d.attachments.filter(doc_content=1).all(),
        'comment_attachments': d.attachments.filter(doc_content=2).all(),
        'feedback_form': feedback_form,
        'comment_list': DocumentComment.objects.filter(document=d, user=request.user)
    }

    if hasattr(d, 'feedback'):
        feedback_form = FeedbackForm(initial={
            'platform_awareness': d.feedback.platform_awareness,
            'main_points': d.feedback.main_points,
            'secondary_points': d.feedback.secondary_points,
            'guids_and_recommendations': d.feedback.guids_and_recommendations
        })
        context['feedback_form'] = feedback_form

    if request.POST and 'document_form_submit' in request.POST:
        document_form = DocumentForm(request.POST, initial={'system': d.system})
        document_form.fields.pop('system')
        document_form.fields.pop('templateversion')
        document_form.fields.pop('doc_type')
        document_form.fields.pop('docdate')

        if document_form.is_valid():
            status = document_form.cleaned_data['status']
            d.status = status
            d.save()

            context['doc'] = d
            context['form'] = DocumentForm(model_to_dict(d), None)
            context['doc_attachments'] = d.attachments.filter(doc_content=0).all()
            context['report_attachments'] = d.attachments.filter(doc_content=1).all()
            context['comment_attachments'] = d.attachments.filter(doc_content=2).all()

            messages.success(request, '...')
            return render(request, "man/documents/document-detail.html", context)
        else:
            for _, error in document_form.errors.items():
                messages.error(request, error)
            return render(request, "man/documents/document-detail.html", context)

    if request.POST and 'feedback_form_submit' in request.POST:
        feedback_form = FeedbackForm(request.POST)
        if feedback_form.is_valid():
            platform_awareness = feedback_form.cleaned_data['platform_awareness']
            main_points = feedback_form.cleaned_data['main_points']
            secondary_points = feedback_form.cleaned_data['secondary_points']
            guids_and_recommendations = feedback_form.cleaned_data['guids_and_recommendations']

            if hasattr(d, 'feedback'):
                d.feedback.platform_awareness = platform_awareness
                d.feedback.main_points = main_points
                d.feedback.secondary_points = secondary_points
                d.feedback.guids_and_recommendations = guids_and_recommendations
                d.feedback.save()

                feedback_form = FeedbackForm(initial={
                    'platform_awareness': platform_awareness,
                    'main_points': main_points,
                    'secondary_points': secondary_points,
                    'guids_and_recommendations': guids_and_recommendations
                })

            else:
                f = Feedback.objects.create(
                    document=d,
                    platform_awareness=platform_awareness,
                    main_points=main_points,
                    secondary_points=secondary_points,
                    guids_and_recommendations=guids_and_recommendations
                )
                feedback_form = FeedbackForm(initial={
                    'platform_awareness': f.platform_awareness,
                    'main_points': f.main_points,
                    'secondary_points': f.secondary_points,
                    'guids_and_recommendations': f.guids_and_recommendations
                })

            context['feedback_form'] = feedback_form
            messages.success(request, '...د')
            return render(request, "man/documents/document-detail.html", context)
        else:
            for _, error in feedback_form.errors.items():
                messages.error(request, error)
                return render(request, "man/documents/document-detail.html", context)

    if request.POST and 'comment_form_submit' in request.POST:
        feedback_comment_form = FeedbackCommentForm(request.POST)
        if feedback_comment_form.is_valid():
            body = feedback_comment_form.cleaned_data['body']
            DocumentComment.objects.create(
                document=d,
                user=request.user,
                body=body,
                comment_datetime=datetime.datetime.now()
            )

            context['comment_list'] = DocumentComment.objects.filter(document=d, user=request.user)
            messages.success(request, '...')
            return render(request, "man/documents/document-detail.html", context)
        else:
            for _, error in feedback_comment_form.errors.items():
                messages.error(request, error)
                return render(request, "man/documents/document-detail.html", context)

    return render(request, "man/documents/document-detail.html", context)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

def documents_document_detail(request, doc_id):

d = Document.objects.get(pk=doc_id)

feedback_form = FeedbackForm()

context = {

'feedback_comment_form': FeedbackCommentForm(),

'doc': d,

'id': doc_id,

'service': d.system,

'form': DocumentForm(model_to_dict(d), None),

'doc_attachments': d.attachments.filter(doc_content=0).all(),

'report_attachments': d.attachments.filter(doc_content=1).all(),

'comment_attachments': d.attachments.filter(doc_content=2).all(),

'feedback_form': feedback_form,

'comment_list': DocumentComment.objects.filter(document=d, user=request.user)

}

if hasattr(d, 'feedback'):

feedback_form = FeedbackForm(initial={

'platform_awareness': d.feedback.platform_awareness,

'main_points': d.feedback.main_points,

'secondary_points': d.feedback.secondary_points,

'guids_and_recommendations': d.feedback.guids_and_recommendations

})

context['feedback_form'] = feedback_form

if request.POST and 'document_form_submit' in request.POST:

document_form = DocumentForm(request.POST, initial={'system': d.system})

document_form.fields.pop('system')

document_form.fields.pop('templateversion')

document_form.fields.pop('doc_type')

document_form.fields.pop('docdate')

if document_form.is_valid():

status = document_form.cleaned_data['status']

d.status = status

d.save()

context['doc'] = d

context['form'] = DocumentForm(model_to_dict(d), None)

context['doc_attachments'] = d.attachments.filter(doc_content=0).all()

context['report_attachments'] = d.attachments.filter(doc_content=1).all()

context['comment_attachments'] = d.attachments.filter(doc_content=2).all()

messages.success(request, '...')

return render(request, "man/documents/document-detail.html", context)

else:

for _, error in document_form.errors.items():

messages.error(request, error)

return render(request, "man/documents/document-detail.html", context)

if request.POST and 'feedback_form_submit' in request.POST:

feedback_form = FeedbackForm(request.POST)

if feedback_form.is_valid():

platform_awareness = feedback_form.cleaned_data['platform_awareness']

main_points = feedback_form.cleaned_data['main_points']

secondary_points = feedback_form.cleaned_data['secondary_points']

guids_and_recommendations = feedback_form.cleaned_data['guids_and_recommendations']

if hasattr(d, 'feedback'):

d.feedback.platform_awareness = platform_awareness

d.feedback.main_points = main_points

d.feedback.secondary_points = secondary_points

d.feedback.guids_and_recommendations = guids_and_recommendations

d.feedback.save()

feedback_form = FeedbackForm(initial={

'platform_awareness': platform_awareness,

'main_points': main_points,

'secondary_points': secondary_points,

'guids_and_recommendations': guids_and_recommendations

})

else:

f = Feedback.objects.create(

document=d,

platform_awareness=platform_awareness,

main_points=main_points,

secondary_points=secondary_points,

guids_and_recommendations=guids_and_recommendations

)

feedback_form = FeedbackForm(initial={

'platform_awareness': f.platform_awareness,

'main_points': f.main_points,

'secondary_points': f.secondary_points,

'guids_and_recommendations': f.guids_and_recommendations

})

context['feedback_form'] = feedback_form

messages.success(request, '...د')

return render(request, "man/documents/document-detail.html", context)

else:

for _, error in feedback_form.errors.items():

messages.error(request, error)

return render(request, "man/documents/document-detail.html", context)

if request.POST and 'comment_form_submit' in request.POST:

feedback_comment_form = FeedbackCommentForm(request.POST)

if feedback_comment_form.is_valid():

body = feedback_comment_form.cleaned_data['body']

DocumentComment.objects.create(

document=d,

user=request.user,

body=body,

comment_datetime=datetime.datetime.now()

)

context['comment_list'] = DocumentComment.objects.filter(document=d, user=request.user)

messages.success(request, '...')

return render(request, "man/documents/document-detail.html", context)

else:

for _, error in feedback_comment_form.errors.items():

messages.error(request, error)

return render(request, "man/documents/document-detail.html", context)

این جوری اگر درخواست POST باشه فقط اون فرمی که سابمیت شده کدش اجرا میشه و لازم هم نیست که برای هر فرم یک view جدا هم بنویسیم.

Signal Dispatcher در جنگو

Signal Dispatcher به ما این امکان رو میده که اگر توی یک اپلیکیشن اتفاقی افتاد اپلیکیشن های دیگه یا جاهای دیگه رو notify کنیم که اتفاقی انجام شه. مثلا فرض کنید که یک نفر به ما یک پیام میفرسته. اون وقت من باید یه نوتیفیکیشن دریافت کنم. روش کار این مدلیه که یک تکه کد میتونه sender باشه و چند تابع میتونن reciever باشن. در واقع وقتی یک چند تابع به یک event حساس هستن میتونن از این قابلیت استفاده کنن.

در پروژه ای وقتی مستندات در مدلی به نام Document ذخیره میشه و همزمان یک تاریخچه از این ذخیره شدن توی DocumentChangeHistory ذخیره میشه. مشکل اینه که میخام وقتی یک داکیومنت حذف میشه تاریخچه مرتبط با اون هم پاک شه.

برای این کار باید از Signal جنگو استفاده کرد. مثلا

from django.db.models.signals import post_delete
from django.dispatch import receiver
from config.models import Document, DocumentChangeHistory


@receiver(post_delete, sender=Document)
def delete_related_history(sender, instance, **kwargs):
    (DocumentChangeHistory.objects.filter(request_kwargs__has_key='doc_id').
     filter(request_kwargs__doc_id=instance.id).delete())

from django.db.models.signals import post_delete

from django.dispatch import receiver

from config.models import Document, DocumentChangeHistory

@receiver(post_delete, sender=Document)

def delete_related_history(sender, instance, **kwargs):

(DocumentChangeHistory.objects.filter(request_kwargs__has_key='doc_id').

filter(request_kwargs__doc_id=instance.id).delete())

توی کد بالا میگیم که اگر مدل Document چیزی ازش حذف شد. post_delete .اون وقت این بلاک کد اجرا بشه که در واقع از مدل تاریخچه ردیف مرتبط رو حذف میکنه.

باید توی app.py این رو تعریف کنیم. تا کار کنه.

from django.apps import AppConfig


class ManageConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'man'
	
    def ready(self):
        from .signals import delete_related_history

from django.apps import AppConfig

class ManageConfig(AppConfig):

default_auto_field = 'django.db.models.BigAutoField'

name = 'man'

def ready(self):

from .signals import delete_related_history

لینک 1

لینک 2

استقرار پروژه جنگو با بهینگی بالا

در پروژه ای که در بستر داکر روی لینوکس پیاده سازی شده است نرم افزار از performance قابل قبولی برخوردار نیست. بخشی از نرم افزار با websocket کار میکند اما اکثرا در بستر http. دیتابیس، و بخشهای دیگر مثل celery, rabbitMQ و … هر کدام یک کانتینر هستند.

ایرادات استقرار یا deployment:

جنگو در یک کانتینر با دستور runserver اجرا شده است.

کل پروژه از Daphne به عنوان اپلیکیشن وب سرور ASGI استفاده میکند در حالی که بخش کوچکی از نرم افزار asyn است و با websocket کار میکند.

وب سروری مثل Nginx یا Apache وجود ندارد که فایلهای استاتیک را ارائه دهد.

بهتر بود در بخشهای Sync از اپلیکیشن سروری مثل uWSGI که یک WSGI هست استفاده میشد.

بنابراین معماری به شکل زیر تغییر یافت:

NGINX درخواست های websocket رو به Daphne و سایر درخواست ها را به uWSGI ارسال میکند. فایل های کانفیگ از قرار زیر است.

compose.yml

version: '3.8'

services:
    elastic:
        image: elasticsearch:7.17.11
        container_name: elastic
        restart: always
        hostname: es
        environment:
            discovery.type: single-node
            ES_JAVA_OPTS: -Xms512m -Xmx512m
        ports:
            - 9200:9200
    mysql:
        image: mysql
        container_name: mysql
        restart: always
        hostname: mysql.local
        environment:
            discovery.type: single-node
            MYSQL_ROOT_PASSWORD: ...
            MYSQL_DATABASE: ...
        ports:
            - 3306:3306
        volumes:
            - mysql_data:/var/lib/mysql
    nasim:
        build:
            context: .
            dockerfile: DockerfileNasim
        hostname: nasim
        container_name: nasim
        volumes:
            - ./app:/opt/app
            - ./data:/opt/data
        expose:
            - "8000"
        environment:
            - DJANGO_SETTINGS_MODULE=core.settings
    daphne:
        build:
            context: .
            dockerfile: DockerfileNasim
        hostname: daphne
        container_name: daphne
        volumes:
            - ./app:/opt/app
            - ./data:/opt/data
        expose:
            - "8001"
        command: ["daphne", "-b", "0.0.0.0", "-p", "8001", "core.asgi:application"]
        environment:
            - DJANGO_SETTINGS_MODULE=core.settings
    nginx:
        build:
            context: .
            dockerfile: DockerfileNginx
        hostname: nginx
        container_name: nginx
        ports:
            - "80:80"
        depends_on:
            - nasim
            - daphne
        volumes:
            - ./app/static:/static
            - ./nginx/nginx.conf:/etc/nginx/nginx.conf
            
volumes:
    mysql_data:

version: '3.8'

services:

elastic:

image: elasticsearch:7.17.11

container_name: elastic

restart: always

hostname: es

environment:

discovery.type: single-node

ES_JAVA_OPTS: -Xms512m -Xmx512m

ports:

- 9200:9200

mysql:

image: mysql

container_name: mysql

restart: always

hostname: mysql.local

environment:

discovery.type: single-node

MYSQL_ROOT_PASSWORD: ...

MYSQL_DATABASE: ...

ports:

- 3306:3306

volumes:

- mysql_data:/var/lib/mysql

nasim:

build:

context: .

dockerfile: DockerfileNasim

hostname: nasim

container_name: nasim

volumes:

- ./app:/opt/app

- ./data:/opt/data

expose:

- "8000"

environment:

- DJANGO_SETTINGS_MODULE=core.settings

daphne:

build:

context: .

dockerfile: DockerfileNasim

hostname: daphne

container_name: daphne

volumes:

- ./app:/opt/app

- ./data:/opt/data

expose:

- "8001"

command: ["daphne", "-b", "0.0.0.0", "-p", "8001", "core.asgi:application"]

environment:

- DJANGO_SETTINGS_MODULE=core.settings

nginx:

build:

context: .

dockerfile: DockerfileNginx

hostname: nginx

container_name: nginx

ports:

- "80:80"

depends_on:

- nasim

- daphne

volumes:

- ./app/static:/static

- ./nginx/nginx.conf:/etc/nginx/nginx.conf

volumes:

mysql_data:

nginx.conf

# nginx/nginx.conf
events {
  worker_connections 1024;
}

http {
  include       /etc/nginx/mime.types;
  default_type  application/octet-stream;
  sendfile        on;
  keepalive_timeout  65;

  upstream nasim {
    server nasim:8000;
  }

  server {
    listen 80;

    location / {
      proxy_pass http://nasim;
      proxy_set_header Host $host;
      proxy_set_header X-Real-IP $remote_addr;
      proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_set_header X-Forwarded-Proto $scheme;
    }
	
    location /ws/ {
      proxy_pass http://daphne:8001;
      proxy_http_version 1.1;
      proxy_set_header Upgrade $http_upgrade;
      proxy_set_header Connection "upgrade";
      proxy_set_header Host $host;
      proxy_set_header X-Real-IP $remote_addr;
      proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
      proxy_set_header X-Forwarded-Proto $scheme;
    }

    location /static/ {
      alias /static/;
    }
  }
}

# nginx/nginx.conf

events {

worker_connections 1024;

}

http {

include /etc/nginx/mime.types;

default_type application/octet-stream;

sendfile on;

keepalive_timeout 65;

upstream nasim {

server nasim:8000;

}

server {

listen 80;

location / {

proxy_pass http://nasim;

proxy_set_header Host $host;

proxy_set_header X-Real-IP $remote_addr;

proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

proxy_set_header X-Forwarded-Proto $scheme;

}

location /ws/ {

proxy_pass http://daphne:8001;

proxy_http_version 1.1;

proxy_set_header Upgrade $http_upgrade;

proxy_set_header Connection "upgrade";

proxy_set_header Host $host;

proxy_set_header X-Real-IP $remote_addr;

proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

proxy_set_header X-Forwarded-Proto $scheme;

}

location /static/ {

alias /static/;

}

DockerfileNginx

# DockerfileNginx
FROM nginx:alpine

# Remove the default configuration file
RUN rm /etc/nginx/conf.d/default.conf

# Copy the nginx configuration file
COPY nginx/nginx.conf /etc/nginx/nginx.conf

# DockerfileNginx

FROM nginx:alpine

# Remove the default configuration file

RUN rm /etc/nginx/conf.d/default.conf

# Copy the nginx configuration file

COPY nginx/nginx.conf /etc/nginx/nginx.conf

DockerfileNasim

FROM python:3.11-slim-bullseye AS nasim_core

# Install system dependencies with retry logic
RUN apt-get update --fix-missing

RUN apt-get install apt-utils -y
RUN apt-get install build-essential -y
RUN apt-get install gcc -y
RUN apt-get install graphviz -y
RUN apt-get install graphviz-dev -y
RUN apt-get install vim -y
RUN apt-get install libmagickwand-dev -y
RUN apt-get install curl -y
RUN apt-get install iputils-ping -y
RUN apt-get install python3-dev -y
RUN apt-get install default-libmysqlclient-dev -y
RUN apt-get install pkg-config -y
RUN apt-get install libffi-dev -y


# Create and set work directory
RUN mkdir /opt/app && chown www-data:www-data /opt/app
WORKDIR /opt/app

COPY ./app/requirements.txt /opt/app
RUN pip install --upgrade pip
RUN pip install --upgrade -r requirements.txt --no-cache-dir
RUN pip install uwsgi

# Create logs directory and set permissions
RUN mkdir /opt/app/logs && touch /opt/app/logs/operation.log /opt/app/logs/system.log
RUN mkdir -p /opt/app/uploads/documents


RUN chmod -R 777 /opt/app/logs


# Set environment variables
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1


# Copy project
COPY ./app /opt/app/

# Run Daphne
CMD ["uwsgi", "--http", "0.0.0.0:8000", "--module", "core.wsgi:application", "--master", "--processes", "4", "--threads", "4"]

FROM python:3.11-slim-bullseye AS nasim_core

# Install system dependencies with retry logic

RUN apt-get update --fix-missing

RUN apt-get install apt-utils -y

RUN apt-get install build-essential -y

RUN apt-get install gcc -y

RUN apt-get install graphviz -y

RUN apt-get install graphviz-dev -y

RUN apt-get install vim -y

RUN apt-get install libmagickwand-dev -y

RUN apt-get install curl -y

RUN apt-get install iputils-ping -y

RUN apt-get install python3-dev -y

RUN apt-get install default-libmysqlclient-dev -y

RUN apt-get install pkg-config -y

RUN apt-get install libffi-dev -y

# Create and set work directory

RUN mkdir /opt/app && chown www-data:www-data /opt/app

WORKDIR /opt/app

COPY ./app/requirements.txt /opt/app

RUN pip install --upgrade pip

RUN pip install --upgrade -r requirements.txt --no-cache-dir

RUN pip install uwsgi

# Create logs directory and set permissions

RUN mkdir /opt/app/logs && touch /opt/app/logs/operation.log /opt/app/logs/system.log

RUN mkdir -p /opt/app/uploads/documents

RUN chmod -R 777 /opt/app/logs

# Set environment variables

ENV PYTHONDONTWRITEBYTECODE 1

ENV PYTHONUNBUFFERED 1

# Copy project

COPY ./app /opt/app/

# Run Daphne

CMD ["uwsgi", "--http", "0.0.0.0:8000", "--module", "core.wsgi:application", "--master", "--processes", "4", "--threads", "4"]

لینک های مرتبط:

Signal و Middleware توی پروسه لاگین (جلوگیری از چندین لاگین همزمان با یک نام کاربری)

توی یه پروژه خواستیم یک نام کاربری دو تا session نداشته باشه یعنی یه کاربر توی دوتا مرورگر یا دو تا سیستم لاگین نکنه. برای این کار از Middleware و Signal توی پروژه استفاده کردم.

توی Middleware.py توی app مربط کد زیر کد زیر رو زدم:

from django.contrib.auth import logout
from django.utils.deprecation import MiddlewareMixin


class PreventConcurrentLoginsMiddleware(MiddlewareMixin):
    def process_request(self, request):
        if request.user.is_authenticated:
            current_session_key = request.session.session_key
            user = request.user

            if user.session_key and user.session_key != current_session_key:
                logout(request)

            elif not user.session_key:
                user.session_key = current_session_key
                user.save()

from django.contrib.auth import logout

from django.utils.deprecation import MiddlewareMixin

class PreventConcurrentLoginsMiddleware(MiddlewareMixin):

def process_request(self, request):

if request.user.is_authenticated:

current_session_key = request.session.session_key

user = request.user

if user.session_key and user.session_key != current_session_key:

logout(request)

elif not user.session_key:

user.session_key = current_session_key

user.save()

حالا اینو توی MIDDLEWARE توی SETTING اضافه میکنیم.

Signal زیر رو اضافه میکنم که اگر کسی logtout کرد session اش پاک شه که بشه جای دیگه لاگین کنن:

from django.contrib.auth.signals import user_logged_out
from django.dispatch import receiver
from django.conf import settings
from django.contrib.auth import get_user_model


@receiver(user_logged_out)
def reset_session_key(sender, request, user, **kwargs):
    User = get_user_model()
    if user.is_authenticated:
        user.session_key = None
        user.save()

from django.contrib.auth.signals import user_logged_out

from django.dispatch import receiver

from django.conf import settings

from django.contrib.auth import get_user_model

@receiver(user_logged_out)

def reset_session_key(sender, request, user, **kwargs):

User = get_user_model()

if user.is_authenticated:

user.session_key = None

user.save()

افزودن فیلد به مدل با کمترین impact

یه مدل داریم تو جنگو که پروفایل شرکت های مختلف توی اون ذخیره میشه:

class Company(models.Model, Template):
    name = models.CharField(max_length=80, unique=True)
    name_fixed = models.CharField(max_length=80, null=True)
    name_fa = models.CharField(max_length=80)
    homepage = models.URLField(max_length=200, null=True)
    tel = models.CharField(max_length=80, null=True)
    color_code = ColorField(max_length=7, default="#1EABFF")
    active = models.BooleanField(default=True)

class Company(models.Model, Template):

name = models.CharField(max_length=80, unique=True)

name_fixed = models.CharField(max_length=80, null=True)

name_fa = models.CharField(max_length=80)

homepage = models.URLField(max_length=200, null=True)

tel = models.CharField(max_length=80, null=True)

color_code = ColorField(max_length=7, default="#1EABFF")

active = models.BooleanField(default=True)

لازم شد که لوگو شرکت هم اضافه بشه اما از اونجایی که این مدل جاهای مختلف استفاده شده بود و توی فرمها و view های مختلف ازش استفاده شده اگه یه فیلد برای لوگو بهش اضافه کنیم قطعا باید خیلی از جاهای کد هم تغییر بدیم و احتمالا باگ هایی تولید بشن و از اونجایی که این سیستم براش تست نوشته نشده ممکن اطفاقات بدی بیوفته.

بنابراین رفتم دنبال اینکه چطور میشه بدون تغییر این مدل لوگو رو اضافه کنم. حداقل طوری که اگر جایی از این مدل داره استفاده میشه دیگه فیلدهاش عوض نشه. راه های مختلفی وجود داره ولی بهترینش بنظرم اینه که یه مدل جدید برای لوگو بسازیم و ارتباط OneToOne بینشون برقرار کنیم.

class CompanyLogo(models.Model):
    company = models.OneToOneField(Company, on_delete=models.CASCADE, related_name='logo')
    logo = models.ImageField(upload_to="company_images/", blank=True, null=True)

class CompanyLogo(models.Model):

company = models.OneToOneField(Company, on_delete=models.CASCADE, related_name='logo')

logo = models.ImageField(upload_to="company_images/", blank=True, null=True)

و اینطوری ازش استفاده کنم

def vendor_detail(request, vendor_id):
    company = Company.objects.get(pk=vendor_id)
    company_logo = CompanyLogo.objects.filter(company=company).first()

    context = {
        'company': company,
        'company_logo': company_logo,
    }




{% if company_logo %}
                            <img src="{{ company_logo.logo.url }}" alt="{{ company.name }} Logo" width="150" height="100">

def vendor_detail(request, vendor_id):

company = Company.objects.get(pk=vendor_id)

company_logo = CompanyLogo.objects.filter(company=company).first()

context = {

'company': company,

'company_logo': company_logo,

}

{% if company_logo %}

Coroutine و Awaitables

Coroutine در واقع یک تابع هست که میتونیم اونو pause کنیم. فرض کنیم دو تا تابع داریم و میخایم کنترل اجرا بین این دو تا جابجا بشه مثل ping pong مثلا تابع اول چند خط رو اجرا کنه بعد کنترل رو به تابع دوم بده و اون چند خط اجرا کنه و … در واقع دو تا routine که اینجا بهشون میگیم تابع با هم cooperate کنن.

برای تریفش توی پایتون از async کلیدواژه استفاده میکنیم. که اون وقت اون تابع coroutine میشه. میتونیم هر جای کد coroutine ازش بخایم وایسه تا اجرای یه تابع دیگه تموم شه برای این کار await میکنیم.

مقایسه Celery و Channels و Singnal

در جدول زیر مقایسه Celery و Django Channels و Django Signals رو میبینیم

Here’s a comparison of Django Channels, Django Signals, and Celery, presented in a tabular format, focusing on their use cases, features, and suitable scenarios:

Feature/Aspect	Django Channels	Django Signals	Celery
Primary Use Case	Real-time communication, WebSockets, and long-lived connections	Decoupled event handling within Django applications	Asynchronous task queue/job queue for running background tasks
Communication	WebSockets, HTTP2, server-sent events	In-process communication within the same Django app	Distributed communication, allowing task distribution across multiple workers
Concurrency	Handles concurrent connections and async processing	Synchronous, works within the Django request-response cycle	Handles concurrent tasks, designed for parallel processing
Scalability	Scalable for handling many concurrent connections	Limited by the Django process	Highly scalable, can distribute tasks across multiple servers
Real-time Updates	Yes, supports real-time updates	No	No, tasks are run asynchronously but not in real-time
Event Handling	Handles real-time events	Handles events within the Django lifecycle (e.g., model save, user login)	Handles events as background tasks
Persistent Connections	Yes, supports long-lived connections (e.g., chat apps)	No, operates within the request-response cycle	No, designed for discrete, short-lived tasks
Integration	Integrates with WebSockets, WebRTC, etc.	Integrates with Django ORM and lifecycle hooks	Integrates with message brokers (RabbitMQ, Redis, etc.)
Typical Use Cases	Chat applications, live notifications, live dashboards	Sending signals on model changes, user actions	Sending emails, processing data, generating reports, scheduled tasks
Example	Live chat app with instant messaging	Logging changes to models when saved	Sending periodic email reports, processing user-uploaded files in the background
Fault Tolerance	Basic fault tolerance, relies on infrastructure	Basic, depends on Django’s fault tolerance	High fault tolerance, can retry failed tasks, distributed task execution
Dependencies	Redis, RabbitMQ (for channel layers)	None beyond Django	Message broker (e.g., RabbitMQ, Redis), result backend (optional)
Ease of Use	Requires understanding of asynchronous programming	Simple to use with Django	Requires setup of a message broker and understanding of task queues

Detailed Use Case Examples

Django Channels:
- Chat Application: Users can send and receive messages in real-time, with updates being pushed to all connected clients instantly.
- Live Notifications: Real-time notifications for events such as new comments, likes, or mentions in a social media application.
- Live Dashboard: Real-time updates of metrics and analytics in a monitoring dashboard.
Django Signals:
- Model Change Notifications: Automatically trigger a function to log changes or send an email when a model instance is saved or deleted.
- User Activity Tracking: Capture user login, logout, or profile update events to maintain an activity log.
- Post-Processing: Perform additional actions after saving a model, such as updating related records or sending a confirmation email.
Celery:
- Email Sending: Send emails asynchronously to avoid blocking the main thread during the request-response cycle.
- Data Processing: Handle large data processing tasks, such as generating reports or processing user-uploaded files, in the background.
- Scheduled Tasks: Run periodic tasks like clearing expired sessions, generating nightly reports, or updating external APIs.

By comparing these tools, you can choose the most appropriate solution based on the specific requirements and characteristics of your project.