Python Async Programming: सम्पूर्ण मार्गदर्शिका

जब आपका Python स्क्रिप्ट API रिस्पॉन्स, डेटाबेस क्वेरी, या फ़ाइल ऑपरेशन पूरे होने का इंतज़ार करता है, तो वह समय अक्सर यूँ ही निकल जाता है। Python async प्रोग्रामिंग के साथ, आपका कोड एक साथ कई कार्य संभाल सकता है। यानी जब एक ऑपरेशन प्रतीक्षा में हो, तो बाकी आगे बढ़ते रहें—निष्क्रिय पलों को उत्पादक काम में बदलते हुए, और अक्सर मिनटों की प्रतीक्षा को कुछ सेकंड में समेट दें।

इस गाइड में, मैं छोटे-छोटे प्रोजेक्ट बनाते हुए आपको Python में async प्रोग्रामिंग की बुनियादी बातें सिखाऊँगा। आप देखेंगे कि coroutines, event loop, और async I/O आपके कोड को कितना अधिक उत्तरदायी बना सकते हैं।

यदि आप asynchronous वेब API बनाना सीखना चाहते हैं, तो FastAPI पर यह कोर्स अवश्य देखें।

Python Async Programming क्या है?

पारंपरिक synchronous Python में, आपका कोड एक समय में एक ही लाइन निष्पादित करता है। उदाहरण के लिए, जब आप किसी API को कॉल करते हैं, तो आपका प्रोग्राम रिस्पॉन्स का इंतज़ार करता है। यदि इसमें दो सेकंड लगते हैं, तो पूरा प्रोग्राम दो सेकंड तक स्थिर रहता है। Asynchronous प्रोग्रामिंग आपके कोड को API कॉल शुरू करने और फिर अन्य कार्यों पर आगे बढ़ने देती है।

जब रिस्पॉन्स आता है, तो आपका कोड वहीं से फिर शुरू हो जाता है जहाँ रुका था। हर ऑपरेशन का इंतज़ार करने के बजाय, आप एक साथ कई ऑपरेशन चला सकते हैं। यह खासकर तब मायने रखता है जब आपका कोड डेटाबेस, APIs, या फ़ाइल सिस्टम जैसे बाहरी सिस्टम के जवाब का इंतज़ार करता है। 

इसे काम में लाने के लिए, Python की async प्रणाली कुछ मूल अवधारणाएँ इस्तेमाल करती है:

• Coroutines: फंक्शन जिन्हें async def से परिभाषित किया जाता है, def के बजाय। ये निष्पादन को रोक और फिर शुरू कर सकते हैं, इसलिए इंतज़ार वाले ऑपरेशन के लिए उपयुक्त हैं।

• await: यह कीवर्ड Python को बताता है, "इस coroutine को तब तक रोको जब तक यह ऑपरेशन पूरा न हो जाए, पर इस बीच अन्य कोड चलने दो।"

• Event loop: इंजन जो आपकी सभी coroutines को मैनेज करता है, यह तय करता है कि किसे कब चलाना है और कब उनके बीच स्विच करना है।

• Tasks: concurrent निष्पादन के लिए लिपटी coroutines। आप asyncio.create_task() के साथ इन्हें बनाते हैं ताकि कई ऑपरेशन एक साथ चल सकें।

भ्रम से बचने के लिए कि async प्रोग्रामिंग क्या कर सकती है (और क्या नहीं), इसे ध्यान में रखें:

• Async I/O-bound कार्यों के साथ सबसे अच्छा काम करती है जैसे HTTP रिक्वेस्ट, डेटाबेस क्वेरी, और फ़ाइल ऑपरेशन, जहाँ आपका कोड बाहरी सिस्टम का इंतज़ार करता है।

• Async CPU-bound कार्यों में मदद नहीं करती जैसे जटिल गणना या डेटा प्रोसेसिंग, जहाँ आपका कोड इंतज़ार नहीं बल्कि सक्रिय रूप से गणना कर रहा होता है।

इन अवधारणाओं को पक्का करने का सबसे अच्छा तरीका है वास्तविक async कोड लिखना। अगले भाग में, आप अपना पहला async फ़ंक्शन बनाएँगे और देखेंगे कि coroutines और event loop साथ मिलकर कैसे काम करते हैं।

आपका पहला Python Async फ़ंक्शन

Async कोड लिखने से पहले, एक सामान्य synchronous फ़ंक्शन देखें जो कुछ करने से पहले इंतज़ार करता है:

import time

def greet_after_delay():
    print("Starting...")
    time.sleep(2)  # Blocks for 2 seconds
    print("Hello!")

greet_after_delay()संचालित द्वारा

Starting...
Hello!संचालित द्वारा

यह फ़ंक्शन काम करता है, पर time.sleep(2) पूरा प्रोग्राम ब्लॉक कर देता है। उन दो सेकंड के दौरान कुछ और नहीं चल सकता।

अब इसका async संस्करण देखें:

import asyncio

async def greet_after_delay():
    print("Starting...")
    await asyncio.sleep(2)  # Pauses, but doesn't block
    print("Hello!")

asyncio.run(greet_after_delay())संचालित द्वारा

Starting...
Hello!संचालित द्वारा

आउटपुट एक जैसा दिखता है, पर अंदरूनी प्रक्रिया अलग है। तीन बदलावों ने इसे async बनाया:

1. def की जगह async def ने इसे coroutine घोषित किया।

2. time.sleep(2) की जगह await asyncio.sleep(2) ने बिना ब्लॉक किए विराम दिया।

3. asyncio.run() ने event loop शुरू कर coroutine चलाया।

ध्यान दें कि asyncio.sleep() खुद एक async फ़ंक्शन है, इसलिए इसे await चाहिए। यह एक अहम नियम है: हर async फ़ंक्शन को await के साथ ही कॉल करना चाहिए। चाहे वह asyncio.sleep() जैसा बिल्ट-इन हो या आपका खुद का—await भूलने पर वह वास्तव में चलेगा ही नहीं।

अभी async संस्करण तेज़ नहीं दिखता। क्योंकि हमारे पास केवल एक ही कार्य है। असली फ़ायदा तब मिलता है जब आप कई coroutines एक साथ चलाते हैं—अगले भाग में यही देखेंगे।

एक और ज़रूरी बात: आप async फ़ंक्शन को साधारण फ़ंक्शन की तरह सीधे कॉल नहीं कर सकते। चलिए आज़माते हैं:

result = greet_after_delay()
print(result)
print(type(result))संचालित द्वारा

<coroutine object greet_after_delay at 0x...>
<class 'coroutine'>संचालित द्वारा

greet_after_delay() को कॉल करने पर एक coroutine ऑब्जेक्ट मिलता है, न कि परिणाम। फ़ंक्शन चलता ही नहीं। इसे चलाने के लिए asyncio.run() या किसी अन्य फ़ंक्शन के अंदर await चाहिए।

Event loop कैसे काम करता है

Event loop async प्रोग्रामिंग का इंजन है। यह आपकी coroutines को मैनेज करता है और तय करता है कि कब क्या चलेगा। async greet_after_delay() चलाने पर चरण-दर-चरण यह होता है:

1. asyncio.run() एक event loop बनाता है।

2. Event loop greet_after_delay() शुरू करता है।

3. "Starting..." प्रिंट होता है।

4. await asyncio.sleep(2) पर पहुँचते हैं → coroutine ठहरता है।

5. Event loop जाँचता है: "कोई और कार्य है?" (अभी नहीं)।

6. 2 सेकंड बीतते हैं, sleep पूरा होता है।

7. Event loop greet_after_delay() फिर शुरू करता है।

8. "Hello!" प्रिंट होता है।

9. फ़ंक्शन पूरा → event loop बंद।

चरण 5 पर async रोचक हो जाता है। एक coroutine हो तो करने को कुछ नहीं। पर कई coroutines हों, तो event loop एक के इंतज़ार में दूसरे काम पर स्विच कर लेता है। दो सेकंड के sleep के दौरान बैठने के बजाय, यह और कोड चला सकता है।

Event loop को ट्रैफ़िक कंट्रोलर समझें। यह किसी कार को तेज़ नहीं बनाता—यह ट्रैफ़िक को चलाए रखता है, ताकि एक कार रुकी हो तो दूसरी आगे बढ़ जाए।

आम async भूल: await भूल जाना

शुरुआती लोगों की एक आम गलती है किसी async फ़ंक्शन के अंदर coroutine कॉल करते समय await भूल जाना:

import asyncio

async def get_message():
    await asyncio.sleep(1)
    return "Hello!"

async def main():
    message = get_message()  # Missing await!
    print(message)

asyncio.run(main())संचालित द्वारा

<coroutine object get_message at 0x...>
RuntimeWarning: coroutine 'get_message' was never awaitedसंचालित द्वारा

await के बिना, आपको रिटर्न वैल्यू की जगह coroutine ऑब्जेक्ट मिलता है। Python यह चेतावनी भी देता है कि coroutine कभी awaited नहीं हुआ।

समाधान सरल है:

async def main():
    message = await get_message()  # Added await
    print(message)

asyncio.run(main())संचालित द्वारा

Hello!संचालित द्वारा

जब भी आपको unawaited coroutine का RuntimeWarning दिखे, जाँचें कि हर async कॉल पर await लगाया है।

Concurrent Python Async Tasks 

पिछले भाग में, हमने एक sync फ़ंक्शन को async में बदला। पर यह तेज़ नहीं था। क्योंकि हमने केवल एक coroutine चलाई। Async की असली ताकत तब दिखती है जब आप कई coroutines एक साथ चलाते हैं।

Sequential await फिर भी sequential क्यों है

आप सोच सकते हैं कि कई async फ़ंक्शन कॉल अपने-आप साथ-साथ चलेंगे। पर देखें, यदि हम greet_after_delay() तीन बार कॉल करें तो क्या होता है:

import asyncio
import time

async def greet_after_delay(name):
    print(f"Starting {name}...")
    await asyncio.sleep(2)
    print(f"Hello, {name}!")

async def main():
    start = time.perf_counter()
    
    await greet_after_delay("Alice")
    await greet_after_delay("Bob")
    await greet_after_delay("Charlie")
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())संचालित द्वारा

Starting Alice...
Hello, Alice!
Starting Bob...
Hello, Bob!
Starting Charlie...
Hello, Charlie!
Total time: 6.01 secondsसंचालित द्वारा

दो-दो सेकंड के तीन कार्यों में छह सेकंड लगे। हर await अगली लाइन पर बढ़ने से पहले अपनी coroutine के पूरा होने की प्रतीक्षा करता है। कोड async है, पर चलना sequential है।

async कार्यों को साथ-साथ चलाना: asyncio.gather() का उपयोग 

Coroutines को एक साथ चलाने के लिए asyncio.gather() का उपयोग करें। यह कई coroutines लेकर उन्हें concurrent चलाता है:

async def main():
    start = time.perf_counter()
    
    await asyncio.gather(
        greet_after_delay("Alice"),
        greet_after_delay("Bob"),
        greet_after_delay("Charlie"),
    )
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())संचालित द्वारा

Starting Alice...
Starting Bob...
Starting Charlie...
Hello, Alice!
Hello, Bob!
Hello, Charlie!
Total time: 2.00 secondsसंचालित द्वारा

छह की जगह दो सेकंड। तीनों coroutines तुरंत शुरू हुईं, साथ-साथ सोईं, और साथ ही पूरी हुईं। केवल एक बदलाव से 3x गति मिली।

आउटपुट क्रम पर ध्यान दें: सभी "Starting..." संदेश पहले आते हैं, फिर "Hello..." संदेश। यह दिखाता है कि सभी coroutines एक ही दो-सेकंड की खिड़की में चल रहीं हैं, न कि एक-दूसरे का इंतज़ार कर रही हैं।

asyncio.gather() आपको परिणाम उसी क्रम में देता है जिस क्रम में आपने coroutines पास की थीं। यदि आपकी coroutines वैल्यू लौटाती हैं, तो उन्हें कैप्चर कर सकते हैं:

async def fetch_number(n):
    await asyncio.sleep(1)
    return n * 10

async def main():
    results = await asyncio.gather(
        fetch_number(1),
        fetch_number(2),
        fetch_number(3),
    )
    print(results)

asyncio.run(main())संचालित द्वारा

[10, 20, 30]संचालित द्वारा

नतीजे क्रम में आते हैं [10, 20, 30], जो gather() को पास की गई coroutines के क्रम से मेल खाते हैं।

aiohttp के साथ Python Async HTTP रिक्वेस्ट

अब तक हम देरी को सिमुलेट करने के लिए asyncio.sleep() इस्तेमाल कर रहे थे। अब असली HTTP रिक्वेस्ट करें। आप शायद requests लाइब्रेरी उठाएँगे, पर यह यहाँ काम नहीं करेगी। requests synchronous है और event loop को ब्लॉक करती है—async का मकसद ही खत्म।

इसके बजाय aiohttp का उपयोग करें—यह इसी उद्देश्य के लिए बना async HTTP क्लाइंट है।

aiohttp का परिचय

aiohttp से URL कैसे फ़ेच करें:

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch("https://example.com")
    print(f"Fetched {len(html)} characters")

asyncio.run(main())संचालित द्वारा

Fetched 513 charactersसंचालित द्वारा

ध्यान दें कि यहाँ दो nested async with ब्लॉक हैं। प्रत्येक अलग संसाधन मैनेज करता है—उन्हें समझना aiohttp को सही से उपयोग करने की कुंजी है।

aiohttp में ClientSession कैसे काम करता है

aiohttp से रिक्वेस्ट करने पर चरण-दर-चरण यह होता है:

1. aiohttp.ClientSession() एक कनेक्शन पूल बनाता है (शुरू में खाली)।

2. session.get(url) पूल जाँचता है: "क्या इस होस्ट से कोई खुला कनेक्शन है?"

3. यदि कनेक्शन नहीं है, तो नया TCP कनेक्शन और SSL हैंडशेक बनते हैं।

4. HTTP रिक्वेस्ट भेजी जाती है, और हम रिस्पॉन्स हेडर का इंतज़ार करते हैं।

5. रिस्पॉन्स ऑब्जेक्ट कनेक्शन को होल्ड करता है।

6. await response.text() नेटवर्क से बॉडी डेटा पढ़ता है।

7. अंदरूनी async with से बाहर निकलते ही: कनेक्शन पूल में लौटता है (खुला रहता है!).

8. उसी होस्ट पर अगली रिक्वेस्ट पूल से कनेक्शन री-यूज़ करती है (चरण 3 स्किप)।

9. बाहरी async with से बाहर: सभी पूल्ड कनेक्शन बंद हो जाते हैं।

चरण 7 और 8 मुख्य बात हैं। कनेक्शन पूल रिक्वेस्ट्स के बीच कनेक्शन ज़िंदा रखता है। उसी होस्ट पर अगली रिक्वेस्ट TCP और SSL हैंडशेक को पूरी तरह स्किप कर देती है।

यह इसलिए मायने रखता है क्योंकि नया कनेक्शन स्थापित करना धीमा होता है। TCP हैंडशेक में सर्वर तक एक राउंड-ट्रिप लगता है। SSL हैंडशेक में दो और। लैटेंसी पर निर्भर करते हुए, यह पहले बाइट भेजने से पहले ही 100–300ms ले सकता है।

सभी रिक्वेस्ट के लिए shared session का उपयोग

अब आप समझ गए होंगे कि हर रिक्वेस्ट के लिए नई session बनाना समस्या क्यों है:

# Wrong: new session for each request
async def fetch_bad(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    results = await asyncio.gather(*[fetch_bad(url) for url in urls])संचालित द्वारा

fetch_bad() की हर कॉल एक नई session और खाली पूल बनाती है। हर रिक्वेस्ट फुल हैंडशेक लागत चुकाती है, जबकि सभी एक ही होस्ट पर जा रही हैं।

समाधान है—एक session बनाकर उसे fetch फ़ंक्शन को पास करें:

# Right: reuse a single session
async def fetch_good(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch_good(session, url) for url in urls])संचालित द्वारा

shared session के साथ, पहली रिक्वेस्ट कनेक्शन बनाती है, बाकी नौ उसे री-यूज़ करती हैं। दस में एक हैंडशेक।

Async HTTP रिक्वेस्ट उदाहरण: Hacker News स्क्रैपिंग

इसे Hacker News API के साथ आज़माएँ। यह API asynchronous व्यवहार दिखाने के लिए उत्तम है क्योंकि स्टोरी फ़ेच करने में कई रिक्वेस्ट लगती हैं। यदि आप Python में REST APIs पर नए हैं, तो बुनियादी बातें सीखने के लिए Python APIs: A Guide to Building and Using APIs देखें।

Hacker News API की संरचना:

• https://hacker-news.firebaseio.com/v0/topstories.json स्टोरी IDs की सूची देता है (सिर्फ संख्याएँ)

• https://hacker-news.firebaseio.com/v0/item/{id}.json एक स्टोरी का विवरण देता है

10 स्टोरी पाने के लिए 11 रिक्वेस्ट चाहिए: IDs की सूची के लिए एक, फिर हर स्टोरी के लिए एक। यहीं async प्रोग्रामिंग चमकती है।

पहले देखें कि पहली स्टोरी लाने पर API क्या लौटाता है:

import aiohttp
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def main():
    async with aiohttp.ClientSession() as session:
        # Get top story IDs
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        print(f"Found {len(story_ids)} stories")
        print(f"First 5 IDs: {story_ids[:5]}")
        
        # Fetch first story details
        first_id = story_ids[0]
        async with session.get(f"{HN_API}/item/{first_id}.json") as response:
            story = await response.json()
        
        print(f"\nStory structure:")
        for key, value in story.items():
            print(f"  {key}: {repr(value)[:50]}")

asyncio.run(main())संचालित द्वारा

Found 500 stories
First 5 IDs: [46051449, 46055298, 46021577, 46053566, 45984864]

Story structure:
  by: 'mikeayles'
  descendants: 22
  id: 46051449
  kids: [46054027, 46053889, 46053275, 46053515, 46053002,
  score: 217
  text: 'I got DOOM running in KiCad by rendering it with 
  time: 1764108815
  title: 'Show HN: KiDoom – Running DOOM on PCB Traces'
  type: 'story'
  url: 'https://www.mikeayles.com/#kidoom'संचालित द्वारा

API 500 स्टोरी IDs लौटाती है, और हर स्टोरी में title, url, score, और by (लेखक) जैसे फ़ील्ड होते हैं। 

कई नतीजे sequential बनाम concurrent फ़ेच करना

अब 10 स्टोरी sequential फ़ेच करें:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        stories = []
        for story_id in story_ids[:10]:
            story = await fetch_story(session, story_id)
            stories.append(story)
        elapsed = time.perf_counter() - start
        
        print(f"Sequential: Fetched {len(stories)} stories in {elapsed:.2f} seconds")

asyncio.run(main())संचालित द्वारा

Sequential: Fetched 10 stories in 2.41 secondsसंचालित द्वारा

अब वही स्टोरी concurrent फ़ेच करें:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        tasks = [fetch_story(session, story_id) for story_id in story_ids[:10]]
        stories = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - start
        
        print(f"Concurrent: Fetched {len(stories)} stories in {elapsed:.2f} seconds")
        print("\nTop 3 stories:")
        for story in stories[:3]:
            print(f"  - {story.get('title', 'No title')}")

asyncio.run(main())संचालित द्वारा

Concurrent: Fetched 10 stories in 0.69 seconds

Top 3 stories:
  - Show HN: KiDoom – Running DOOM on PCB Traces
  - AWS is 10x slower than a dedicated server for the same price [video]
  - Surprisingly, Emacs on Android is pretty goodसंचालित द्वारा

Concurrent संस्करण 3.5x तेज़ है। हर रिक्वेस्ट के पूरा होने का इंतज़ार करने के बजाय, सभी 10 रिक्वेस्ट एक साथ चलती हैं। असली नेटवर्क I/O में async का लाभ यहीं दिखता है।

Python Async Error Handling और Rate Limiting

डेटा को concurrent फ़ेच करते समय कई चीज़ें ग़लत हो सकती हैं। आप बहुत अधिक रिक्वेस्ट भेजकर सर्वर को ओवरवhelm कर सकते हैं। कुछ रिक्वेस्ट हमेशा के लिए अटक सकती हैं। कुछ सीधा फ़ेल हो सकती हैं। और जब विफलताएँ हों, तो रिकवरी रणनीति चाहिए।

यह भाग उन्हीं चिंताओं को उनके होने के क्रम में कवर करता है: कितनी रिक्वेस्ट बाहर जाएँ, समय सीमाएँ तय करना, विफलताओं को संभालना, और ज़रूरत हो तो retry करना। यदि आपको Python के exception handling की बुनियाद दोहरानी हो, तो Exception & Error Handling in Python देखें। हम पूरे भाग में यह बेस सेटअप उपयोग करेंगे:

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()संचालित द्वारा

Semaphores के साथ rate limiting

पिछले भाग में, हमने एक साथ 10 रिक्वेस्ट चलाईं। वह ठीक चला। पर क्या हो जब आपको 500 स्टोरी फ़ेच करनी हों? या 10,000 पेज स्क्रैप करने हों?

ज़्यादातर APIs rate limits लागू करती हैं। हो सकता है वे प्रति सेकंड 10 रिक्वेस्ट या 100 concurrent कनेक्शन की अनुमति दें। सीमा लांघने पर आपको ब्लॉक, थ्रॉटल, या बैन किया जा सकता है। भले ही API सीमाएँ न लगाए, हज़ारों रिक्वेस्ट एक साथ भेजना आपके सिस्टम या सर्वर को ठप कर सकता है।

आपको नियंत्रित करना होगा कि किसी भी पल कितनी रिक्वेस्ट "in flight" हों। यही काम semaphore करता है।

Semaphore परमिट सिस्टम की तरह काम करता है। मान लें आपके पास तीन परमिट हैं। कोई भी कार्य रिक्वेस्ट भेजने से पहले परमिट लेता है। काम होने पर परमिट लौटा देता है ताकि अगला उपयोग कर सके। यदि कोई परमिट उपलब्ध नहीं, तो कार्य इंतज़ार करता है जब तक एक खाली न हो।

3 परमिट और 4+ कार्य होने पर यह ऐसे चलता है:

1. तीन परमिट उपलब्ध हैं।

2. कार्य A एक परमिट लेता है (2 शेष), अपनी रिक्वेस्ट शुरू करता है।

3. कार्य B एक परमिट लेता है (1 शेष), रिक्वेस्ट शुरू।

4. कार्य C एक परमिट लेता है (0 शेष), रिक्वेस्ट शुरू।

5. कार्य D परमिट चाहता है, पर उपलब्ध नहीं—इंतज़ार करता है।

6. कार्य A पूरा, परमिट लौटाता है (1 उपलब्ध)।

7. कार्य D वह परमिट लेकर रिक्वेस्ट शुरू करता है।

8. यह सिलसिला सभी कार्य पूरे होने तक चलता रहता है।

चरण 5 में जो इंतज़ार है, वह कुशल है। कार्य "परमिट खाली हुआ?" पूछते हुए घूमता नहीं रहता—वह निलंबित हो जाता है और event loop अन्य कोड चलाता है। परमिट उपलब्ध होते ही loop उसे जगा देता है।

अब कोड देखें। asyncio में आप asyncio.Semaphore(n) से semaphore बनाते हैं, जहाँ n परमिट की संख्या है। इसे उपयोग करने के लिए अपने कोड को async with semaphore: में लपेटें। ब्लॉक में घुसते ही परमिट मिल जाता है और बाहर निकलते ही अपने-आप रिलीज़ हो जाता है:

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:  # Acquire permit (or wait if none available)
        async with session.get(f"{HN_API}/item/{story_id}.json") as response:
            return await response.json()
    # Permit automatically released hereसंचालित द्वारा

अब 30 स्टोरी semaphore के साथ और बिना तुलना करें:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = (await response.json())[:30]

        # Without rate limiting: all 30 at once
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story(session, sid) for sid in story_ids])
        print(f"No limit: {time.perf_counter() - start:.2f}s (30 concurrent)")

        # With Semaphore(5): max 5 at a time
        semaphore = asyncio.Semaphore(5)
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story_limited(session, sid, semaphore) for sid in story_ids])
        print(f"Semaphore(5): {time.perf_counter() - start:.2f}s (5 concurrent)")

asyncio.run(main())संचालित द्वारा

No limit: 0.62s (30 concurrent)
Semaphore(5): 1.50s (5 concurrent)संचालित द्वारा

Semaphore वाला संस्करण धीमा है क्योंकि यह पाँच-पाँच के बैच में प्रोसेस करता है। यही समझौता है: गति के बदले नियंत्रित, सर्वर-अनुकूल व्यवहार।

ध्यान दें: semaphore समय-आधारित सीमा नहीं लगाता, बल्कि समानांतर रिक्वेस्ट की सीमा लगाता है। Semaphore(10) का मतलब है "एक समय में अधिकतम 10 रिक्वेस्ट", न कि "प्रति सेकंड 10 रिक्वेस्ट"। यदि आपको सख्त समय-आधारित rate limiting चाहिए (जैसे ठीक 10 प्रति सेकंड), तो semaphore को बैच देरी के साथ जोड़ें, या aiolimiter जैसी लाइब्रेरी उपयोग करें।

asyncio.wait_for() के साथ timeouts

भले ही concurrency नियंत्रित हो, कुछ रिक्वेस्ट अटक सकती हैं। सर्वर कनेक्शन स्वीकार कर ले, पर जवाब न दे। बिना timeout के, आपका प्रोग्राम अनिश्चितकाल तक इंतज़ार करेगा।

asyncio.wait_for() किसी भी coroutine पर समय-सीमा लपेट देता है। आप coroutine और सेकंड में timeout पास करते हैं। समय रहते ऑपरेशन पूरा न होने पर यह asyncio.TimeoutError उठाता है:

async def slow_operation():
    print("Starting slow operation...")
    await asyncio.sleep(5)
    return "Done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")

asyncio.run(main())संचालित द्वारा

Starting slow operation...
Operation timed out after 2 secondsसंचालित द्वारा

Timeout होने पर wait_for() coroutine को cancel कर देता है। आप TimeoutError पकड़कर तय कर सकते हैं—रिक्वेस्ट छोड़ें, डिफ़ॉल्ट वैल्यू लौटाएँ, या retry करें।

Concurrent रिक्वेस्ट के लिए, हर एक को अलग से लपेटें। यह हेल्पर error dictionary लौटाता है, exception उठाने के बजाय:

async def fetch_story_with_timeout(session, story_id, timeout=5.0):
    try:
        coro = fetch_story(session, story_id)
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": f"Story {story_id} timed out"}संचालित द्वारा

जब कोई coroutine cancel होती है (timeout या किसी और वजह से), Python उसके अंदर asyncio.CancelledError उठाता है। यदि आपका coroutine फ़ाइल हैंडल या कनेक्शन जैसे संसाधन पकड़े हो, तो सुनिश्चित सफ़ाई के लिए try/finally का उपयोग करें—even cancellation पर भी:

async def fetch_with_cleanup(session, url):
    print("Starting fetch...")
    try:
        async with session.get(url) as response:
            return await response.text()
    finally:
        print("Cleanup complete")  # Runs even on cancellationसंचालित द्वारा

asyncio.gather() के साथ error handling

Timeouts धीमी रिक्वेस्ट पकड़ते हैं। पर कुछ रिक्वेस्ट तुरंत error के साथ फ़ेल हो सकती हैं। देखें, बैच में एक के फ़ेल होने पर क्या होता है।

पहले, fetch_story() का ऐसा संस्करण चाहिए जो अमान्य IDs पर exception उठाए:

async def fetch_story_strict(session, story_id):
    story = await fetch_story(session, story_id)
    if story is None:
        raise ValueError(f"Story not found: {story_id}")
    return storyसंचालित द्वारा

अब चार वैध IDs और एक अमान्य ID फ़ेच करें:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]  # 4 valid + 1 invalid

        try:
            stories = await asyncio.gather(
                *[fetch_story_strict(session, sid) for sid in ids_to_fetch]
            )
            print(f"Got {len(stories)} stories")
        except ValueError as e:
            print(f"ERROR: {e}")

asyncio.run(main())संचालित द्वारा

ERROR: Story not found: 99999999999संचालित द्वारा

एक अमान्य ID के साथ, हम चार सफल नतीजे भी खो देते हैं। डिफ़ॉल्ट रूप से, gather() fail-fast होता है: एक exception सब कुछ cancel कर देता है और ऊपर तक चला जाता है।

आंशिक नतीजे रखना हो तो return_exceptions=True जोड़ें। इससे gather() का व्यवहार बदलता है: exception उठाने की जगह, वे results सूची में items के रूप में लौटते हैं, सफल मानों के साथ-साथ:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]

        results = await asyncio.gather(
            *[fetch_story_strict(session, sid) for sid in ids_to_fetch],
            return_exceptions=True  # Don't raise, return exceptions in list
        )

        # Separate successes from failures using isinstance()
        stories = [r for r in results if not isinstance(r, Exception)]
        errors = [r for r in results if isinstance(r, Exception)]

        print(f"Got {len(stories)} stories, {len(errors)} failed")

asyncio.run(main())संचालित द्वारा

Got 4 stories, 1 failedसंचालित द्वारा

isinstance(result, Exception) जाँच से आप सफल नतीजों को errors से अलग कर सकते हैं। फिर सफल को प्रोसेस करें और विफलताओं को लॉग या retry करें।

Exponential backoff के साथ retry लॉजिक

कुछ विफलताएँ अस्थायी होती हैं। सर्वर कुछ देर के लिए ओवरलोड हो सकता है, या नेटवर्क में रुकावट आ सकती है। ऐसे में retry करना समझदारी है।

पर तुरंत retry करना स्थिति बिगाड़ सकता है। यदि सर्वर जूझ रहा है, तो एक साथ की गई retries समस्या बढ़ाती हैं। Exponential backoff इसका हल है—हर प्रयास के बीच प्रतीक्षा समय बढ़ता जाता है।

पैटर्न 2 ** attempt से इंतज़ार तय करता है: प्रयास 0 पर 1 सेकंड (2⁰), प्रयास 1 पर 2 सेकंड (2¹), प्रयास 2 पर 4 सेकंड (2²), आदि। इससे सर्वर को उबरने का बढ़ता समय मिलता है:

async def fetch_with_retry(session, story_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            story = await fetch_story(session, story_id)
            if story is None:
                raise ValueError(f"Story {story_id} not found")
            return story
        except (aiohttp.ClientError, ValueError):  # Catch specific exceptions
            if attempt == max_retries - 1:
                print(f"Story {story_id}: Failed after {max_retries} attempts")
                return None

            backoff = 2 ** attempt  # 1s, 2s, 4s...
            print(f"Story {story_id}: Attempt {attempt + 1} failed, retrying in {backoff}s...")
            await asyncio.sleep(backoff)संचालित द्वारा

ध्यान दें हम विशेष exceptions (aiohttp.ClientError, ValueError) पकड़ रहे हैं, न कि बिना शर्त except। इससे केवल अस्थायी errors पर ही retry होती है। कोड की गलती से आई KeyError पर retry नहीं होनी चाहिए।

अब वैध और अमान्य IDs के मिश्रण पर टेस्ट करें:

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        test_ids = [story_ids[0], 99999999999, story_ids[1], 88888888888, story_ids[2]]

        results = await asyncio.gather(*[fetch_with_retry(session, sid) for sid in test_ids])

        successful = [r for r in results if r is not None]
        print(f"\nSuccessful: {len(successful)}, Failed: {len(test_ids) - len(successful)}")

asyncio.run(main())संचालित द्वारा

Story 99999999999: Attempt 1 failed, retrying in 1s...
Story 88888888888: Attempt 1 failed, retrying in 1s...
Story 99999999999: Attempt 2 failed, retrying in 2s...
Story 88888888888: Attempt 2 failed, retrying in 2s...
Story 99999999999: Failed after 3 attempts
Story 88888888888: Failed after 3 attempts

Successful: 3, Failed: 2संचालित द्वारा

प्रोडक्शन में, आप jitter (छोटी यादृच्छिक देरी) भी जोड़ेंगे ताकि कई विफल रिक्वेस्ट एक ही पल में retry न करें। साथ ही, केवल अस्थायी errors (जैसे 503) पर retry करें, स्थायी (जैसे 404 या 401) पर तुरंत हार मानें।

aiosqlite के साथ Python Async डेटाबेस स्टोरेज

हमने Hacker News स्टोरीज़ को सही rate limiting, timeouts, और error handling के साथ फ़ेच किया है। अब उन्हें डेटाबेस में स्टोर करें।

साधारण synchronous डेटाबेस लाइब्रेरी जैसे sqlite3 event loop को क्वेरी के दौरान ब्लॉक कर देंगी—async का उद्देश्य निष्फल। जब आपका कोड डेटाबेस का इंतज़ार करता है, कोई अन्य coroutine नहीं चल पाएगा। Async एप्लिकेशन के लिए async डेटाबेस लाइब्रेरी चाहिए।

aiosqlite Python की बिल्ट-इन sqlite3 को async इंटरफ़ेस में लपेटती है। यह डेटाबेस ऑपरेशन को थ्रेड पूल में चलाती है ताकि event loop ब्लॉक न हो। SQLite को किसी सर्वर सेटअप की ज़रूरत नहीं—यह सिर्फ़ एक फ़ाइल है—इसलिए आप यह कोड तुरंत चला सकते हैं। यदि आप Python में डेटाबेस पर नए हैं, तो Introduction to Databases in Python कोर्स synchronous नींव कवर करता है, जिस पर aiosqlite आधारित है।

डेटाबेस सेट करना

पैटर्न परिचित लगेगा। ठीक aiohttp.ClientSession की तरह, आप कनेक्शन को मैनेज करने के लिए async with का उपयोग करते हैं:

import aiosqlite

async def init_db(db_path):
    async with aiosqlite.connect(db_path) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.commit()

asyncio.run(init_db("stories.db"))संचालित द्वारा

मुख्य फ़ंक्शन:

• aiosqlite.connect(path) डेटाबेस फ़ाइल खोलता (या बनाता) है।

• await db.execute(sql) एक SQL स्टेटमेंट चलाता है।

• await db.commit() परिवर्तन डिस्क पर सहेजता है।

स्टोरी सहेजना

एक स्टोरी सेव करने का फ़ंक्शन:

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )संचालित द्वारा

? placeholders SQL injection से बचाते हैं—मान सीधे डालने के लिए कभी f-strings का उपयोग न करें। INSERT OR REPLACE दोबारा फ़ेच होने पर मौजूद स्टोरी को अपडेट कर देता है।

Python में पूर्ण Async पाइपलाइन: फ़ेच और स्टोर

अब इस ट्यूटोरियल की सारी बातें जोड़ें। हम 20 Hacker News स्टोरीज़ rate limiting के साथ फ़ेच करेंगे और उन्हें डेटाबेस में स्टोर करेंगे:

import aiohttp
import aiosqlite
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:
        story = await fetch_story(session, story_id)
        if story:
            return story
        return None

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )

async def main():
    # Initialize database
    async with aiosqlite.connect("hn_stories.db") as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Fetch stories
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{HN_API}/topstories.json") as response:
                story_ids = await response.json()
            
            semaphore = asyncio.Semaphore(5)
            tasks = [fetch_story_limited(session, sid, semaphore) for sid in story_ids[:20]]
            stories = await asyncio.gather(*tasks)
        
        # Save to database
        for story in stories:
            if story:
                await save_story(db, story)
        await db.commit()
        
        # Query and display
        cursor = await db.execute("SELECT id, title, score FROM stories ORDER BY score DESC LIMIT 5")
        rows = await cursor.fetchall()
        
        print(f"Saved {len([s for s in stories if s])} stories. Top 5 by score:")
        for row in rows:
            print(f"  [{row[2]}] {row[1][:50]}")

asyncio.run(main())संचालित द्वारा

Saved 20 stories. Top 5 by score:
  [671] Google Antigravity exfiltrates data via indirect p
  [453] Trillions spent and big software projects are stil
  [319] Ilya Sutskever: We're moving from the age of scali
  [311] Show HN: We built an open source, zero webhooks pa
  [306] FLUX.2: Frontier Visual Intelligenceसंचालित द्वारा

यह पाइपलाइन हर भाग के पैटर्न अपनाती है: कनेक्शन पूलिंग के लिए ClientSession, rate limiting के लिए Semaphore(5), concurrent फ़ेचिंग के लिए gather(), और अब async स्टोरेज के लिए aiosqlite। हर घटक अपना हिस्सा बिना दूसरों को ब्लॉक किए निभाता है।

जब भी आप यह वर्कफ़्लो चलाएँगे, आपको दिन की शीर्ष स्टोरीज़ मिलेंगी।

निष्कर्ष

इस ट्यूटोरियल ने आपको बुनियादी async/await सिंटैक्स से लेकर एक संपूर्ण डेटा पाइपलाइन तक पहुँचाया। आपने सीखा कि coroutines कैसे ठहरती और फिर शुरू होती हैं, event loop concurrent कार्यों को कैसे मैनेज करता है, और asyncio.gather() कैसे एक साथ कई ऑपरेशन चलाता है। आपने aiohttp के साथ वास्तविक HTTP रिक्वेस्ट जोड़ीं, semaphores से concurrency नियंत्रित की, timeouts और retries से विफलताएँ संभालीं, और aiosqlite के साथ परिणाम डेटाबेस में सहेजे।

Async का उपयोग तब करें जब आपका कोड बाहरी सिस्टम का इंतज़ार करता है: HTTP APIs, डेटाबेस, फ़ाइल I/O, या नेटवर्क सॉकेट्स। CPU-भारी कार्यों जैसे डेटा प्रोसेसिंग या गणना के लिए async मदद नहीं करेगी—इसके बजाय multiprocessing या concurrent.futures देखें। और गहराई में जाने के लिए asyncio दस्तावेज़ और async वेब APIs बनाने के लिए FastAPI पर विचार करें। 

यदि आप इसी ज्ञान पर आगे बढ़ते हुए बुद्धिमान एप्लिकेशन डिज़ाइन करना सीखना चाहते हैं, तो Associate AI Engineer for Developers करियर ट्रैक ज़रूर देखें।