तो आज, मैं सेलेरी के एक विकल्प के बारे में बात करूंगा जिसका नाम है ह्यूई , जो सेलेरी की तुलना में बहुत आसान सेटअप के साथ आता है और सेलेरी की तुलना में आकार में बहुत छोटा है।
मैंने ह्यूई को आजमाने का निर्णय इसलिए लिया क्योंकि कभी-कभी मुझे कुछ सामान्य कार्यों को करने में सेलेरी के साथ कुछ समस्याओं का सामना करना पड़ा, क्योंकि दस्तावेज़ीकरण बहुत अच्छा नहीं था।
जो लोग नहीं जानते कि सेलेरी क्या है या जिन्होंने पहले इसका उपयोग नहीं किया है, उनके लिए बता दें कि ह्युई एक अतुल्यकालिक कार्य पंक्ति है जो आपको पृष्ठभूमि में निर्धारित कार्य या लंबे समय तक चलने वाले कार्य करने की अनुमति देती है।
हम निम्नलिखित पैकेज स्थापित करेंगे:
निम्नलिखित ब्लॉग के साथ एक GitHub repo भी दिया गया है जिसका उपयोग आप हमारे द्वारा बनाए जाने वाले डेमो प्रोजेक्ट का परीक्षण करने के लिए कर सकते हैं।
रेपो देखने के लिए यहां क्लिक करें।
टर्मिनल खोलें, और डायरेक्टरी बनाने के लिए निम्नलिखित टाइप करें; आप इस चरण को छोड़ सकते हैं और इसे फ़ाइल एक्सप्लोरर से ही कर सकते हैं।
mkdir huey_demo
आइए सबसे पहले अपनी परियोजना निर्भरताओं को स्थापित करने के लिए एक वर्चुअलएनवी बनाएं:
python -m venv venv
वर्चुअलएन्व (लिनक्स) सक्रिय करें:
source venv/bin/activate
सभी निर्भरताओं को स्थापित करने के लिए टर्मिनल में निम्नलिखित कमांड टाइप करें:
pip install Django==4.0.4 redis==4.2.2 huey==2.4.3
इस लेख को लिखते समय, ये वे संस्करण थे जिनके साथ मैंने इस सेटअप का परीक्षण किया था, भविष्य में नवीनतम संस्करण के अनुसार किसी भी अपडेट के लिए गिटहब रेपो पर नज़र रखें।
टर्मिनल में निम्नलिखित कमांड टाइप करके Django प्रोजेक्ट बनाएं:
django-admin startproject django_huey_demo
निर्देशिका को Django प्रोजेक्ट निर्देशिका में बदलें:
cd django_huey_demo
हमारी परियोजना के अंतर्गत ऐप बनाएं:
python manage.py startapp demo
प्रोजेक्ट settings.py
में बनाए गए ऐप को शामिल करें, निम्नलिखित परिवर्तन करें:
INSTALLED_APPS = [ # Existing Apps "demo.apps.DemoConfig", # <== Add this line ]
settings.py
में डीबग मोड को False
पर सेट करें:
DEBUG=False
हम Debug को False पर सेट कर रहे हैं ताकि हम देख सकें कि Huey उत्पादन में कैसे चलता है, इस पर बाद में और अधिक जानकारी दी जाएगी।
अब जबकि हमने अपनी परियोजना तैयार कर ली है, तो यह सही समय है कि हम आपको बताएं कि हम आज क्या बनाएंगे।
हम रोजाना वर्डनिक एपीआई से "वर्ड ऑफ द डे" प्राप्त करेंगे। फिर हम शब्द, उसकी परिभाषा और वाक्य में शब्द का एक उदाहरण हमारे डेटाबेस में संग्रहीत करेंगे।
हम ह्यूई का उपयोग करके एक आवधिक कार्य स्थापित करेंगे जो दिन का शब्द लाएगा और उसे संग्रहीत करेगा।
शब्द को संग्रहीत करने के लिए हम उसी का एक Django मॉडल बनाएंगे।
आप API कुंजी प्राप्त करने के लिए इस गाइड का अनुसरण कर सकते हैं।
हमें अपने प्रोजेक्ट के इंस्टॉल किए गए ऐप्स में Huey को जोड़ना है, इसलिए settings.py
फ़ाइल में निम्नलिखित परिवर्तन करें:
INSTALLED_APPS = [ # Existing apps 'huey.contrib.djhuey', # <== Add this line ]
हमें Huey के लिए Redis को इंस्टॉल करना होगा ताकि इसमें कतारबद्ध कार्यों के बारे में जानकारी संग्रहीत की जा सके जैसा कि हम Celery के साथ भी करते थे। आप अपने विशिष्ट ऑपरेटिंग सिस्टम के आधार पर Redis को इंस्टॉल करने के लिए निम्न लिंक का संदर्भ ले सकते हैं।
यदि आप Docker का उपयोग करने में सहज हैं, तो आप निम्नलिखित कमांड का उपयोग कर सकते हैं:
docker run --name redis_huey -p 6379:6379 -d redis
डिफ़ॉल्ट रूप से, Huey localhost:6379
पर चल रहे Redis सर्वर से कनेक्ट करने का प्रयास करेगा। यदि यह मौजूद नहीं है, तो यह एक त्रुटि उत्पन्न करेगा।
अपनी demo/models.py
फ़ाइल में निम्नलिखित कोड जोड़ें:
from django.db import models class Word(models.Model): word = models.CharField(max_length=200) part_of_speech = models.CharField(max_length=100) definition = models.TextField() example = models.TextField() def __str__(self): return self.word
माइग्रेशन करें:
python manage.py makemigrations demo
माइग्रेशन लागू करें:
python manage.py migrate demo
डेमो ऐप डायरेक्टरी में tasks.py
नाम की एक फ़ाइल बनाएँ। हमने अपनी फ़ाइल का नाम tasks.py
इसलिए रखा है ताकि Huey को हमारे रजिस्टर्ड ऐप्स में मौजूद टास्क को अपने आप खोजने में मदद मिल सके, अगर हमने अपनी फ़ाइल का नाम इसके अलावा कुछ और रखा होता, तो हमें अपने टास्क को मैन्युअली रजिस्टर करना पड़ता। अगर आप और जानना चाहते हैं, तो आप Huey डॉक्यूमेंटेशन यहाँ देख सकते हैं।
कार्य परिभाषा लिखने से पहले, हमें एक अतिरिक्त निर्भरता requests
स्थापित करने की आवश्यकता है। अपने टर्मिनल में निम्नलिखित टाइप करके इसे स्थापित करें:
pip install requests==2.27.1
अब, कोड आता है:
import requests from django.conf import settings from huey import crontab from huey.contrib.djhuey import db_periodic_task from demo.models import Word @db_periodic_task(crontab(hour="18", minute="00")) def fetch_daily_word(): r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}") data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )
अपनी परियोजना सेटिंग्स में निम्नलिखित पंक्ति जोड़ें:
WORDNIK_API_KEY = "api-key-here"
यह कोड ब्लॉक समझने में बहुत कठिन हो सकता है, इसलिए आइए इसमें दी गई बातों को एक-एक करके समझते हैं:
ह्युई डेकोरेटर
from huey.contrib.djhuey import db_periodic_task
यह ह्यूई द्वारा प्रदान किया गया एक डेकोरेटर है जो डेटाबेस के साथ काम करने वाले आवधिक कार्यों को पंजीकृत करता है, यह डेकोरेटर कार्य पूरा होने पर डेटाबेस कनेक्शन को स्वचालित रूप से बंद कर देता है, अधिक जानकारी के लिए, आप यहां देख सकते हैं।
क्रोनटैब अनुसूची
@db_periodic_task(crontab(hour="18", minute="00"))
हम अपने आवधिक कार्य डेकोरेटर को तर्क crontab(hour="18", minute="00")
दे रहे हैं, यह Huey को हर दिन शाम 6 बजे हमारे कार्य को चलाने के लिए कहता है। आप अपने crontab शेड्यूल बनाने के लिए इस वेबसाइट का उपयोग कर सकते हैं, मैं हर बार इसका उपयोग करता हूँ।
वर्डनिक एपीआई कुंजी
from django.conf import settings # Usage ## settings.WORDNIK_API_KEY
from django.conf import settings
हमारे प्रोजेक्ट सेटिंग्स से किसी भी डेटा को आयात करने का मानक तरीका है, यह उन मामलों में उपयोगी है जहां हमारे पास अलग-अलग वातावरणों के लिए कई सेटिंग्स फ़ाइलें सेट अप हैं, इसलिए यह बिना किसी चिंता के जान जाएगा कि कौन सी फ़ाइल चुननी है। यह पता लगाता है कि हम DJANGO_SETTINGS_MODULE
पर्यावरण चर से कौन सी सेटिंग फ़ाइल का उपयोग कर रहे हैं। लेकिन आपको इन विवरणों के बारे में चिंता करने की ज़रूरत नहीं है।
फिर हम अपने वर्डनिक एपीआई कॉल में कुंजी का उपयोग करते हैं।
वर्डनिक एपीआई कॉल
r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}")
यहां, हम प्रमाणीकरण के लिए अपनी API कुंजी पास करते समय Wordnik API को GET अनुरोध करने के लिए अनुरोध मॉड्यूल का उपयोग कर रहे हैं।
डेटाबेस में शब्द संग्रहीत करना
data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )
API प्रतिक्रिया को पार्स करने के बाद, हम अपने डेटाबेस में शब्द परिभाषा संग्रहीत कर रहे हैं। हम यहाँ create
विधि के बजाय get_or_create
विधि का उपयोग कर रहे हैं ताकि हम अपने डेटाबेस में उसी शब्द की कई प्रतियाँ न बनाएँ यदि वह शब्द कभी Wordnik API द्वारा दोहराया जाता है।
वर्डनिक एपीआई प्रतिक्रिया
वर्ड ऑफ़ द डे एंडपॉइंट के लिए वर्डनिक एपीआई प्रतिक्रिया इस प्रकार दिखती है। संक्षिप्तता के उद्देश्य से प्रतिक्रिया के कुछ अप्रासंगिक अनुभागों को छोटा कर दिया गया है।
{ "word": "stolon", "definitions": [ { "source": "ahd-5", "text": "A long thin stem that usually grows horizontally along the ground and produces roots and shoots at widely spaced nodes, as in a strawberry plant.", "note": null, "partOfSpeech": "noun" }, // More definitions here... ], "publishDate": "2022-05-08T03:00:00.000Z", "examples": [ { "title": "4.1 Nursery establishment", "text": "A stolon is a stem that grows along the ground, producing at its nodes new plants with roots and upright stems.", // Additional data here... }, // More examples here... ], // Additional fields here... }
आप अपने टर्मिनल में निम्नलिखित कमांड टाइप करके ह्यूई वर्कर को प्रारंभ कर सकते हैं:
python manage.py run_huey
आप उपरोक्त कमांड में कई फ्लैग पास कर सकते हैं, जिससे कंसोल में लॉग की जाने वाली जानकारी बदल जाएगी, जैसे:
-v, --verbose
- विस्तृत लॉगिंग (DEBUG स्तर शामिल है)-q, --quiet
- न्यूनतम लॉगिंग-S, --simple
- सरल लॉगिंग प्रारूप ("समय संदेश")
लॉगिंग के लिए अन्य विभिन्न विकल्पों को देखने के लिए, यहां दस्तावेज़ देखें।
आप कार्य के भीतर क्या कार्य कर रहे हैं, इस पर निर्भर करते हुए, ह्यूई कई कार्य डेकोरेटर्स के साथ आता है।
मैं नीचे संक्षेप में बताऊंगा कि ये सभी क्या करते हैं।
यहां सभी डेकोरेटर्स के लिए आयात विवरण दिया गया है:
from huey.contrib.djhuey import task, periodic_task, db_task, db_periodic_task
task
: एक नियमित कार्य.periodic_task
: जब आप किसी कार्य को शेड्यूल के आधार पर आवधिक रूप से चलाना चाहते हैं।db_task
: जब आप अपने कार्य के भीतर DB संचालन करना चाहते हैं।db_periodic_task
: जब आप किसी आवधिक कार्य में DB संचालन करना चाहते हैं।मैं आपको कुछ और उदाहरण दिखाता हूँ कि आप अपने कार्यों को शेड्यूल करने के लिए crontab का उपयोग कैसे कर सकते हैं।
crontab(minute='*/3')
कार्य को हर तीन मिनट में चलाने के लिए शेड्यूल करेगा।crontab(hour='*/3', minute='5')
एक ऐसा कार्य बनाएगा जो हर तीसरे घंटे के 5 मिनट बाद चलेगा।crontab(minute='00', hour='10', month='*/2', day_of_week='*/5')
एक कार्य बनाएगा जो प्रत्येक 2वें माह के सप्ताह के प्रत्येक 5वें दिन प्रातः 10:00 बजे चलेगा। उदाहरण के लिए, आपके पास tasks.py
के अंदर निम्नलिखित कार्य परिभाषित है:
from huey.contrib.djhuey import task @task() def count(): for i in range(10): print(i)
अब, यदि आप इस कार्य को कॉल करना चाहते हैं, लेकिन इसे 5 सेकंड के बाद चलाना चाहते हैं, तो आप निम्नलिखित कर सकते हैं:
count.schedule(delay=5)
delay
पैरामीटर सेकंड में मान लेता है, इसलिए यदि आप इसे 5 मिनट के बाद निष्पादित करना चाहते हैं, तो 300 सेकंड निर्दिष्ट करें।
मान लीजिए आप हमारे मौजूदा कार्य में निम्नलिखित तर्क जोड़ते हैं:
@db_periodic_task(crontab(hour="18", minute="00"), retries=2) def fetch_daily_word(): r = requests.get( f"https://api.wordnik.com/v4/words.json/wordOfTheDay?api_key={settings.WORDNIK_API_KEY}") if r.status_code != 200: raise Exception("Unable to fetch data from Wordnik API") ## Add this logic else: data = r.json() Word.objects.get_or_create( word=data["word"], part_of_speech=data["definitions"][0]["partOfSpeech"], definition=data["definitions"][0]["text"], example=data["examples"][0]["text"] )
इसलिए, हमने प्रतिक्रिया के स्टेटस कोड की जांच करने के लिए तर्क जोड़ा, और यदि यह 200 के अलावा कुछ और है, तो यह उस कार्य को 2 बार तक पुनः प्रयास करेगा। लेकिन ये पुनः प्रयास दो प्रयासों के बीच किसी भी समय अंतराल के बिना होंगे। अब, क्या होगा यदि आप इस कार्य के कई प्रयासों में देरी करना चाहते हैं? हम retry_delay
तर्क को पास करके ऐसा कर सकते हैं, यह सेकंड में मान स्वीकार करता है।
@db_periodic_task(crontab(hour="18", minute="00"), retries=2, retry_delay=10)
इससे कई प्रयासों के बीच 10 सेकंड का विलंब होगा।
Huey एक डिफ़ॉल्ट सेटिंग के साथ आता है जो Django में विकास के दौरान Huey के साथ काम करना आसान बनाता है। इसलिए, जब भी आपके settings.py
फ़ाइल में DEBUG=True
मौजूद होता है, तो कार्य नियमित फ़ंक्शन कॉल की तरह ही सिंक्रोनस रूप से निष्पादित किए जाएँगे। इसका उद्देश्य विकास या परीक्षण चलाते समय Redis और एक अतिरिक्त उपभोक्ता प्रक्रिया दोनों को चलाने से बचना है। आप इसके बारे में यहाँ और अधिक पढ़ सकते हैं।
इसके लिए हमें settings.py
में निम्नलिखित पंक्ति जोड़नी होगी:
HUEY = {}
हालाँकि, यदि आप इस व्यवहार को ओवरराइड करना चाहते हैं, तो आप इसके बजाय निम्नलिखित Huey कॉन्फ़िगरेशन जोड़ सकते हैं:
HUEY = { "immediate": False }
यदि आपके पास settings.py
में उल्लिखित उपरोक्त कॉन्फ़िगरेशन है, और DEBUG=True
है, तो Huey को आपसे Redis सेट अप करने और run_huey
कमांड का उपयोग करके Huey Worker चलाने की आवश्यकता होगी।
अजवाइन की तुलना में ह्युई के बारे में कुछ अवलोकन इस प्रकार हैं:
run_huey
कमांड का उपयोग करके केवल एक सेवा चलाने की आवश्यकता होती है।