📘 Chapter 2: *Asyncio Basics* 🚀

Chapter 2 delves into the foundational aspects of asyncio in Python, focusing on how it enables single-threaded concurrency using coroutines, tasks, and event loops. Here’s a breakdown:


🌀 2.1 Introducing Coroutines
- Coroutines are special Python functions that can pause and resume execution when encountering a potentially long-running task.
- When a coroutine pauses to wait for an operation, other tasks can run concurrently, providing concurrency. 💡
- `async` and `await` are the two essential keywords:
- async defines a function as a coroutine.
- await pauses the coroutine until a result is available from an asynchronous operation.

⚙️ Example of Creating a Coroutine

async def my_coroutine() -> None:
    print("Hello world!")

This is similar to a normal Python function but can pause its execution.


⏳ 2.2 Introducing Long-Running Coroutines with `sleep`
- Asyncio’s sleep function allows us to pause execution, simulating real-world, long-running operations like web requests or database queries 🌐.
- When await asyncio.sleep() is called, other tasks can be executed during the pause.

⚙️ Example of Using asyncio.sleep

async def hello_world_message() -> str:
    await asyncio.sleep(1)
    return "Hello World!"

This coroutine pauses for 1 second before returning "Hello World!". During that second, other coroutines can run concurrently.


🔄 2.3 Running Concurrently with Tasks
- A task is a wrapper around a coroutine that schedules it to run on the event loop 🕑.
- Tasks allow coroutines to be run concurrently, as they don’t block the event loop, unlike await, which pauses until a result is returned.

⚙️ Example of Creating a Task

import asyncio
from util import delay

async def main():
    task = asyncio.create_task(delay(3))
    await task

Here, the task delay(3) runs concurrently, while other code can execute.


⛔ 2.4 Canceling Tasks and Setting Timeouts
- Tasks can be canceled using task.cancel(), raising a CancelledError within the task. If a task is taking too long, we can also set timeouts using asyncio.wait_for 🕒.

⚙️ Example of Cancelling a Task

async def cancel_task(task):
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

💼 2.5 Tasks, Coroutines, Futures, and Awaitables
- Futures represent a value that will be available in the future but might not exist yet. They are used internally in asyncio and can be awaited 🎯.
- Coroutines and tasks can both be used in await expressions.

⚙️ Example of Working with Futures

from asyncio import Future

my_future = Future()
my_future.set_result(42)
print(my_future.result())  # Outputs: 42

⏱️ 2.6 Measuring Coroutine Execution Time with Decorators
- By using decorators, we can measure the execution time of coroutines for performance analysis. ⌛


⚠️ 2.7 Pitfalls of Coroutines and Tasks
- Be cautious with CPU-bound code inside coroutines, as it will block the event loop.
- Avoid using blocking I/O APIs; use asyncio-compatible libraries to ensure the event loop runs smoothly 🛠️.


🔧 2.8 Accessing and Manually Managing the Event Loop
- You can access the event loop directly using asyncio.get_event_loop(), though asyncio.run() is recommended for running main coroutines.


🛠️ 2.9 Using Debug Mode
- Debug mode helps in identifying long-running coroutines or tasks that block the event loop 🧐. You can enable it with:

python3 -X dev program.py

Or by setting the PYTHONASYNCIODEBUG environment variable.

⚙️ Example of Running in Debug Mode

asyncio.run(main(), debug=True)

#PythonConcurrencyWithAsyncio
#Chapter_02
#Notes #Book

سلام خدمت دوستان

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

برای همین یه کلاینت کوچیک نوشتم واسه استفاده از API های پنل مرزبان داخل پایتون

آدرس پروژه:
https://github.com/Cmatrix1/MarzbanAPIClient/


پنل مرزبان چیه؟
مرزبان یک نرم افزار (وب اپلیکیشن) مدیریت پروکسیه که امکان مدیریت چند صد حساب پروکسی رو به شما میده. مرزبان از Xray-core قدرت گرفته و درواقع با پایتون هم توسعه داده شده
بنظرم حتما یه سری به گیت هاب مرزبان هم بزنید پروژه فوقالعاده ای هست


@Pythonic_Dev

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

https://vrgl.ir/tM8oi

اگر خوشتون اومد و استقبال کردید، می‌تونم یک ویدیو هم ضبط کنم که با همین روش، یک وبسایت خفن رو از صفر تا صد بالا بیاریم! 🚀
ممنون می‌شم نظرات و پیشنهاداتتون رو با من به اشتراک بذارید.

💡 کاپلینگ (Coupling) در مهندسی نرم‌افزار

یکی از مفاهیم مهم در طراحی نرم‌افزار که خیلی وقت‌ها به چشم نمیاد ولی تأثیر عمیقی روی کیفیت، توسعه‌پذیری و نگهداری کد داره، کاپلینگ یا همون وابستگی بین اجزای سیستم هست.

هرچقدر این وابستگی بین بخش‌های مختلف کد کمتر باشه، اون سیستم قابل اعتمادتر، تست‌پذیرتر و توسعه‌پذیرتر میشه.

> 🎯 هدف کلی اینه که تا جای ممکن، کاپلینگ رو کاهش بدیم تا اجزای سیستم بتونن جدا از هم تکامل پیدا کنن.


🔧 کاپلینگ یعنی چی؟

به زبان ساده، یعنی اینکه یک ماژول یا کلاس چقدر به بقیه‌ی ماژول‌ها وابسته‌ست.

- کاپلینگ زیاد → تغییر در یک قسمت باعث خرابی قسمت‌های دیگه میشه
- کاپلینگ کم→ ماژول‌ها مستقل‌ترن و راحت‌تر میشه تغییرشون داد

📚 انواع کاپلینگ با مثال:

❌ 1. Content Coupling (بدترین نوع)
وقتی یه ماژول مستقیم به جزئیات داخلی ماژول دیگه دسترسی داره:

class User:
    def __init__(self):
        self._password = "secret"

class AuthService:
    def check_password(self, user):
        return user._password == "secret"  # ❌ دسترسی مستقیم

✅ بهتر:

class User:

    def check_password(self, pwd):
        return self._password == pwd

2. Common / Global Coupling وابستگی چند ماژول به یه متغیر global:

# settings.py
shared_config = {"theme": "dark"}

✅ راه بهتر: استفاده از پارامترها یا Dependency Injection

3. Control Coupling
وقتی یک تابع رفتار تابع دیگه رو با فلگ (flag) کنترل می‌کنه:

def process(user, is_admin):
    if is_admin:
        print("Admin user")
    else:
        print("Normal user")

✅ راه بهتر:

class User:
    def process(self):
        print("Normal user")

class AdminUser(User):
    def process(self):

        print("Admin user")

4. Stamp Coupling
تابعی یک شیء کامل رو دریافت می‌کنه ولی فقط بخشی از اون رو استفاده می‌کنه:

def send_email(user):
    print(f"Sending email to {user.email}")

✅ بهتر:

def send_email(email):
    print(f"Sending email to {email}")

یا استفاده از Protocol / Interface

5. External Coupling
وابستگی مستقیم به فرمت داده یا API خارجی:

    username, age = data.split(",")

✅ بهتر: استفاده از abstraction یا Adapter

✅ 6. Data Coupling (بهترین نوع) ماژول فقط داده‌های مورد نیاز رو می‌گیره:

def calculate_discount(price, percent):
    return price * (1 - percent)

سادگی، تست‌پذیری و استقلال 👌

🎯 جمع‌بندی نهایی
یکی از اشتباهات رایج اینه که دنبال "بهترین راه ممکن" بگردیم.

اما واقعیت اینه که:
هیچ راه مطلقی وجود نداره. هر پروژه نیازهای خودشو داره و ممکنه یه نوع کاپلینگ توی یه شرایط خاص، بهترین انتخاب باشه.


اطلاعات بیشتر:

https://www.youtube.com/watch?v=MM9VQp-k0JQ

IRAN

این ربات هم کاربردی هست
برای تبدیل فایل های تلگرام و لینک یوتیوب به لینک داخلی

از سایت های دیگه هم پشتیبانی میکنه


@MeliFileToLinkBot

تو این پست یه توضیح کوتاهی میخوام راجع به Semaphore توی asyncio بدم.
بگم که Semaphore یک synchronization primitive هست.
سوال: synchronization primitive چیه؟

توی برنامه‌های concurrent چند execution flow داریم:
Thread
Process
Coroutine / Task


وقتی چند execution flow به منابع مشترک دسترسی داشته باشن:
فایل
Database
API
Shared state
Socket
Cache

نیاز داریم دسترسیشون رو مدیریت و محدود کنیم درغیراین صورت مشکلاتی به وجود میاد مثل race condition و فشار زیاد روی منابع

چند نمونه معروف synchronization primitive :
• Lock → فقط یک نفر همزمان وارد بشه
• Semaphore → اجازه بده N نفر همزمان وارد بشن
• Event → به بقیه خبر بده یه اتفاق افتاده
• Condition → ترکیب Lock و Event


یه مدل ذهنی خوب براش اینه که بهش مثل پارکینگ نگاه کنید:
فرض کنید پارکینگ ۳ جا داره؛ تا وقتی جا خالی هست ماشین وارد میشه، ولی ماشین چهارم باید منتظر بمونه تا یکی خارج بشه.
استفاده ساده‌ش هم به این شکل هست:

import asyncio

sem = asyncio.Semaphore(3)

async def my_coroutine(sem):
    async with sem:
        print("Acquired")

        await asyncio.sleep(1)

        print("Released")


async def main():
    tasks = [
        asyncio.create_task(
            my_coroutine(sem)
        )
        for _ in range(10)
    ]

    await asyncio.gather(*tasks)


asyncio.run(main())

توی این کد وقتی اجرا میشه فقط سه تا task همزمان اجرا میشن و بقیه taskها صبر میکنن تا یکی از taskها تموم بشه و یه slot خالی بشه.
یعنی اول یک slot میگیره و وقتی کار تموم شد slot رو آزاد میکنه.
یه سینتکس دیگه هم داره که به صورت دستی میتونید انجام بدید، شبیه بقیه context manager ها ولی خب آدم سالم تا وقتی context manager هست دستی انجام نمیده 😄

await sem.acquire()

try:
    # critical section

finally:
    sem.release()

بنظرم خوبه که یک صحبتی راجب به CI, CD بکنیم
خیلی ها کلا سمت اینجور مسایل نمیرن چون با خودشون میگن ما برنامه نویسیم و اصلا نباید سمت چیزای DevOps بریم و اصلا وظیفه ما نیست

اما بنظرم داشتن دانش نسبی توی این زمینه خیلی میتونه مفید باشه به خصوص اگر پروژه شخصی دارید که روی سرور ران کردید یا کلا پروژه هاتون رو خودتون دیپلوی میکنید.

حالا اصلا CI و CD چی هستن؟

عبارت CI مخفف Continuous Integration هست. به زبان ساده یعنی هر بار که کد جدیدی به پروژه اضافه میشه، یک سری چک به صورت خودکار انجام بشه تا مطمئن بشیم چیزی خراب نشده. مثلاً به محض اینکه روی GitHub پوش میکنید، تست‌ها اجرا بشن، لینتر اجرا بشه، بررسی امنیتی انجام بشه و اگر مشکلی وجود داشت همون لحظه مشخص بشه. اینطوری به جای اینکه دو هفته بعد وسط Production بفهمید یک نفر یه چیزی رو خراب کرده، همون موقع متوجه میشید و فیکسش میکنید.

و CD هم مخفف Continuous Delivery یا Continuous Deployment هست که مرحله بعد از CI محسوب میشه. یعنی وقتی تمام چک‌ها با موفقیت انجام شدن، پروژه به صورت خودکار روی سرور دیپلوی بشه. مثلاً شما کد رو روی شاخه main پوش میکنید، GitHub Actions تست‌ها رو اجرا میکنه و اگر همه چیز سبز بود خودش به سرور وصل میشه، نسخه جدید رو دریافت میکنه و سرویس رو آپدیت میکنه. نتیجه اینه که دیگه لازم نیست هر بار SSH بزنید، git pull بگیرید و سرویس‌ها رو دستی ریستارت کنید و کل فرآیند انتشار نرم‌افزار reliable، سریع‌تر و کم‌خطاتر میشه.


اما یک نکته مهم وجود داره. خیلی‌ها فکر میکنن CI/CD یعنی GitHub Actions یا GitLab CI یا Jenkins. در صورتی که این‌ها فقط ابزار هستن. CI/CD در اصل یک فرآیند و طرز فکره. هدفش اینه که هر تغییری که وارد پروژه میشه به صورت استاندارد بررسی بشه و انتشار نسخه‌های جدید reliable تر باشه.


یکی از اشتباهات رایج هم اینه که بعضی‌ها فکر میکنن برای داشتن CI باید برن سراغ Kubernetes، AWS، Terraform و کلی ابزار عجیب و غریب. در حالی که برای اکثر پروژه‌های شخصی و حتی خیلی از پروژه‌های واقعی، اولین CI میتونه فقط همین باشه:

git push
↓
ruff
↓
pytest
↓
notification

موضوع مهم بعدی تست‌ هست. واقعیت اینه که CI بدون تست ارزش خیلی زیادی نداره. اگر Pipeline شما فقط اجرا بشه و هیچ تستی وجود نداشته باشه، عملاً فقط دارید یک سری دستورات رو اتوماتیک اجرا میکنید. که دقیقا همون کار رو با pre commit هم میتونید انجام بدید.

تمام هدف CICD اینه که اعتماد شما به کدتون بیشتر بشه و با خیال راحت تری بتونید نسخه های جدید پروژه رو دیپلوی کنید.

معمولاً داخل Pipelineهای حرفه‌ای ممکنه کارهای زیر انجام بشه:

✅ Linting

✅ Formatting Check

✅ Unit Tests

✅ Integration Tests

✅ Type Checking

✅ Security Scanning

✅ Migration Validation

✅ Docker Build

✅ Deploy

✅ Health Check

✅ Notification


در نهایت به نظرم یک Backend Developer لازم نیست DevOps Engineer باشه، اما باید حداقل درکی از فرآیند Build، Test، Deploy و نگهداری سرویس داشته باشه. توانایی طراحی یک Pipeline ساده، اجرای تست‌ها و دیپلوی خودکار پروژه، بیشتر از اینکه یک مهارت DevOps باشه، بخشی از مهارت‌های پایه‌ای یک Software Engineer محسوب میشه.