Refactor

bigquery.py: fix cannot get field of Row_object when calling _query_template()

analytic.py: build `BaseAnaltic` class then inherit it (refactor and clean code)

ga4/analytic/__init__.py

@@ -1 +1,2 @@
+from ga4.analytic.analytic import DeviceAnalytic, EventAnalytic, UserAnalytic
 from ga4.analytic.data_transform import Transformer

ga4/analytic/analytic.py

@@ -4,64 +4,77 @@
 from ga4.model.bigquery import Ga4Table
 
 
-class User:
-    """The features of user"""
+class BaseAnalytic:
     @staticmethod
-    def countries_distribution(table: Ga4Table) -> list:
-        all_users_country = table.geo_country_list
-        counts = Counter(all_users_country)
-
-        return counts
-
+    def attribute_distribution(table: Ga4Table, attribute_name: str) -> Counter:
+        attribute_list = getattr(table, attribute_name)
+        counts = Counter(attribute_list)
 
-class Device:
-    """The features of technology"""
-    @staticmethod
-    def os_distribution(table: Ga4Table) -> list:
-        all_devices_os = table.device_operating_system_list
-        counts = Counter(all_devices_os)
-
         return counts
+
 
-    @staticmethod
-    def category_distribution(table: Ga4Table) -> list:
-        all_devices_category = table.device_category_list
-        counts = Counter(all_devices_category)
-
-        return counts
+class UserAnalytic(BaseAnalytic):
+    """The features of user"""
 
-    @staticmethod
-    def browser_distribution(table: Ga4Table) -> list:
-        all_devices_browser = table.device_web_info_browser_list
-        counts = Counter(all_devices_browser)
-
-        return counts
+    def __init__(self, table: Ga4Table) -> None:
+        self.table = table
 
-    @staticmethod
-    def mobile_brand_distribution(table: Ga4Table) -> list:
-        all_devices_mobile_brand = table.device_mobile_brand_name_list
-        counts = Counter(all_devices_mobile_brand)
-
-        return counts
+    @property
+    def countries_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'geo_country_list'
+        )
 
-    @staticmethod
-    def mobile_model_distribution(table: Ga4Table) -> list:
-        all_devices_mobile_model = table.device_mobile_model_name_list
-        counts = Counter(all_devices_mobile_model)
-
-        return counts
 
+class DeviceAnalytic(BaseAnalytic):
+    """The features of technology"""
 
-class Event:
+    def __init__(self, table: Ga4Table) -> None:
+        self.table = table
+
+    @property
+    def os_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'device_operating_system_list'
+        )
+
+    @property
+    def category_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'device_category_list'
+        )
+
+    @property
+    def browser_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'device_web_info_browser_list'
+        )
+
+    @property
+    def mobile_brand_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'device_mobile_brand_name_list'
+        )
+
+    @property
+    def mobile_model_distribution(self) -> Counter:
+        return self.attribute_distribution(
+            self.table, 'device_mobile_model_name_list'
+        )
+
+
+class EventAnalytic(BaseAnalytic):
     """The features of event"""
 
-    @staticmethod
-    def pages_distribution(table: Ga4Table) -> list:
+    def __init__(self, table: Ga4Table) -> None:
+        self.table = table
+
+    @property
+    def pages_distribution(self) -> Counter:
         """Return most common pages for all users"""
-        all_pages_loc = table.page_location_list
-        counts = Counter(all_pages_loc)
-
-        return counts
+        return self.attribute_distribution(
+            self.table, 'page_location_list'
+        )
 
     @staticmethod
     def track_user_loc():

ga4/model/bigquery.py

@@ -90,7 +90,9 @@ def _query_template(self, query_target: str) -> list:
         self._query_job = self.client.query(query, job_config=self.query_config)
         results = self._query_job.result()
 
-        return [row.user_id for row in results]
+        field_name = query_target.split('.')[-1] if '.' in query_target else query_target
+
+        return [getattr(row, field_name) for row in results]
 
     @property
     @calculate_bytes_processed
@@ -212,12 +214,12 @@ def page_location_list(self) -> list:
         self._query_job = self.client.query(query, job_config=self.query_config)
         results = self._query_job.result()
 
-        return [row.user_id for row in results]
+        return [row.page_location for row in results]
 
     @calculate_bytes_processed
     def query(self, query: str) -> list:
         """Custom query from data table"""
         self._query_job = self.client.query(query, job_config=self.query_config)
         results = self._query_job.result()
 
-        return [row.user_id for row in results]
+        return results