Ref #256: Move newsletters mapping to database (#456)

* Read Acoustic Newsletters mapping from database (ref #256)

* Mention command in docs

* Use official return codes

* Use os.EX_DATAERR instead of os.EX_NOTFOUND (available in Unix only)

* Address code review comments

* Address changes on the acoustic_fields command too

* Adjust usage info in commands

ctms/acoustic_service.py

@@ -58,31 +58,6 @@ def force_bytes(s, encoding="utf-8", strings_only=False, errors="strict"):
 
 
 class AcousticResources:
-    MAIN_TABLE_SUBSCR_FLAGS = {
-        # maps the Basket/CTMS newsletter name to the Acoustic name
-        "about-mozilla": "sub_about_mozilla",
-        "app-dev": "sub_apps_and_hacks",
-        "common-voice": "sub_common_voice",
-        "firefox-accounts-journey": "sub_firefox_accounts_journey",
-        "firefox-news": "sub_firefox_news",
-        "firefox-sweepstakes": "sub_firefox_sweepstakes",
-        "hubs": "sub_hubs",
-        "internet-health-report": "sub_internet_health_report",
-        "knowledge-is-power": "sub_knowledge_is_power",
-        "miti": "sub_miti",
-        "mixed-reality": "sub_mixed_reality",
-        "mozilla-and-you": "sub_firefox_news",
-        "mozilla-fellowship-awardee-alumni": "sub_mozilla_fellowship_awardee_alumni",
-        "mozilla-festival": "sub_mozilla_festival",
-        "mozilla-foundation": "sub_mozilla_foundation",
-        "mozilla-rally": "sub_rally",
-        "mozilla-technology": "sub_mozilla_technology",
-        "mozillians-nda": "sub_mozillians_nda",
-        "security-privacy-news": "sub_security_privacy_news",
-        "take-action-for-the-internet": "sub_take_action_for_the_internet",
-        "test-pilot": "sub_test_pilot",
-    }
-
     SKIP_FIELDS = set(
         (
             # Known skipped fields from CTMS
@@ -159,21 +134,21 @@ def __init__(
         self.context: Dict[str, Union[str, int, List[str]]] = {}
         self.metric_service = metric_service
 
-    def convert_ctms_to_acoustic(self, contact: ContactSchema, main_fields: set[str]):
+    def convert_ctms_to_acoustic(
+        self,
+        contact: ContactSchema,
+        main_fields: set[str],
+        newsletters_mapping: dict[str, str],
+    ):
         acoustic_main_table = self._main_table_converter(contact, main_fields)
         newsletter_rows, acoustic_main_table = self._newsletter_converter(
-            acoustic_main_table, contact
+            acoustic_main_table, contact, newsletters_mapping
         )
         product_rows = self._product_converter(contact)
         return acoustic_main_table, newsletter_rows, product_rows
 
     def _main_table_converter(self, contact, main_fields):
-        acoustic_main_table = {
-            # populate with all the sub_flags set to false
-            # they'll get set to true below, as-needed
-            v: "0"
-            for v in AcousticResources.MAIN_TABLE_SUBSCR_FLAGS.values()
-        }
+        acoustic_main_table = {}
         acceptable_subdicts = ["email", "amo", "fxa", "vpn_waitlist", "relay_waitlist"]
         special_cases = {
             ("fxa", "fxa_id"): "fxa_id",
@@ -234,22 +209,28 @@ def fxa_created_date_string_to_datetime(self, inner_value):
             self.context["fxa_created_date_converted"] = "skipped"
         return inner_value
 
-    def _newsletter_converter(self, acoustic_main_table, contact):
+    def _newsletter_converter(self, acoustic_main_table, contact, newsletters_mapping):
         # create the RT rows for the newsletter table in acoustic
         newsletter_rows = []
         contact_newsletters: List[NewsletterSchema] = contact.newsletters
         contact_email_id = str(contact.email.email_id)
         contact_email_format = contact.email.email_format
         contact_email_lang = contact.email.email_lang
         skipped = []
+
+        # populate with all the sub_flags set to false
+        # they'll get set to true below, as-needed
+        for sub_flag in newsletters_mapping.values():
+            acoustic_main_table[sub_flag] = "0"
+
         for newsletter in contact_newsletters:
             newsletter_template = {
                 "email_id": contact_email_id,
                 "newsletter_format": contact_email_format,
                 "newsletter_lang": contact_email_lang,
             }
 
-            if newsletter.name in AcousticResources.MAIN_TABLE_SUBSCR_FLAGS:
+            if newsletter.name in newsletters_mapping:
                 newsletter_dict = newsletter.dict()
                 _today = datetime.date.today().isoformat()
                 newsletter_template["create_timestamp"] = newsletter_dict.get(
@@ -268,9 +249,7 @@ def _newsletter_converter(self, acoustic_main_table, contact):
                 newsletter_rows.append(newsletter_template)
                 # and finally flip the main table's sub_<newsletter> flag to true for each subscription
                 if newsletter.subscribed:
-                    acoustic_main_table[
-                        AcousticResources.MAIN_TABLE_SUBSCR_FLAGS[newsletter.name]
-                    ] = "1"
+                    acoustic_main_table[newsletters_mapping[newsletter.name]] = "1"
             else:
                 skipped.append(newsletter.name)
         if skipped:
@@ -418,7 +397,10 @@ def _insert_update_relational_table(self, table_name, rows):
                 self.context[f"{table_name}_duration_s"] = duration_s
 
     def attempt_to_upload_ctms_contact(
-        self, contact: ContactSchema, main_fields: set[str]
+        self,
+        contact: ContactSchema,
+        main_fields: set[str],
+        newsletters_mapping: dict[str, str],
     ) -> bool:
         """
 
@@ -428,7 +410,7 @@ def attempt_to_upload_ctms_contact(
         self.context = {}
         try:
             main_table_data, nl_data, prod_data = self.convert_ctms_to_acoustic(
-                contact, main_fields
+                contact, main_fields, newsletters_mapping
             )
             main_table_id = str(self.acoustic_main_table_id)
             email_id = main_table_data["email_id"]

ctms/bin/acoustic_fields.py

@@ -1,29 +1,35 @@
 #!/usr/bin/env python3
-"""Manage Acoustic fields: add and remove from the db."""
+"""Manage Acoustic fields: add, list, and remove from the db."""
 
 import argparse
 import os
 import sys
 
 from ctms import config
+from ctms.crud import (
+    create_acoustic_field,
+    delete_acoustic_field,
+    get_all_acoustic_fields,
+)
 from ctms.database import get_db_engine
-from ctms.models import AcousticField
 
 
 def main(dbsession, args=None) -> int:
     parser = argparse.ArgumentParser(
-        description="""
-        Manage Acoustic fields
-        """
+        description="Manage Acoustic fields",
     )
     subparsers = parser.add_subparsers(dest="action")
-    parser_add = subparsers.add_parser("add")
+    parser_add = subparsers.add_parser(
+        "add", usage="""python acoustic_fields.py add "fxaid" """
+    )
     parser_add.add_argument("field")
     parser_add.add_argument(
         "--tablename", "-t", help="Acoustic table name", default="main"
     )
 
-    parser_remove = subparsers.add_parser("remove")
+    parser_remove = subparsers.add_parser(
+        "remove", usage="""python acoustic_fields.py remove "fxaid" """
+    )
     parser_remove.add_argument("field")
     parser_remove.add_argument(
         "--tablename",
@@ -42,37 +48,26 @@ def main(dbsession, args=None) -> int:
 
     args = parser.parse_args(args=args)
     if args.action == "add":
-        dbsession.merge(AcousticField(tablename=args.tablename, field=args.field))
-        dbsession.commit()
-        print("Added.")
+        row = create_acoustic_field(dbsession, args.tablename, args.field)
+        print(f"Added '{row.tablename}.{row.field}'.")
     elif args.action == "remove":
-        row = (
-            dbsession.query(AcousticField)
-            .filter(
-                AcousticField.tablename == args.tablename,
-                AcousticField.field == args.field,
-            )
-            .one_or_none()
-        )
+        row = delete_acoustic_field(dbsession, args.tablename, args.field)
         if not row:
             print(f"Unknown field '{args.tablename}.{args.field}'. Give up.")
             return os.EX_DATAERR
-        dbsession.delete(row)
-        dbsession.commit()
-        print("Removed.")
+        print(f"Removed '{row.tablename}.{row.field}'.")
     else:
-        entries = dbsession.query(AcousticField).all()
+        entries = get_all_acoustic_fields(dbsession)
         print("\n".join(sorted(f"- {e.tablename}.{e.field}" for e in entries)))
 
     return os.EX_OK
 
 
 if __name__ == "__main__":
-    # Get the database
     config_settings = config.Settings()
     engine, session_factory = get_db_engine(config_settings)
     session = session_factory()
     with engine.connect() as connection:
-        ret = main(session)  # pylint:disable = invalid-name
+        return_code = main(session)  # pylint:disable = invalid-name
 
-    sys.exit(ret)
+    sys.exit(return_code)

ctms/bin/acoustic_newsletters_mapping.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""Manage Acoustic newsletters mapping: add, list, and remove from the db."""
+
+import argparse
+import os
+import sys
+
+from ctms import config
+from ctms.crud import (
+    create_acoustic_newsletters_mapping,
+    delete_acoustic_newsletters_mapping,
+    get_all_acoustic_newsletters_mapping,
+)
+from ctms.database import get_db_engine
+
+
+def main(dbsession, test_args=None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Manage Acoustic Newsletter mapping",
+    )
+    subparsers = parser.add_subparsers(dest="action")
+    parser_add = subparsers.add_parser(
+        "add",
+        usage="""python acoustic_newsletters_mapping.py add "test-pilot:sub_new_test_pilot" """,
+    )
+    parser_add.add_argument(
+        "mapping", help="Add newsletter mapping specified as 'source:destination'"
+    )
+
+    parser_remove = subparsers.add_parser(
+        "remove",
+        usage="""python acoustic_newsletters_mapping.py remove "test-pilot" """,
+    )
+    parser_remove.add_argument(
+        "source", help="Remove newsletter mapping with specified 'source'"
+    )
+
+    subparsers.add_parser("list")
+
+    args = parser.parse_args(args=test_args)
+    if args.action == "add":
+        source, destination = args.mapping.split(":")
+        # This will fail if mapping already exists.
+        create_acoustic_newsletters_mapping(dbsession, source, destination)
+        print(f"Added {source!r} → {destination!r}.")
+    elif args.action == "remove":
+        row = delete_acoustic_newsletters_mapping(dbsession, args.source)
+        if not row:
+            print(f"Unknown mapping '{args.source}'. Give up.")
+            return os.EX_DATAERR
+        print(f"Removed {row.source!r} → {row.destination!r}.")
+    else:
+        entries = get_all_acoustic_newsletters_mapping(dbsession)
+        print("\n".join(sorted(f"- {e.source!r} → {e.destination!r}" for e in entries)))
+
+    return os.EX_OK
+
+
+if __name__ == "__main__":
+    config_settings = config.Settings()
+    engine, session_factory = get_db_engine(config_settings)
+    session = session_factory()
+    with engine.connect() as connection:
+        return_code = main(session)  # pylint:disable = invalid-name
+
+    sys.exit(return_code)

ctms/crud.py

@@ -14,6 +14,8 @@
 from .auth import hash_password
 from .database import Base
 from .models import (
+    AcousticField,
+    AcousticNewsletterMapping,
     AmoAccount,
     ApiClient,
     Email,
@@ -843,3 +845,58 @@ def get_product_id(prod: ProductBaseSchema) -> str:
 
     products.sort(key=get_product_id)
     return products
+
+
+def get_all_acoustic_fields(dbsession, tablename=None):
+    query = dbsession.query(AcousticField)
+    if tablename:
+        query = query.filter(AcousticField.tablename == tablename)
+    return query.all()
+
+
+def create_acoustic_field(dbsession, tablename, field):
+    row = AcousticField(tablename=tablename, field=field)
+    dbsession.merge(row)
+    dbsession.commit()
+    return row
+
+
+def delete_acoustic_field(dbsession, tablename, field):
+    row = (
+        dbsession.query(AcousticField)
+        .filter(
+            AcousticField.tablename == tablename,
+            AcousticField.field == field,
+        )
+        .one_or_none()
+    )
+    if row is None:
+        return None
+    dbsession.delete(row)
+    dbsession.commit()
+    return row
+
+
+def get_all_acoustic_newsletters_mapping(dbsession):
+    return dbsession.query(AcousticNewsletterMapping).all()
+
+
+def create_acoustic_newsletters_mapping(dbsession, source, destination):
+    row = AcousticNewsletterMapping(source=source, destination=destination)
+    # This will fail if the mapping already exists.
+    dbsession.add(row)
+    dbsession.commit()
+    return row
+
+
+def delete_acoustic_newsletters_mapping(dbsession, source):
+    row = (
+        dbsession.query(AcousticNewsletterMapping)
+        .filter(AcousticNewsletterMapping.source == source)
+        .one_or_none()
+    )
+    if not row:
+        return None
+    dbsession.delete(row)
+    dbsession.commit()
+    return row

ctms/models.py

@@ -199,6 +199,13 @@ class AcousticField(Base):
     )
 
 
+class AcousticNewsletterMapping(Base):
+    __tablename__ = "acoustic_newsletter_mapping"
+
+    source = Column(String, primary_key=True)
+    destination = Column(String)
+
+
 class VpnWaitlist(Base):
     __tablename__ = "vpn_waitlist"