Sync user notification settings

fixattiosync/__init__.py

@@ -11,4 +11,4 @@
 __author__ = "Some Engineering Inc."
 __license__ = "AGPL-3.0"
 __copyright__ = "Copyright © 2024 Some Engineering Inc."
-__version__ = "0.0.17"
+__version__ = "0.0.18"

fixattiosync/fixdata.py

@@ -5,7 +5,7 @@
 from uuid import UUID
 from argparse import ArgumentParser
 from .logger import log
-from .fixresources import FixUser, FixWorkspace, FixCloudAccount, FixRoles
+from .fixresources import FixUser, FixWorkspace, FixCloudAccount, FixRoles, FixUserNotificationSettings
 from typing import Optional
 
 
@@ -83,6 +83,15 @@ def hydrate(self) -> None:
                     rows = cursor.fetchall()
                     for row in rows:
                         self.__workspaces[row["organization_id"]].owner = self.__users[row["user_id"]]
+                with self.conn.cursor(row_factory=dict_row) as cursor:
+                    cursor.execute('SELECT * FROM public."user_notification_settings";')
+                    rows = cursor.fetchall()
+                    for row in rows:
+                        if row["user_id"] in self.__users:
+                            user = self.__users[row["user_id"]]
+                            user.notification_settings = FixUserNotificationSettings(**row)
+                        else:
+                            log.error(f"Data error: notification settings for user {row['user_id']} not found")
                 with self.conn.cursor(row_factory=dict_row) as cursor:
                     cursor.execute('SELECT * FROM public."cloud_account";')
                     rows = cursor.fetchall()

fixattiosync/fixresources.py

@@ -31,6 +31,7 @@ class FixUser:
     workspace_has_subscription: Optional[bool] = None
     registered_at: Optional[datetime] = None
     last_active_at: Optional[datetime] = None
+    notification_settings: Optional[FixUserNotificationSettings] = None
 
     def __post_init__(self) -> None:
         self.registered_at = self.created_at.replace(microsecond=0)
@@ -138,6 +139,8 @@ def attio_person(self) -> dict[str, Any]:
         }
 
     def update_info(self) -> None:
+        if self.notification_settings is not None:
+            self.user_email_notifications_disabled = not self.notification_settings.marketing
         best_workspace = None
         for workspace in self.workspaces:
             if self.workspace_roles[workspace.id] & (
@@ -299,3 +302,14 @@ class FixCloudAccount:
     azure_credential_id: Optional[UUID]
     last_scan_resources_errors: int
     last_degraded_scan_started_at: Optional[datetime]
+
+
+@dataclass
+class FixUserNotificationSettings:
+    user_id: UUID
+    weekly_report: bool
+    inactivity_reminder: bool
+    tutorial: bool
+    marketing: bool
+    created_at: datetime
+    updated_at: datetime

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fixattiosync"
-version = "0.0.17"
+version = "0.0.18"
 authors = [{name="Some Engineering Inc."}]
 description = "Fix Attio Sync"
 license = {file="LICENSE"}