הבלוג של ינון פרק

הרבה פעמים בתכנות יש שגיאות שנראות ממבט ראשון לא קשורות לחיים, אבל אחרי שאנחנו מבינים מה הקוד שלנו באמת עושה אנחנו מגלים שהשגיאה היא תיאור מדויק של מה שקרה לא בסדר, ושל הדרך לתקן אותו.

כך השגיאה שבכותרת ששמחתי לפגוש היום בקוד פייתון שכתבתי. התחלתי עם גירסא פשוטה לפונקציה download הבאה שלוקחת נתיב ושומרת אותו לקובץ מקומי על המחשב באמצעות Python ו asyncio:

import aiohttp, asyncio, aiofiles

async def download(url, to):
    print(f"Download {url} to {to}")
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            if resp.status == 200:
                f = await aiofiles.open(to, mode='wb')
                await f.write(await resp.read())
                await f.close()

הפונקציה עובדת ואפילו אפשר להשתמש בה כדי להוריד כמה URL-ים במקביל, למשל הקוד הבא מוריד מספר פוסטים מהבלוג:

async def main():
    await asyncio.wait([
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-git-rm-vs-reset', 'post1.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-keep-on-learning', 'post2.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-python-memory-management', 'post3.html')),
        ])

asyncio.run(main())

אבל אתם יודעים איך זה כשלוקחים קוד דוגמא ומעבירים אותו לעולם האמיתי אנחנו צריכים להתחיל להלביש עליו גם בגדים של העולם האמיתי. במקרה של asyncio ותקשורת הבגדים האלה נקראים Throttling, שזה הרעיון שאנחנו לא רוצים להוציא יותר מדי בקשות רשת במקביל. גם בגלל שזה לא בריא למחשב שלנו, אבל בעיקר בגלל שזה לא בריא לשרת וגם לא מנומס. פה בטוקוד יש מנגנון אוטומטי שחוסם את כתובת ה IP שלכם לכמה שעות אם אתם שולחים יותר מדי בקשות במקביל (הכנסתי אותו אחרי שכמה טורקים הפילו לי את האתר פעם אחר פעם עם כלים אוטומטיים שחיפשו פירצות אבטחה).

בכל מקרה ובחזרה לפייתון חשבתי להשתמש ב asyncio.Semaphore כדי לשלוט בכמות הבקשות שנשלחות במקביל: כל בקשה לוקחת אסימון מהסמפור, ואם אין אסימון פנוי מחכים שבקשה אחרת תסיים ותחזיר אחד. הניסיון הראשון שלי לכתוב את זה נראה כך:

# DO NOT USE - CODE WITH BUG
import aiohttp, asyncio, aiofiles

throttle = asyncio.Semaphore(2)

async def download(url, to):
    async with throttle:
        print(f"Download {url} to {to}")
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                if resp.status == 200:
                    f = await aiofiles.open(to, mode='wb')
                    await f.write(await resp.read())
                    await f.close()

async def main():
    await asyncio.wait([
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-git-rm-vs-reset', 'post1.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-keep-on-learning', 'post2.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-python-memory-management', 'post3.html')),
        ])

asyncio.run(main())

והשגיאה לא איחרה לבוא:

Download https://www.tocode.co.il/blog/2019-12-git-rm-vs-reset to post1.html
Download https://www.tocode.co.il/blog/2019-12-keep-on-learning to post2.html
Task exception was never retrieved
future: <Task finished name='Task-4' coro=<download() done, defined at post.py:5> exception=RuntimeError("Task <Task pending name='Task-4' coro=<download() running at post.py:6> cb=[_wait.<locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/tasks.py:507]> got Future <Future pending> attached to a different loop")>
Traceback (most recent call last):
  File "post.py", line 6, in download
    async with throttle:
  File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/locks.py", line 97, in __aenter__
    await self.acquire()
  File "/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/locks.py", line 496, in acquire
    await fut
RuntimeError: Task <Task pending name='Task-4' coro=<download() running at post.py:6> cb=[_wait.<locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/asyncio/tasks.py:507]> got Future <Future pending> attached to a different loop

מה קרה כאן? ההודעה אומרת שיש לנו Future Object שמחובר ל Event Loop אחר. פה המקום להזכיר ש asyncio מאפשר לנו ליצור מספר Event Loops במקביל. כל פעם שאנחנו יוצרים Task או Future Object אותו אוביקט מחובר לאיזושהי Event Loop. הקוד הראשי:

asyncio.run(main())

יוצר את ה Event Loop הראשון במערכת ומריץ בתוכה את הפונקציה main.

בחזרה לשגיאה - אנחנו יכולים לראות שהיא הופיעה רק בקובץ השלישי שאנחנו מנסים להוריד, לכן בוודאות היא הגיעה מה Semaphore, וספציפית מהניסיון להמתין על ה Semaphore (כי שתי הבקשות הראשונות עברו חלק). מזה קל להסיק שהדבר שאנחנו מנסים להמתין עליו אבל לא מחובר ל Event Loop הנוכחי הוא בדיוק ה Semaphore.

ואיך Semaphore (או כל אוביקט אחר) יודע לאיזה Event Loop הוא מחובר? ל asyncio יש פונקציה גלובלית בשם get_event_loop() שמחזירה בדיוק את זה. הבנאי של Semaphore קורא לפונקציה זו ושומר את ה Event Loop. אני מעדכן קצת את הקוד כדי לראות מה ה Event Loop שאותו בנאי יקבל ולהשוות אותה ל Event Loop הראשית של התוכנית:

import aiohttp, asyncio, aiofiles

print("Event Loop when creating the semaphore: ", id(asyncio.get_event_loop()))
throttle = asyncio.Semaphore(2)

async def main():
    print("Event Loop in main()", id(asyncio.get_event_loop()))

asyncio.run(main())

והנה התוצאה:

Event Loop when creating the semaphore:  4516635360
Event Loop in main() 4533746416

עכשיו הסיפור ברור: יצרתי את ה Semaphore מוקדם מדי ולכן הוא מחובר ל Event Loop שונה מזו של ה main. בשביל לתקן את הקוד צריך רק להזיז את יצירת ה Semaphore פנימה לתוך ה main:

import aiohttp, asyncio, aiofiles

throttle = None

async def download(url, to):
    async with throttle:
        print(f"Download {url} to {to}")
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                if resp.status == 200:
                    f = await aiofiles.open(to, mode='wb')
                    await f.write(await resp.read())
                    await f.close()

async def main():
    global throttle
    throttle = asyncio.Semaphore(2)
    await asyncio.wait([
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-git-rm-vs-reset', 'post1.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-keep-on-learning', 'post2.html')),
        asyncio.create_task(download('https://www.tocode.co.il/blog/2019-12-python-memory-management', 'post3.html')),
        ])

asyncio.run(main())

נ.ב. הרבה פעמים נרצה מנגנונים יותר מתוחכמים כדי להאט את התקשורת, למשל נרצה להגדיר שלא נוציא יותר מ-5 בקשות במקביל אבל גם בשום מקרה לא נוציא יותר מ 10 בקשות בשניה (גם אם השרת עונה ממש מהר לשאילתות שלנו). ספריה נחמדה שעוזרת ליצור Throttling מתוחכם יותר ל asyncio נקראת asyncio-throttle ושווה לבדוק אותה לפני שמתחילים לבנות מנגנון לבד בעולם האמיתי.

שפות תכנות רבות דורשות מהמתכנתים לנהל את הזיכרון כלומר להפעיל פקודה כדי "לקבל" זיכרון מהמערכת ואז פקודה נוספת כדי "להחזיר" את הזיכרון למערכת. בשפות כאלה מתכנתים ששכחו להחזיר זיכרון למערכת יצרו קוד שככל שפעל יותר זמן רק דרש יותר ויותר זיכרון מהמערכת, עד שהזיכרון של המכונה נגמר. במצבים כאלה היינו אומרים שיש בתוכנית זליגת זיכרון או Memory Leak באנגלית.

המעבר לשפות שמנהלות לעצמן את הזיכרון תיקן חלקית את הבעיה אבל גם שפות שמנהלות לעצמן את הזיכרון עשויות לטעות ולייצר זליגות זיכרון. שני המודלים המרכזיים בהם שפות מנהלות לעצמן את הזיכרון נקראים Reference Count ו Garbage Collection.

מנגנון Reference Count אומר שהשפה שומרת ליד כל אוביקט מספר שמציין כמה משתנים מתיחסים לאוביקט זה. כשהמספר יורד לאפס באופן אוטומטי הזיכרון של האוביקט חוזר למערכת ההפעלה.

מנגנון Garbage Collection אומר שהשפה מפעילה כל כמה דקות תהליך שמחפש אוביקטים שאי אפשר להגיע אליהם ומחזיר אותם למערכת ההפעלה.

שפת פייתון משתמשת בשילוב של שני המנגנונים כדי לנהל את הזיכרון במקום המתכנת. מנגנון Reference Count פועל בצורה מיידית ומנקה את הזיכרון כשמספר השמות של אוביקט מסוים יורד לאפס, ומנגנון Garbage Collection משמש לגיבוי כדי לזהות מעגלים.

מהו מעגל? שמח ששאלתם. מעגל בהקשר של משתנים וזיכרון אומר שיש לנו מספר משתנים שכל אחד מצביע על אחד אחר. דוגמא קלה היא הקוד הבא ב Python:

a = []
a.append(a)

יצרנו רשימה של איבר אחד שהאיבר הזה הוא מצביע לאותה רשימה ממש. זה נחמד כי עכשיו אפשר לגשת לתא הראשון ברשימה כמה פעמים שרק נרצה:

>>> len(a)
1
>>> len(a[0])
1
>>> len(a[0][0])
1
>>> len(a[0][0][0])
1
>>> len(a[0][0][0][0])
1
>>> len(a[0][0][0][0][0])
1

אבל החיסרון הוא שמנגנון Reference Count כבר לא יכול לעבוד. נלך לראות מה פייתון חושב על הקוד שלנו.

המודול sys מספק פונקציה בשם getrefcount שמחזירה את מספר השמות של אוביקט נתון. הנה כמה דוגמאות:

>>> a = []
>>> sys.getrefcount(a)
2

>>> b = {}
>>> sys.getrefcount(b)
2

>>> a = 10
>>> sys.getrefcount(a)
12

שני הראשונים די מסבירים את עצמם: השם a הוא השם הראשון של הרשימה, ואז בתוך הפונקציה getrefcount יש לרשימה זו שם נוסף (זה שם הפרמטר שהופיע בסוגריים בהגדרת הפונקציה).

הקטע האחרון קצת מבלבל - מאיפה הגיעו 12 שמות למספר 10? הסיבה שבכל מקום שאנחנו כותבים 10 יצרנו עוד שם לדבר הזה בזיכרון שאנחנו קוראים לו המספר 10, וגם פייתון עצמו משתמש בקבוע עשר מדי פעם. ככל שנהיה יותר יצירתיים מספר השמות יורד:

>>> a = 19821
>>> sys.getrefcount(a)
2

עכשיו אפשר לחזור למעגל שלנו:

>>> a = []
>>> a.append(a)
>>> sys.getrefcount(a)
3

לאוביקט יש הפעם 3 שמות: השניים הראשונים מוכרים לנו מהדוגמאות הקודמות והשם השלישי הוא ההצבעה הפנימית. מנגנון Reference Count לעולם לא יוכל למחוק משתנה זה מהזיכרון כי מספר השמות שלו לא ירד אף פעם ל-0. ההצבעה הפנימית תימחק רק כשנאפס אותה ידנית.

לכן פייתון משתמש במנגנון משלים שנקרא Garbage Collection. במנגנון זה פייתון יחפש בצורה פעילה משתנים שאי אפשר להגיע אליהם וימחק אותם מהזיכרון.

נתחיל בחיסול האוביקט שלנו וננסה לראות אם פייתון מחק אותו לבד מהזיכרון:

import sys, gc
a = []
a.append(a)
ida = id(a)
a = 10

# is there still an object with id == i ?
>>> ida in [id(x) for x in gc.get_objects()]
True

הפקודה האחרונה משתמשת במנגנון ה Garbage Collector כדי לזהות האם עדיין יש בזיכרון אוביקט שה id שלו הוא מה שהיה ה id של הרשימה שלנו שהצביעה לעצמה. התשובה היא כמובן True.

זאת בדיוק הסיבה שפייתון מחזיקה גם מנגנון של Garbage Collection לגיבוי. מנגנון זה מכיר את כל האוביקטים במערכת וימחק אוביקטים כאשר אי אפשר להגיע אליהם, גם אם הם כוללים הצבעה מעגלית. מנגנון ה gc מתעורר אוטומטית אחת לכמה דקות או כשאנחנו מבקשים להפעיל אותו עם הפקודה gc.collect. ננסה את זה:

>>> gc.collect()
1

>>> ida in [id(x) for x in gc.get_objects()]
False

ועכשיו הרשימה עם המעגל נמחקה מהזיכרון.

בזכות השילוב בין שני המנגנונים הרבה יותר קשה לייצר זליגות זיכרון בתוכניות פייתון. כדאי לשים לב עם זאת שמנגנון ה Garbage Collector מתעורר בזמנים לא צפויים ועלול לפגוע בביצועים של התוכנית אם אנחנו רוצים לכתוב תוכנית שתגיב בזמן אמת לאירועים חיצוניים. לכן קיימת בפייתון הפקודה gc.disable שמבטלת את מנגנון ה Garbage Collection ומשאירה אותנו רק עם Reference Count. אם אתם בטוחים שאין לכם הצבעות מעגליות באפליקציה וממש צריכים ביצועי Real Time אולי זה משהו ששווה לבדוק אותו.

בגלל העבודה עם קורוטינות ב asyncio שילוב קוד בדיקות איתו עלול להיות קצת מעצבן, לפחות עד שמגלים את הספריה pytest-asyncio.

הספריה pytest-asyncio היא פלאגין ל pytest שמוסיפה אפשרות לסמן בדיקות או Fixtures בתור בדיקות אסינכרוניות ובאופן אוטומטי תפעיל את הבדיקות האלה בתור קורוטינות ותעביר להן את ה event loop.

בשביל לשלב אותה בקוד הבדיקות שלכם מספיק להתקין מ pip ואז לסמן את הבדיקה בתור בדיקה אסינכרונית באופן הבא:

@pytest.mark.asyncio
async def test_luke_name(aiosession):
    data = await fetch(aiosession, 'https://swapi.co/api/people/1/')
    assert data['name'] == 'Luke Skywalker'

בדיקה זו משתמשת בפונקציית עזר בשם fetch (אסינכרונית גם היא) וב Fixture אסינכרוני בשם aiosession. הנה הקוד לשניהם:

@pytest.fixture()
async def aiosession():
    async with aiohttp.ClientSession() as session:
        yield session

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.json()

כל Fixture אסינכרוני מוגדר באופן אוטומטי להיות בעל תחום הגדרה של פונקציה. בשביל להשתמש בתחום הגדרה גדול יותר אנחנו צריכים לדרוס את ה Fixture שנקרא event_loop של ספריית pytest-asyncio ולתת לו תחום הגדרה גדול יותר. בדוגמא הבאה אני מגדיל את תחום ההגדרה של aiosession כך שיישמר לאורך כל חיי תוכנית הבדיקה:

@pytest.fixture(scope='session')
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope='session')
async def aiosession():
    async with aiohttp.ClientSession() as session:
        yield session

חלק גדול מהכיף בעבודה עם pytest הוא שיש פיתרונות מוכנים כמעט לכל סוג בדיקות שתרצו לכתוב. כמו בהרבה מקרים כשמדובר בפייתון שווה לעשות מאמץ ולחפש פיתרון מדף ב pip לפני שרצים לכתוב לבד.

ספריות בדיקות רבות תומכות בהפעלת קוד משותף לפני כל בדיקה, ובספריית pytest קוד האתחול הזה מקבל צורה חדשה ומרעננת שנקראת Fixture. בואו נראה מה זה Fixture ואיך משתמשים בו.

השבוע בשיעור שבכלל לא דיבר על Decorators יצא לי לכתוב עם התלמידים קוד עבור מפענח שקורא הוראות מקלט חיצוני ומבצע אותן לפי הסדר. המפענח שלנו החזיק "זיכרון" וכל הוראה יכלה לקבל פרמטר מספרי או פרמטר שהוא אות, כאשר אם קיבלנו אות היינו צריכים לקרוא את הערך שמתאים לאות זו מהזיכרון.

לדוגמא המפענח תומך בפעולת set ששומרת ערך לזיכרון. אפשרות אחת היא להעביר ל set מספר:

set a 10

ואז בתא a בזיכרון נשמור את המספר 10. אפשרות שניה היא להעביר ל set אות ואז נצטרך לקרוא מהזיכרון את הערך בתא המתאים, לדוגמא:

set a 10
set b a

ישמור בתא a את הערך 10 ובתא b את הערך ששמור עכשיו ב a, כלומר גם 10.

בקיצור יצא שכתבנו קוד עם אלמנט חזרתי שנראה בערך כך:

class MyParser:
    # ...
    def set(self, reg, val):
        ival = self.read_register_or_value(val)
        self.memory[reg] = val

    def sub(self, reg, val):
        ival = self.read_register_or_value(val)
        self.memory[reg] -= ival

    def mul(self, reg, val):
        ival = self.read_register_or_value(val)
        self.memory[reg] *= ival

כל פונקציה משתמשת בפונקציית עזר כדי לבדוק אם הערך שקיבלה הוא מספר או אות, ואם אות אז לקרוא מהזיכרון את הערך המתאים.

הקוד עובד והשימוש החוזר בפונקציית העזר באמת משפר את המצב בהשוואה לכתיבת הלוגיקה המלאה שלוש פעמים אבל עדיין משהו כאן צורם: כל פונקציה צריכה להגדיר משתנה חדש מה שמבזבז לנו שם, והמבנה של שורה ראשונה בפונקציה שמוקדשת להכנת המשתנים עלול לשעמם כשרואים אותו שוב ושוב.

דרך אחת להפוך את הקוד לקצת יותר נעים לעין היא להמיר את הקריאה היזומה בקריאה ל Decorator:

class MyParser:
    @read_value_from_register
    def set(self, reg, val):
        self.memory[reg] = val

    @read_value_from_register
    def sub(self, reg, val):
        self.memory[reg] -= val

    @read_value_from_register
    def mul(self, reg, val):
        self.memory[reg] *= val

כל מה שצריך בשביל המעבר למבנה החדש הוא להגדיר את הפונקציה read_value_from_register מחוץ למחלקה בתור Decorator:

def read_value_from_register(f):
    def inner(self, reg, val):
        ival = self.read_register_or_value(val)
        f(self, reg, ival)

    return inner

מעניין לשים לב שבמקרה כזה הפונקציה f שעוברת כפרמטר היא עדיין לא Object Method אלא פונקציה פשוטה. היא עדיין לא קשורה לאוביקט מאחר וה Decorator נקרא לפני שנוצרו אוביקטים מסוג MyParser. מסיבה זו בהפעלת הפונקציה אנחנו חייבים להעביר את self בצורה יזומה בתור הפרמטר הראשון. מלבד תשומת הלב המיוחדת ל self, הגדרת Decorator למתודה (שנמצאת בתוך מחלקה) זהה לחלוטין להגדרת Decorator לפונקציה רגילה.

בפייתון PEP 249 מאפיין את הממשק הרצוי לדרייברים לבסיסי הנתונים השונים בפייתון. בואו נראה 3 דוגמאות לעבודה עם בסיס נתונים SQLite כדי לקבל מושג איך הממשק הזה נראה.

הכי כיף לשלב בין הידע שלכם בתכנות לחיים האמיתיים ולכתוב כלים קטנים שעוזרים בכל מיני משימות של היום יום. בפוסט היום נכתוב כלי קטן של כרטיסיות מילים שיקרא רשימת כרטיסיות מילים מקובץ ויבקש מאתנו לתרגם מילה בין שתי שפות. בדרך נלמד גם קצת פייתון.

הרבה פעמים מעניין אותנו להריץ סקריפט ולזהות באיזה מודולים שיש לנו על המכונה הסקריפט משתמש, למשל כדי שנוכל לבנות סביבה וירטואלית עבור הרצת סקריפט זה. מסתבר שלפייתון יש פיתרון ממש פשוט לבעיה, והוא נקרא modulefinder.

קחו את התוכנית הבאה לדוגמא:

import re
import sys

for line in sys.stdin:
    if re.search(r'\b[A-Z][a-z]*\b', line):
        sys.stdout.write(line)
    else:
        print("No interesting words found")

בשביל לגלות באיזה מודולים היא משתמשת, ומי מהם הותקן בצורה חיצונית, נוכל להשתמש ב modulefinder. בהנחה ששמרתי את הסקריפט בקובץ בשם hello.py התוכנית הבאה תדפיס את כל המודולים שהקובץ מייבא, ועבור כל מודול תדפיס האם הוא הגיע עם פייתון או הותקן בנפרד:

from modulefinder import ModuleFinder
from stdlib_list import stdlib_list
libraries = stdlib_list("3.7")
import sys
finder = ModuleFinder()
finder.run_script('hello.py')

print('Loaded modules:')
for name, mod in finder.modules.items():
    if '.' in name: continue
    if name in sys.builtin_module_names: continue

    print('%s: ' % name, end='')
    if name in libraries:
        print("Builtin module")
    else:
        print("External module")

כמה הערות:

1. בשביל להשתמש ב modulefinder אנחנו צריכים בסך הכל להריץ את הסקריפט עם finder.run_script ואחרי זה לקרוא את הרשימה מ finder.modules.

2. אני משתמש בשתי היוריסטיקות כדי לדעת אם מודול הוא חלק מפייתון. זה מכסה חלק מהמקרים אבל עדיין לא את כולם. הספריה stdlib_list לוקחת מהאתר של פייתון את רשימת המודולים ששייכים לגירסא מסוימת, והמשתנה sys.builtin_module_names מכיל את רשימת המודולים שבנויים לתוך פייתון. עדיין יש מודולים שהם חלק ממודול אחר, לדוגמא _ssl שהוא חלק ממודול ssl ואותם הסקריפט לא מזהה כמודולים פנימיים. רעיונות להיפטר ממודולים אלה יתקבלו בברכה.

המושג העמסת פונקציות מתיחס ליכולת של שפות תכנות מסוימות להדביק מספר מימושים לכל שם של פונקציה, ולהחליט בזמן הקומפילציה לפי החתימה איזה מימוש להפעיל בכל סיטואציה. בשפות דינמיות אין קומפיילר ולכן המושג המקביל בשפות כמו רובי או פייתון הוא Multiple Dispatching. גם המימוש קצת שונה, ומנגנון Multiple Dispatching עובד על הרעיון שבזמן ריצה אנחנו מסתכלים איזה פרמטרים פונקציה קיבלה ולפי זה מחליטים איזה מימוש שלה להפעיל.

אם ניקח את Python כדוגמא נוכל לדמיין את הקוד הבא שמגדיר שני מימושים לאותה פונקציה:

@dispatch(tuple)
def go(point):
    x, y = point
    go(x, y)

@dispatch(int, int)
def go(x, y):
    print(f'x = {x}, y = {y}')

כאשר כל אחד מהמימושים אחראי על סט פרמטרים אחר. נרצה שהראשון ייקרא אם נפעיל את הפונקציה עם פרמטר יחיד:

loc = (10, 10)
go(loc)

ונרצה שהשני ייקרא כשנפעיל את הפונקציה עם שני פרמטרים:

go(10, 10)

נשמע לכם דמיוני? מסתבר שכל מה שצריך בשביל לגרום לקסם הזה לעבוד הוא לכתוב את ה Decorator. המפתח למימוש ה Decorator הוא לשמור מילון של מילונים, בו שם של פונקציה הוא המפתח הראשי, ואז לכל פונקציה אנחנו שומרים מילון שהמפתח שלו הוא סט פרמטרים מסוים והערך הוא הגדרה מסוימת של הפונקציה. במקרה שלנו המילון הגדול יראה כך:

{
    go: {
        (typle): <function at 0x1088b01e0>
        (int, int): <function at 0x1089d1488>
    }
}

הפונקציה go עצמה שהתוכנית מפעילה, כלומר זו שה Decorator מחזיר תסתכל ברשימת הפרמטרים שקיבלה, תלך למילון לחפש שם מפתח שמתאים לסוגים שלהם ולפי זה תחליט לאיזה פונקציה לקרוא.

מימוש פשוט של המנגנון יכול להיראות כך:

import collections

md = collections.defaultdict(dict)

def dispatch(*types):
    def decorator(f):
        md[f.__name__][types] = f
        def wrapper(*args, **kwargs):
            res = md[f.__name__][tuple(type(a) for a in args)]
            res(*args, **kwargs)

        return wrapper
    return decorator

ואם אתם בעניין של מימוש יותר רציני שגם נבדק בעולם האמיתי וכולל די הרבה אופטימיזציות שווה לבדוק את הספריה multipledispatch. זה דף הבית שלהם:

https://github.com/mrocklin/multipledispatch

חבר כותב סקריפט בפייתון ובמהלך הכתיבה הוא כל הזמן מעדכן דברים קטנים בסקריפט ומריץ לקבלת קובץ פלט. בסוף הוא נתקע עם 20 קבצי פלט ל-20 הרצות שונות, אבל לא בדיוק זוכר איזה קובץ מתאים לאיזה פרמטר או שינוי בסקריפט.

מאחר והוא עובד ב git הצעתי שיוסיף את מזהה הקומיט האחרון לשם קובץ הפלט. בפייתון אפשר לעשות את זה בדרך שלא תלויה בפלטפורמה ותעבוד אפילו אם אין לכם גיט מותקן על המחשב - והמפתח הוא הספריה dulwich. ספריה זו היא מימוש בפייתון של פרוטוקול גיט. תוכלו לכתוב איתה סקריפטים מתוחכמים שעוברים על הריפוזיטורי, או במקרה שלנו פשוט לשלוף את מזהה הקומיט האחרון.

מתקינים את הספריה עם:

$ pip install dulwich --global-option="--pure"

ומתוך תוכנית פייתון נוכל עכשיו לכתוב:

from dulwich.repo import Repo
r = Repo('.')
last_commit_id = r.head().decode('ascii')

result_filename = f'result.{last_commit_id}.txt'

with open(result_filename, 'w') as f:
    f.write('Hello World\n')

כדי לקבל את קובץ התוצאות עם מזהה הקומיט האחרון.