उदाहरण द्वारा फ्लास्क - एक रेडिस कार्य कतार को लागू करना

ट्यूटोरियल के इस भाग में टेक्स्ट प्रोसेसिंग को हैंडल करने के लिए रेडिस टास्क क्यू को लागू करने का तरीका बताया गया है।

अपडेट:

• 02/12/2020:पायथन संस्करण 3.8.1 के साथ-साथ रेडिस, पायथन रेडिस और आरक्यू के नवीनतम संस्करणों में अपग्रेड किया गया। विवरण के लिए नीचे देखें। नवीनतम आरक्यू संस्करण में एक बग का उल्लेख करें और समाधान प्रदान करें। https बग से पहले http को हल किया।
• 03/22/2016:पायथन संस्करण 3.5.1 के साथ-साथ रेडिस, पायथन रेडिस और आरक्यू के नवीनतम संस्करणों में अपग्रेड किया गया। विवरण के लिए नीचे देखें।
• 02/22/2015:जोड़ा गया पायथन 3 समर्थन।

मुफ़्त बोनस: एक निःशुल्क फ्लास्क + पायथन वीडियो ट्यूटोरियल तक पहुंच प्राप्त करने के लिए यहां क्लिक करें जो आपको दिखाता है कि फ्लास्क वेब ऐप कैसे बनाया जाता है, चरण-दर-चरण।

याद रखें:यहां हम निर्माण कर रहे हैं—एक फ्लास्क ऐप जो किसी दिए गए URL के टेक्स्ट के आधार पर शब्द-आवृत्ति जोड़े की गणना करता है।

1. भाग एक:एक स्थानीय विकास वातावरण स्थापित करें और फिर हेरोकू पर एक मंचन और एक उत्पादन वातावरण दोनों को तैनात करें।
2. भाग दो:माइग्रेशन को संभालने के लिए SQLAlchemy और एलेम्बिक के साथ PostgreSQL डेटाबेस सेट करें।
3. भाग तीन:स्क्रैप करने के लिए बैक-एंड लॉजिक जोड़ें और फिर अनुरोधों, ब्यूटीफुल सूप और नेचुरल लैंग्वेज टूलकिट (एनएलटीके) लाइब्रेरी का उपयोग करके वेबपेज से शब्द गणना को संसाधित करें।
4. भाग चार:पाठ प्रसंस्करण को संभालने के लिए एक रेडिस कार्य कतार लागू करें। (वर्तमान )
5. भाग पांच:यह देखने के लिए कि अनुरोध संसाधित किया गया है या नहीं, बैक-एंड को लगातार मतदान करने के लिए फ्रंट-एंड पर एंगुलर सेट करें।
6. भाग छह:Heroku पर स्टेजिंग सर्वर पर पुश करें - Redis की स्थापना और एक ही Dyno पर दो प्रक्रियाओं (वेब ​​और कार्यकर्ता) को चलाने के तरीके का विवरण देना।
7. भाग सात:फ़्रंट-एंड को अधिक उपयोगकर्ता-अनुकूल बनाने के लिए अपडेट करें।
8. भाग आठ:JavaScript और D3 का उपयोग करके आवृत्ति वितरण चार्ट प्रदर्शित करने के लिए एक कस्टम कोणीय निर्देश बनाएं।

<चिह्न>कोड चाहिए? इसे रेपो से प्राप्त करें।

आवश्यकताएं स्थापित करें

उपयोग किए गए उपकरण:

• रेडिस (5.0.7)
• पायथन रेडिस (3.4.1)
• RQ (1.2.2) - कार्य कतार बनाने के लिए एक साधारण पुस्तकालय

आधिकारिक साइट से या होमब्रे के माध्यम से रेडिस को डाउनलोड और इंस्टॉल करके प्रारंभ करें (brew install redis ) एक बार स्थापित होने के बाद, रेडिस सर्वर शुरू करें:

$ redis-server

इसके बाद एक नई टर्मिनल विंडो में Python Redis और RQ इंस्टॉल करें:

$ cd flask-by-example
$ python -m pip install redis==3.4.1 rq==1.2.2
$ python -m pip freeze > requirements.txt

कार्यकर्ता सेट करें

आइए कतारबद्ध कार्यों को सुनने के लिए एक कार्यकर्ता प्रक्रिया बनाकर शुरू करें। एक नई फ़ाइल बनाएँ worker.py , और यह कोड जोड़ें:

import os

import redis
from rq import Worker, Queue, Connection

listen = ['default']

redis_url = os.getenv('REDISTOGO_URL', 'redis://localhost:6379')

conn = redis.from_url(redis_url)

if __name__ == '__main__':
    with Connection(conn):
        worker = Worker(list(map(Queue, listen)))
        worker.work()

यहां, हमने default . नामक एक कतार के बारे में सुना और Redis सर्वर से localhost:6379 . पर एक कनेक्शन स्थापित किया ।

इसे किसी अन्य टर्मिनल विंडो में सक्रिय करें:

$ cd flask-by-example
$ python worker.py
17:01:29 RQ worker started, version 0.5.6
17:01:29
17:01:29 *** Listening on default...

अब हमें अपना app.py . अपडेट करना होगा कतार में नौकरी भेजने के लिए…

अपडेट करें app.py

निम्न आयातों को app.py में जोड़ें :

from rq import Queue
from rq.job import Job
from worker import conn

फिर कॉन्फ़िगरेशन अनुभाग को अपडेट करें:

app = Flask(__name__)
app.config.from_object(os.environ['APP_SETTINGS'])
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)

q = Queue(connection=conn)

from models import *

q = Queue(connection=conn) एक रेडिस कनेक्शन स्थापित करें और उस कनेक्शन के आधार पर एक कतार शुरू करें।

टेक्स्ट प्रोसेसिंग कार्यक्षमता को हमारे इंडेक्स रूट से बाहर और count_and_save_words() नामक एक नए फ़ंक्शन में ले जाएं। . यह फ़ंक्शन एक तर्क, एक यूआरएल को स्वीकार करता है, जिसे हम अपने इंडेक्स रूट से कॉल करने पर उसे पास कर देंगे।

def count_and_save_words(url):

    errors = []

    try:
        r = requests.get(url)
    except:
        errors.append(
            "Unable to get URL. Please make sure it's valid and try again."
        )
        return {"error": errors}

    # text processing
    raw = BeautifulSoup(r.text).get_text()
    nltk.data.path.append('./nltk_data/')  # set the path
    tokens = nltk.word_tokenize(raw)
    text = nltk.Text(tokens)

    # remove punctuation, count raw words
    nonPunct = re.compile('.*[A-Za-z].*')
    raw_words = [w for w in text if nonPunct.match(w)]
    raw_word_count = Counter(raw_words)

    # stop words
    no_stop_words = [w for w in raw_words if w.lower() not in stops]
    no_stop_words_count = Counter(no_stop_words)

    # save the results
    try:
        result = Result(
            url=url,
            result_all=raw_word_count,
            result_no_stop_words=no_stop_words_count
        )
        db.session.add(result)
        db.session.commit()
        return result.id
    except:
        errors.append("Unable to add item to database.")
        return {"error": errors}


@app.route('/', methods=['GET', 'POST'])
def index():
    results = {}
    if request.method == "POST":
        # this import solves a rq bug which currently exists
        from app import count_and_save_words

        # get url that the person has entered
        url = request.form['url']
        if not url[:8].startswith(('https://', 'http://')):
            url = 'http://' + url
        job = q.enqueue_call(
            func=count_and_save_words, args=(url,), result_ttl=5000
        )
        print(job.get_id())

    return render_template('index.html', results=results)

निम्नलिखित कोड पर ध्यान दें:

job = q.enqueue_call(
    func=count_and_save_words, args=(url,), result_ttl=5000
)
print(job.get_id())

नोट: हमें count_and_save_words . आयात करने की आवश्यकता है हमारे फंक्शन index . में फंक्शन चूंकि RQ पैकेज में वर्तमान में एक बग है, जहां उसे एक ही मॉड्यूल में फ़ंक्शन नहीं मिलेंगे।

यहां हमने उस कतार का उपयोग किया है जिसे हमने पहले शुरू किया था और जिसे enqueue_call() . कहा जाता है समारोह। इसने कतार में एक नया कार्य जोड़ा और वह कार्य count_and_save_words() चला गया तर्क के रूप में URL के साथ कार्य करें। result_ttl=5000 लाइन तर्क आरक्यू को बताता है कि इस मामले में - 5,000 सेकंड के लिए नौकरी के परिणाम को कितने समय तक पकड़ना है। फिर हमने जॉब आईडी को टर्मिनल पर आउटपुट किया। यह आईडी यह देखने के लिए आवश्यक है कि क्या कार्य संसाधित हो गया है।

आइए उसके लिए एक नया मार्ग सेट करें…

परिणाम प्राप्त करें

@app.route("/results/<job_key>", methods=['GET'])
def get_results(job_key):

    job = Job.fetch(job_key, connection=conn)

    if job.is_finished:
        return str(job.result), 200
    else:
        return "Nay!", 202

आइए इसका परीक्षण करें।

सर्वर को सक्रिय करें, http://localhost:5000/ पर नेविगेट करें, URL https://realpython.com का उपयोग करें, और टर्मिनल से जॉब आईडी प्राप्त करें। फिर उस आईडी का उपयोग '/results/' एंडपॉइंट में करें - यानी, http://localhost:5000/results/ef600206-3503-4b87-a436-ddd9438f2197।

जब तक आप स्थिति की जांच करने से पहले 5,000 सेकंड से कम समय बीत चुके हैं, तब आपको एक आईडी नंबर देखना चाहिए, जो हमारे द्वारा डेटाबेस में परिणाम जोड़ने पर उत्पन्न होता है:

# save the results
try:
    from models import Result
    result = Result(
        url=url,
        result_all=raw_word_count,
        result_no_stop_words=no_stop_words_count
    )
    db.session.add(result)
    db.session.commit()
    return result.id

अब, JSON में डेटाबेस से वास्तविक परिणाम वापस करने के लिए रूट को थोड़ा रिफलेक्टर करें:

@app.route("/results/<job_key>", methods=['GET'])
def get_results(job_key):

    job = Job.fetch(job_key, connection=conn)

    if job.is_finished:
        result = Result.query.filter_by(id=job.result).first()
        results = sorted(
            result.result_no_stop_words.items(),
            key=operator.itemgetter(1),
            reverse=True
        )[:10]
        return jsonify(results)
    else:
        return "Nay!", 202

आयात जोड़ना सुनिश्चित करें:

from flask import jsonify

इसका फिर से परीक्षण करें। यदि सब कुछ ठीक रहा, तो आपको अपने ब्राउज़र में कुछ ऐसा ही दिखना चाहिए:

[
  [
    "Python", 
    315
  ], 
  [
    "intermediate", 
    167
  ], 
  [
    "python", 
    161
  ], 
  [
    "basics", 
    118
  ], 
  [
    "web-dev", 
    108
  ], 
  [
    "data-science", 
    51
  ], 
  [
    "best-practices", 
    49
  ], 
  [
    "advanced", 
    45
  ], 
  [
    "django", 
    43
  ], 
  [
    "flask", 
    41
  ]
]

आगे क्या है?

मुफ़्त बोनस: एक निःशुल्क फ्लास्क + पायथन वीडियो ट्यूटोरियल तक पहुंच प्राप्त करने के लिए यहां क्लिक करें जो आपको दिखाता है कि फ्लास्क वेब ऐप कैसे बनाया जाता है, चरण-दर-चरण।

भाग 5 में हम एक पोलर बनाने के लिए मिश्रण में एंगुलर को जोड़कर क्लाइंट और सर्वर को एक साथ लाएंगे, जो हर पांच सेकंड में /results/<job_key> पर एक अनुरोध भेजेगा। समापन बिंदु अद्यतन के लिए पूछ रहा है। डेटा उपलब्ध होने के बाद, हम इसे DOM में जोड़ देंगे।

चीयर्स!

यह स्टार्टअप एडमॉन्टन के सह-संस्थापक कैम लिंके और रियल पायथन के लोगों के बीच एक सहयोग टुकड़ा है