chore(data-warehouse): salesforce add model (#26369)

posthog/migrations/0522_datawarehouse_salesforce_opportunity.py

@@ -0,0 +1,34 @@
+# Generated by Django 4.2.15 on 2024-11-25 02:41
+from django.db import migrations, connection
+
+
+def insert_salesforce_opportunity_schemas(apps, schema_editor):
+    with connection.cursor() as cursor:
+        cursor.execute("SELECT id, team_id FROM posthog_externaldatasource where source_type = 'Salesforce'")
+        salesforce_sources = cursor.fetchall()
+
+    ExternalDataSchema = apps.get_model("posthog", "ExternalDataSchema")
+    for source in salesforce_sources:
+        schema = ExternalDataSchema.objects.create(
+            name="Opportunity",
+            source_id=source[0],
+            team_id=source[1],
+            should_sync=False,
+            sync_type=None,
+            sync_type_config={},
+        )
+        schema.save()
+
+
+def reverse(apps, _):
+    pass
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("posthog", "0521_alter_errortrackingstackframe_context"),
+    ]
+
+    operations = [
+        migrations.RunPython(insert_salesforce_opportunity_schemas, reverse),
+    ]

posthog/migrations/max_migration.txt

@@ -1 +1 @@
-0521_alter_errortrackingstackframe_context
+0522_datawarehouse_salesforce_opportunity

posthog/temporal/data_imports/pipelines/salesforce/__init__.py

@@ -247,6 +247,32 @@ def get_resource(name: str, is_incremental: bool) -> EndpointResource:
             },
             "table_format": "delta",
         },
+        "Opportunity": {
+            "name": "Opportunity",
+            "table_name": "opportunity",
+            **({"primary_key": "Id"} if is_incremental else {}),
+            "write_disposition": {
+                "disposition": "merge",
+                "strategy": "upsert",
+            }
+            if is_incremental
+            else "replace",
+            "endpoint": {
+                "data_selector": "records",
+                "path": "/services/data/v61.0/query",
+                "params": {
+                    "q": {
+                        "type": "incremental",
+                        "cursor_path": "SystemModstamp",
+                        "initial_value": "2000-01-01T00:00:00.000+0000",
+                        "convert": lambda date_str: f"SELECT FIELDS(ALL) FROM Opportunity WHERE SystemModstamp >= {date_str} ORDER BY Id ASC LIMIT 200",
+                    }
+                    if is_incremental
+                    else "SELECT FIELDS(ALL) FROM Opportunity ORDER BY Id ASC LIMIT 200",
+                },
+            },
+            "table_format": "delta",
+        },
         "Account": {
             "name": "Account",
             "table_name": "account",

posthog/temporal/data_imports/pipelines/salesforce/settings.py

@@ -12,6 +12,7 @@
     "Pricebook2",
     "PricebookEntry",
     "Order",
+    "Opportunity",
 )
 
 ENDPOINTS = [
@@ -91,6 +92,14 @@
             "field_type": IncrementalFieldType.DateTime,
         }
     ],
+    "Opportunity": [
+        {
+            "label": "SystemModstamp",
+            "type": IncrementalFieldType.DateTime,
+            "field": "SystemModstamp",
+            "field_type": IncrementalFieldType.DateTime,
+        }
+    ],
     "Account": [
         {
             "label": "SystemModstamp",