First Behavioral Design Pattern (Observer Design Pattern)

covid_nlp/language/ms_translate.py

@@ -1,11 +1,34 @@
 # -*- coding: utf-8 -*-
 import os, requests, uuid, json
 import sys
-
+from typing import List
 import pandas as pd
-
+from observer import Observer, ConcreteObserver
 
 class MSTranslator():
+
+    # concrete subject
+    _observers: List[Observer] = []
+
+    # The subscription management methods.
+    def attach(self, observer: Observer) -> None:
+        print("Subject: Attached an observer.")
+        self._observers.append(observer)
+
+    def detach(self, observer: Observer) -> None:
+        self._observers.remove(observer)
+
+    # concrete notify when event 
+    def notify(self) -> None:
+        """
+        Trigger an update in each subscriber.
+        """
+
+        print("Subject: Notifying observers...")
+        for observer in self._observers:
+            observer.update(self)
+
+
     def __init__(self, key = None, endpoint = None, lang = None):
         if key:
             self.azure_key = key
@@ -27,6 +50,7 @@ def translate(self, text):
         trans_text = ""
         if len(response) > 0:
             trans_text = response[0]['translations'][0]['text']
+            self.notify()
         return trans_text
 
 
@@ -35,6 +59,9 @@ def main():
     azure_endpoint = "https://api.cognitive.microsofttranslator.com/"
     ms_translator = MSTranslator(endpoint = azure_endpoint, lang = lang)
 
+    concreteObs_a = ConcreteObserver()
+    ms_translator.attach(concreteObs_a)
+
     faq_file = "../../data/faqs/faq_covidbert.csv"
     df = pd.read_csv(faq_file)
     df[f'question_{lang}'] = df.apply(lambda x: ms_translator.translate(x.question), axis=1)

covid_nlp/language/observer.py

@@ -0,0 +1,48 @@
+import csv
+from abc import ABC, abstractclassmethod
+from __future__ import annotations
+import pandas as pd
+from typing import List
+from ms_translate import MSTranslator
+
+
+
+class Subject(ABC):
+    """
+    The Subject interface declares a set of methods for managing subscribers.
+    """
+    @abstractclassmethod
+    def attach(self, observer: Observer) -> None:
+        # attach new subscribers 
+        pass
+
+    @abstractclassmethod
+    def detach(self, observer: Observer) -> None:
+        # detach a subscriber
+        pass
+
+    @abstractclassmethod
+    def notify(self) -> None:
+        # notify the observers in the Observer list about the event
+        pass
+
+class Observer(ABC):
+    """
+    The Observer interface declares the update method, used by subjects.
+    """
+    @abstractclassmethod
+    def update(self, subject: Subject) -> None:
+        """
+        Receive update from subject.
+        """
+        pass
+
+
+
+"""
+Concrete Observers react to the updates issued by the Subject they had been
+attached to.
+"""
+class ConcreteObserver(Observer):
+    def update(self, subjects: Subject) -> None:
+        print("Translation has been done and recorded in faq_covidbert successfully!")