ledi12

#python #azure

Ostatnio miałem ciekawy problem ci/cd do rozwiązania :P A mianowicie testowanie aplikacji mikroserwisowej w ekosystemie Azure'a przy użyciu cli do lokalnego postawienia funkcji.

Problem był natury wątkowej. Oprócz postawienia serwera funkcji, potrzebowałem również postawić serwer drugiej apki (agregatora danych, fastApi) z którego całość korzysta.

W przypadku azure devops, natywny paraleism działa na poziomie jobów i stage. Wszystkie te byty mogą działać osobno a wymiana danych pomiędzy nimi może polegać chociażby za pomocą artefaktów. Jednak tworzy to kolejny problem a mianowicie budowanie artefaktów, które mają swoje limity, ale nie o tym.

W tym case potrzebowałem trzech osobnych procesów. Jeden odpowiedzialny za postawienie lokalnego cli azure functions. Drugi za odpalenie lokalnego serwer fastApi. Trzeci do egzekucji pytesta.

Do sprawnej komunikacji, wszystko powinno odbyć się w obrębie jednego "taska" po stronie azure devops. Dzięki temu nie trzeba cudować z dodatkową konfiguracją w przypadku np jobów, tak, żeby azure function widziało lokalną wersję fastApi i odwrotnie.

Niestety azure nie posiada "oficjalnego" podejścia w przypadku paraleismu na poziomie pojedyńczego (synchronicznego) taska.

Z pomocą przychodzi import subprocess :) Biblioteka standardowa pozwalająca egzekwować komendy bashowe "pod spodem". Rozwiązaniem okazała się klasa Popen, która jest niczym innym jak nowym procesem. Dodatkowo jest o tyle fajna, że jej wywołanie zwraca adres procesu, którym możemy sterować z wątku nadrzędnego. Możemy go ubić, zatrzymać, czy przerzucić output.

Finalnie rozwiązaniem okazał się prosty moduł testowy (psudo kod):

fastapi = subprocess.Popen(...)
az_functions = subprocess.Popen(...)
subprocess.run("pytest odpal testy suuuko ;p") # funkcja synchroniczna, która stopuje wątek główny.
fastapi.terminate()
az_functions.terminate()

Następnie po stronie taska w pipeline azure devops, wywoałnie banalnie proste:

- task: AzureCLI@2
  inputs:
    azureSubscription: sub
    scriptType: 'bash'
    inlineScript: |
      python3.10 nasz_modul.py

#python #multithreading #ciekawostka

Ostatnio refaktorowałem jeden ze swoich projektów, gdzie kluczowa część jest oparta na wielowątkowości. Tej "wirtualnej" limitowanej przez GIL'a :)
Jak wiadomo, threading.Thread to tak na prawdę dzielenie głównego procesu na mniejsze subprocesy, które współdzielą pamięć (Mogą też zaalokować swoją przy użyciu Thread.local, ale nadal w obrębie wyższej warstwy). Każdy z tych subprocesów działa we własnym zakresie, ale nadal w obrębie głównego procesu, czyli nie ma tutaj mowy o prawdziwym paraellismie pod taski typu cpu-based. Pod I/O jak najbardziej i taki też był mój case. Jest to tzw time-slicing czyli nic innego jako dzielenie czasu procesora i skakanie z jednego "wątku" na drugi.

Mam use cases, gdzie zagnieżdżam jeden "wątek" w drugim. Z początku byłem przekonany, że takie rozwiązanie "przedzieli" istniejący już subprocess na kolejny subprocess, czyli ponownie - Zagnieżdżony wątek będzie zależny od rodzica w którym został zagnieżdżony. Obrazując:

Jak się jednak okazuję, hierarchia "wirtualnych" wątków jest płaska. Oznacza to w praktyce, że bez względu na zagnieżdżanie, czy też nie, każdy nowy "wątek" zawsze będzie tworzony na jednakowym poziomie, czyli nie występuje relacje parent-child pomiędzy zagnieżdżonymi wątkami (Nie wliczając głównego procesu). Finalnie będzie to wyglądać tak:

Prosty test:

import threading
import time

def child():
    time.sleep(10)
    print("finished child processing")

def parent():
    ch = threading.Thread(target=child)
    ch.start()
    print("finished parent processing")

def main():
    p = threading.Thread(target=parent)
    p.start()
    p.join()
    print("finished main")

if __name__ == "__main__":
    main()

--> finished parent processing
--> finished main
--> finished child processing

Gdy by zaszła relacja parent--child, to w momencie skończenia procesu parent, child powinien zostać zterminowany a tak nie jest. Oczywiście mógłbym dodać join() na poziomie parent, żeby poczekać na pełną egzekucje child. Anyway powyższe wskazuję, że child w żaden sposób nie jest zależny od parent.

#eksperymenty #python

Ostatnio refaktorowałem trochę kodu w pracy. Mamy jeden proces, który regularnie się rozrasta o nowe zależności - Identyfikatory.

Każdy identyfikator jest w postaci oddzielnej klasy.

Żeby dodać nowy identyfikator z nową logiką identyfikacji, trzeba stworzyć nową klasę, zaimportować ją w innym module, wrzucić w tuple, która następnie jest iterowana -> Tworzy obiekt -> wywołuję metodę -> zwraca coś.

Zaczęło mnie to irytować, bo import i tupla regularnie rośnie i jest to taka robota głupiego...

Lubię eksperymentować, więc postanowiłem ubić tego potworka przy pomocy małego "metaprogrammingu" :D

Wjechała nowa abstrakcja:

class Identifier(ABC):
  @classmethod
  def identify(cls, schema):
        for _cls in cls.__subclassess__():
          if _type := _cls(schema)._identify(): return _type
        return "NotSupportedType"

  @abstractmethod
  def _identify(self, schema) -> Optional[str]:
    pass

Oczywiście każdy z identyfikatorów implementuje swoje własne metody dla konkretnego przypadku. Są one wywoływane we współdzielonym _identify i zwracają co mają zwracać.

Oczywiście są pewne ryzyka w takim podejściu. Wystarczy, że jakiś śmieszek przedziedziczy po którymś z identyfikatorów i nagle subclasses hook zwróci dodatkową klasę :P

Anyway, moduł mocno się skurczył :) Hook wyciąga sobie klasy na podstawie __mro__ dzięki czemu nie trzeba tworzyć dodatkowego importu dla poszczególnych klas (I je gdzieś trzymać jako wsad dla iteratora...).

Ostatnio eksperymentuje sobie z pythonowym typowaniem.
W przypadku pythona jestem fanem podejścia funkcyjnego a sam język dostarcza multum narzędzi do tego typu zabawa (Bogate moduły dostarczające mechanizmy aka higher order functions).

Jestem też fanem sensownego typowania które dostarcza chociażby TS i takowe też próbowałem odzwierciedlić w pythonie w przypadku funkcji tworzonych w locie. Jak się okazało, można uzyskać bardzo podobny efekt przy użyciu takich klas jak TypedDict oraz Protocol.

TypedDict to nic innego jak typowana relacja klucz-wartość. Natomiast Protocol to taka pythonowa wersja interfejsu.

Z zasady, w pythonie wszystko jest obiektem i każdy z tych obiektów jako meta class dziedziczy po type. Kolejno, bazowa definicja type dostarcza z automatu definicje metody magicznej __call__, która to określa, czy obiekt jest callable a więc, czy jest może działać jako funkcja.

Przykładowy (bardzo ogólny) kod:

class Brands(Enum):
    MERCEDES = "mercedes"

class Car(NamedTuple):
    brand: Brands
    model: str

def drive_car(car: Car, broken: bool = False) -> None:
    if not broken:
        print(f"I'm driving {car.brand.value} {car.model}")
        return
    print(f"This {car.brand.value} is broken :(")

if __name__ == "__main__":

    mercedes_amg = Car(Brands.MERCEDES, "amg")

    drive_mercedes_amg = partial(drive_car, car=mercedes_amg)
    drive_broken_mercedes_amg = partial(drive_mercedes_amg, broken=True)

No dobra, stworzyliśmy dwie nowe funkcje "w locie", ale jakiego typowania użyć? Teoretycznie typing dostarcza typ Callable określający funkcje i to było by teoretycznie okej. Limitacją takiego podejścia jest fakt, że definiujemy tam jedynie typ argumenty, pomijając jego nazwę. W przypadku obiektów, które potrzebują wiele argumentów, takie podejście może wprowadzić pewien chaos.

Na szczęście mając na uwadze fakt istnienia Protocol i __call__ i TypedDict, możemy zrobić pewien sprytny fikołek i uzyskać oczekiwany efekt.

class Brands(Enum):
    MERCEDES = "mercedes"

class Car(NamedTuple):
    brand: Brands
    model: str

class _DriveCarBase(TypedDict):
    car: Car

class DriveCarFunc(Protocol):
    def __call__(self, **car_base: _DriveCarBase) -> Any:
        pass

class DriveBrokenCarFunc(Protocol):
    def __call__(self, broken: bool = True, **car_base: _DriveCarBase) -> Any:
        pass

def drive_car(car: Car, broken: bool = False) -> None:
    if not broken:
        print(f"I'm driving {car.brand.value} {car.model}")
        return
    print(f"This {car.brand.value} is broken :(")

if __name__ == "__main__":

    mercedes_amg = Car(Brands.MERCEDES, "amg")

    drive_mercedes_amg: DriveCarFunc = partial(drive_car, car=mercedes_amg)
    drive_broken_mercedes_amg: DriveBrokenCarFunc = partial(drive_mercedes_amg, broken=True)

Mimo, że z założenia python jest dynamiczny pod względem typowania, to jednak samo typowanie ciągle ewoluuje i dostarcza co raz to nowszych funkcjonalności. Powoli skręca w kierunku TS co w mojej ocenie jest czymś pozytywnym. W wersji 3.12 zaimplementowano mechanizm obsługi json na wzór tego z js czyli bezpośrednie wywołanie atrybutu (obiekt.atrybut). Istniało to oczywiście wcześniej, ale trzeba było instalować 3rd party lib, natomiast teraz mamy to w standardzie :)