Python asyncio Advanced Guide | Asynchronous Programming | සිංහල | Python Async

ආයුබෝවන්, developer යාළුවනේ! Python වල Asynchronous Programming වලට ඔයාලා කැමති නම්, මේ ලිපිය ඔයාලාට අනිවාර්යයෙන්ම වැදගත් වේවි.

අද අපි කතා කරන්නේ Python වල තියෙන සුපිරිම feature එකක් ගැන – ඒ තමයි Asynchronous Programming, විශේෂයෙන්ම asyncio library එක. සාමාන්‍යයෙන් Python code එකක් run වෙන්නේ එක thread එකක් මත (single-threaded). ඒත් network requests, file reading/writing වගේ I/O-bound operations තියෙන වෙලාවට මේකෙන් program එකේ performance එක ගොඩක් අඩු වෙන්න පුළුවන්.

උදාහරණයක් විදියට, ඔයාට website කීපයකින් එකවර data ටිකක් ගන්න ඕන නම්, එකින් එක request යවද්දී ගොඩක් වෙලා යනවා නේද? asyncio එන්නේ මේකට විසඳුමක් විදියට. එකවර tasks කිහිපයක් manage කරමින්, I/O operations වලදී program එක block නොවී, task එක complete වෙනකල් අනිත් tasks වලට වැඩ කරන්න ඉඩ දෙනවා. මේකෙන් ඔයාගේ application එකේ speed එක සහ responsiveness එක සැලකිය යුතු මට්ටමකින් වැඩි වෙනවා.

මේ guide එකෙන් අපි asyncio වල මූලිකාංග, Tasks, Futures වගේ දේවල් ගැන විතරක් නෙවෙයි, asyncio.gather() වගේ powerful functions පාවිච්චි කරලා multiple URL fetch කරන විදිහත්, aiohttp වගේ external libraries කොහොමද මේකට එකතු කරගන්නේ කියලත් බලමු. ඒ වගේම, asynchronous code වලදී errors handle කරන විදිහ සහ best practices ගැනත් අපි විස්තරාත්මකව කතා කරනවා.

එහෙනම්, අපි පටන් ගමු!

1. Asynchronous Programming වල මූලිකාංග: Tasks සහ Futures

asyncio කියන්නේ Python වල asynchronous code ලියන්න, concurrent I/O operations කරන්න සහ distributed applications ගොඩනගන්න පාවිච්චි කරන library එකක්. මේකේ core concepts තමයි async සහ await keywords.

• async: මේ keyword එක පාවිච්චි කරන්නේ coroutine එකක් (asynchronous function) define කරන්න. සාමාන්‍ය function එකක් වගේම තමයි, නමුත් මේක await කරන්න පුළුවන්.
• await: මේ keyword එක පාවිච්චි කරන්නේ coroutine එකක් complete වෙනකල් program එක pause කරන්න. Program එක block වෙන්නේ නෑ, event loop එකට පුළුවන් මේ වෙලාවේ අනිත් tasks handle කරන්න.

Tasks

Task එකක් කියන්නේ Coroutine එකක් asyncio event loop එකේ run කරන්න schedule කරපු object එකක්. ඔයා asyncio.create_task() call කරලා coroutine එකක් Task එකක් බවට පත් කරනවා. මේ Task එක event loop එකට එකතු වුනාට පස්සේ, loop එකට පුළුවන් ඒක execute කරන්න. එකම event loop එක ඇතුලේ tasks ගොඩක් එකවර (concurrently) run කරන්න පුළුවන්.

Futures

Future එකක් කියන්නේ asynchronous operation එකක result එක (result of an asynchronous operation) හෝ exception එක (exception) represent කරන object එකක්. තවම result එක ලැබිලා නැති operation එකක්. Task එකක් කියන්නේ Future එකක subclass එකක් වගේ. Task එකකින් coroutine එකක් wrap කරලා ඒකේ result එක Future එකක result එකක් විදියට handle කරනවා.

සරල උදාහරණයක් බලමු:

import asyncio

async def greet(name):
    print(f"Hello, {name}!")
    await asyncio.sleep(1) # Simulate some I/O operation that takes 1 second
    print(f"Goodbye, {name}!")

async def main_tasks_futures():
    print("Starting main_tasks_futures...")
    # Creating Tasks from coroutines
    task1 = asyncio.create_task(greet("Alice"))
    task2 = asyncio.create_task(greet("Bob"))

    # Awaiting tasks to complete
    # The program will pause here until task1 and task2 are done.
    # However, since they are independent tasks, they will run concurrently.
    await task1
    await task2
    print("All tasks completed.")

if __name__ == "__main__":
    # Run the main asynchronous function
    asyncio.run(main_tasks_futures())

මේ code එක run කරද්දී, greet("Alice") සහ greet("Bob") කියන coroutines දෙකම එකවර වගේ run වෙනවා. await asyncio.sleep(1) call එකෙන් thread එක block වෙන්නේ නැහැ. ඒ වෙනුවට event loop එකට පුළුවන් අනිත් task එක run කරන්න. ඒ නිසා ඔයාට output එකේ Hello, Alice! සහ Hello, Bob! එක පාරටම වගේ පෙනෙයි, ඊට පස්සේ තත්පරයකට පස්සේ Goodbye, Alice! සහ Goodbye, Bob! පෙනෙයි.

2. Concurrent I/O: asyncio.gather() සහ aiohttp

ඇත්තටම බලපන්, අපිට එකපාර web request ගොඩක් යවන්න වෙන වෙලාවල් තියෙනවා නේද? උදාහරණයක් විදියට, different APIs වලින් data ගන්න, නැත්නම් web scraping කරද්දී website ගොඩකින් එකවර data ගන්න. සාමාන්‍ය synchronous code වලින් මේක කරන්න ගියොත්, එක request එකක් complete වෙනකල් අනිත් request එක යවන්න බෑ. මේකෙන් ගොඩක් වෙලා යනවා.

asyncio.gather() කියන්නේ එකවර coroutines ගොඩක් parallel ව run කරන්න පුළුවන් powerful tool එකක්. මේකෙන් ඔයාට tasks ගොඩක් එකතු කරලා එකවර execute කරන්න පුළුවන්. gather() call එක complete වෙන්නේ ඒක ඇතුලේ තියෙන හැම coroutine එකක්ම complete වුනාට පස්සේ.

aiohttp Library එක

Python වල requests library එක synchronous HTTP requests වලට සුපිරි වුනත්, asynchronous requests වලට නම් අපිට aiohttp වගේ library එකක් ඕන වෙනවා. aiohttp කියන්නේ asyncio මත ගොඩනගපු asynchronous HTTP client/server library එකක්. මේකෙන් අපිට non-blocking HTTP requests යවන්න පුළුවන්.

aiohttp install කරගන්න මේ command එක run කරන්න:

pip install aiohttp

දැන් අපි බලමු asyncio.gather() සහ aiohttp පාවිච්චි කරලා multiple URLs කොහොමද concurrently fetch කරන්නේ කියලා:

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """Fetches content from a given URL asynchronously."""
    try:
        async with session.get(url) as response:
            response.raise_for_status() # Raise an exception for bad status codes
            print(f"Successfully fetched: {url}")
            return await response.text()
    except aiohttp.ClientError as e:
        print(f"Error fetching {url}: {e}")
        return None # Return None or re-raise, depending on handling needs

async def main_concurrent_fetching():
    urls = [
        "http://example.com",
        "http://www.google.com",
        "http://www.bing.com",
        "http://www.yahoo.com",
        "https://nonexistent-domain-12345.com" # An URL that will likely fail
    ]
    
    print("Starting concurrent URL fetching...")
    start_time = time.time()

    async with aiohttp.ClientSession() as session:
        # Create a list of tasks, one for each URL
        tasks = []
        for url in urls:
            tasks.append(fetch_url(session, url))
        
        # Run all tasks concurrently and wait for all to complete
        # return_exceptions=True means that if a task fails, its exception is returned
        # as a result instead of stopping the gather() call.
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        print("\n--- Processing Responses ---")
        for i, result in enumerate(responses):
            url = urls[i]
            if isinstance(result, Exception):
                print(f"Failed to fetch {url}. Error: {result}")
            elif result is None:
                print(f"Skipped {url} due to prior error.")
            else:
                print(f"URL: {url}, Content Length: {len(result)} characters")

    end_time = time.time()
    print(f"\nConcurrent fetching completed in {end_time - start_time:.2f} seconds.")

if __name__ == "__main__":
    asyncio.run(main_concurrent_fetching())

මේ code එක run කරද්දී, fetch_url coroutines 5ම එකවර initiate වෙනවා. asyncio.gather() එකෙන් මේ හැම request එකක්ම complete වෙනකල් බලාගෙන ඉන්නවා, නමුත් එක request එකක් block වුනොත් අනිත් ඒවාට දිගටම run වෙන්න පුළුවන්. මේකෙන් overall time එක ගොඩක් අඩු වෙනවා. Non-existent domain එකෙන් එන error එකත් return_exceptions=True නිසා අනිත් සාර්ථක responses එක්කම අපිට ලැබෙනවා. අපි මේක ගැන ඊලඟට කතා කරමු.

3. Error Handling සහ Best Practices

asynchronous code වලදී errors handle කරන එකත් ටිකක් වෙනස් විදියට වෙන්නේ. සාමාන්‍ය synchronous code වල try...except block එකක් පාවිච්චි කරනවා වගේම, asynchronous code වලත් try...except පාවිච්චි කරන්න පුළුවන්. ඒත් asyncio.gather() වගේ functions පාවිච්චි කරද්දී, exceptions behave කරන විදිහ ගැන දැනුවත් වෙන්න ඕන.

Propagating Exceptions in asyncio.gather()

සාමාන්‍යයෙන්, asyncio.gather() එකක් ඇතුලේ task එකක් fail වුනොත්, ඒ exception එක gather() call එකට propagate වෙනවා. ඒ කියන්නේ අනිත් tasks continue වුනත්, gather() call එක එතනින්ම fail වෙනවා. මේකෙන් අනිත් tasks වල results අපිට ගන්න බැරි වෙන්න පුළුවන්.

ඒත් return_exceptions=True parameter එක asyncio.gather() එකට යොදාගත්තොත්, fail වෙන tasks වල exceptions ඒවායේ results විදියටම return වෙනවා, සාර්ථක results එක්කම. මේකෙන් අපිට හැම task එකකම status එක separately handle කරන්න පුළුවන්.

මේ උදාහරණය බලන්න:

import asyncio

async def risky_task(name, should_fail=False):
    print(f"Task {name}: Starting...")
    await asyncio.sleep(0.5) # Simulate some work
    if should_fail:
        print(f"Task {name}: FAILED!")
        raise ValueError(f"{name} failed intentionally!")
    print(f"Task {name}: Completed successfully.")
    return f"{name} completed successfully."

async def main_error_handling_gather():
    # Example 1: Without return_exceptions (will raise the first error)
    print("\n--- Example 1: asyncio.gather() WITHOUT return_exceptions --- ")
    try:
        results = await asyncio.gather(
            risky_task("Task A"),
            risky_task("Task B", True), # This will fail
            risky_task("Task C")
        )
        print("Results (should not see this line if an error occurred):", results)
    except ValueError as e:
        print(f"Caught expected error in gather: {e}")
    
    # Example 2: With return_exceptions=True
    print("\n--- Example 2: asyncio.gather() WITH return_exceptions=True ---")
    results = await asyncio.gather(
        risky_task("Task X"),
        risky_task("Task Y", True), # This will fail, but result will be an exception object
        risky_task("Task Z"),
        return_exceptions=True # This is key for individual error handling
    )
    print("Results with exceptions returned:", results)
    print("Processing individual results:")
    for res in results:
        if isinstance(res, Exception):
            print(f"  ERROR: {type(res).__name__} - {res}")
        else:
            print(f"  SUCCESS: {res}")

if __name__ == "__main__":
    asyncio.run(main_error_handling_gather())

පළමු උදාහරණයේදී Task B fail වෙන නිසා, asyncio.gather() call එකෙන් අනිත් results ලබා නොදී ValueError එකක් raise කරනවා. දෙවෙනි උදාහරණයේදී, return_exceptions=True නිසා, Task Y එකේ exception එක results list එකේ object එකක් විදියටම ලැබෙනවා. මේකෙන් අපිට හැම task එකකම තත්වය separately check කරලා handle කරන්න පුළුවන්.

Best Practices for Asynchronous Programming

asynchronous code ලියද්දී මතක තියාගන්න ඕන වැදගත් කරුණු කීපයක් තියෙනවා:

1. Avoid Blocking Calls:ඔයාගේ asynchronous functions ඇතුලේ synchronous blocking I/O calls (file operations, database queries) පාවිච්චි කරන එකෙන් වළකින්න. මේවා event loop එක block කරනවා. ඒ වගේ calls කරන්න ඕන නම්, loop.run_in_executor() පාවිච්චි කරන්න. මේකෙන් blocking calls වෙන thread එකක run වෙලා main event loop එක block වීම වළක්වනවා.
2. Keep Coroutines Small and Focused:සෑම coroutine එකක්ම එක වැඩක් විතරක් කරන විදියට (single responsibility) ලියන්න. මේකෙන් code එක maintain කරන්න සහ test කරන්න පහසු වෙනවා.
3. Structured Concurrency (Python 3.11+):Python 3.11 සහ ඊට ඉහල versions වල asyncio.TaskGroup වගේ දේවල් එනවා. මේවා පාවිච්චි කිරීමෙන් concurrent tasks වඩාත් structured විදියට manage කරන්න පුළුවන්, error handling ද පහසු වෙනවා.

Handle Timeouts:Network requests වගේ දේවල් හැම වෙලාවෙම successful වෙන්නේ නෑ. සමහර වෙලාවට server එක respond නොකර ඉන්න පුළුවන්. ඒ වගේ වෙලාවට අනවශ්‍ය විදියට block වෙන්නේ නැතුව asyncio.wait_for() හෝ aiohttp request එකේ timeout parameter එක පාවිච්චි කරන්න.

async def fetch_with_timeout(session, url, timeout=5):
    try:
        async with asyncio.timeout(timeout):
            async with session.get(url) as response:
                return await response.text()
    except asyncio.TimeoutError:
        print(f"Timeout occurred for {url}")
        return None

Use async with for Resource Management:aiohttp.ClientSession() වගේ resources පාවිච්චි කරද්දී, ඒවා නිවැරදිව close කරන එක ගොඩක් වැදගත්. async with statement එක පාවිච්චි කිරීමෙන් resource එක automatic ව handle වෙනවා, exception එකක් ආවත්.

async with aiohttp.ClientSession() as session:
    # Use the session here
    pass

අවසන් වශයෙන්

ඉතින් යාළුවනේ, මේ guide එකෙන් අපි Python asyncio වල advanced features ගැන ගොඩක් දේවල් ඉගෙන ගත්තා. Tasks සහ Futures කියන්නේ මොනවද කියලත්, asyncio.gather() සහ aiohttp වගේ libraries පාවිච්චි කරලා කොහොමද concurrent I/O operations කරන්නේ කියලත් අපි බැලුවා. ඒ වගේම, asynchronous code වලදී errors handle කරන විදිහ සහ ඔයාගේ code එක performant විදියට maintain කරන්න පුළුවන් best practices ගැනත් අපි කතා කළා.

asyncio කියන්නේ මුලින් ටිකක් අමාරු වුනත්, I/O-bound applications වල performance වැඩි කරන්න තියෙන සුපිරිම tool එකක්. මේ concepts හොඳින් තේරුම් අරන්, ඔයාගේ projects වලට මේවා integrate කරගන්න උත්සාහ කරන්න. මේක ඔයාගේ applications වල responsive-ness සහ efficiency එක අනිවාර්යයෙන්ම වැඩි කරයි.

මේ ලිපිය ඔයාට වැදගත් වුනා කියලා හිතනවා. ඔයාගේ අත්දැකීම්, ප්‍රශ්න, නැත්නම් මේ ගැන තියෙන අදහස් comment section එකේ බෙදාගන්නත් අමතක කරන්න එපා. මේ concepts ඔයාගේ ඊළඟ project එකේදී implement කරලා බලන්න!

ජයවේවා!