feat: dlt salesforce scaffold

pyproject.toml

@@ -31,7 +31,11 @@ dependencies = [
   "pypdf>=5.0.0",
   "python-dateutil>=2.9.0.post0",
   "scikit-learn>=1.4.0",
+<<<<<<< HEAD
+  "simple-salesforce>=1.12.6",
+=======
   "sqlalchemy>=2.0.42",
+>>>>>>> main
   "tableauserverclient>=0.25",
   "tenacity>=8.2.3",
   "urllib3>=2.6.3",

src/teamster/code_locations/kipptaf/_dlt/__init__.py

@@ -1,20 +1,32 @@
 from teamster.code_locations.kipptaf._dlt.illuminate.assets import (
     assets as illuminate_assets,
 )
 from teamster.code_locations.kipptaf._dlt.illuminate.schedules import schedules
+<<<<<<< HEAD
+from teamster.code_locations.kipptaf._dlt.salesforce.assets import (
+    assets as salesforce_assets,
+)
+from teamster.code_locations.kipptaf._dlt.zendesk.assets import assets as zendesk_assets
+
+assets = [
+    illuminate_assets,
+    salesforce_assets,
+    zendesk_assets,
+=======
 
 # from teamster.code_locations.kipptaf._dlt.salesforce.assets import (
 #     assets as salesforce_assets,
 # )
 # from teamster.code_locations.kipptaf._dlt.zendesk.assets import assets as zendesk_assets
 
 assets = [
     *illuminate_assets,
     # *zendesk_assets,
     # salesforce_assets,
+>>>>>>> main
 ]
 
 __all__ = [
     "assets",
     "schedules",
 ]

src/teamster/code_locations/kipptaf/definitions.py

@@ -4,7 +4,7 @@
     Definitions,
     load_assets_from_modules,
 )
 from dagster_k8s import k8s_job_executor
 
 from teamster.code_locations.kipptaf import (
     CODE_LOCATION,
@@ -50,7 +50,7 @@
 )
 
 defs = Definitions(
-    executor=k8s_job_executor,
+    # executor=k8s_job_executor,
     assets=[
         *_dbt.asset_specs,
         *_google.asset_specs,

src/teamster/libraries/dlt/salesforce/assets.py

@@ -1,8 +1,18 @@
+<<<<<<< HEAD
+import logging
+
+from dagster import AssetExecutionContext, AssetKey
+from dagster_dlt import DagsterDltResource, DagsterDltTranslator, dlt_assets
+from dlt import pipeline
+from dlt.common.runtime.collector import LogCollector
+from dlt.destinations import bigquery
+=======
 # from dagster import AssetExecutionContext, AssetKey
 # from dagster_dlt import DagsterDltResource, DagsterDltTranslator, dlt_assets
 # from dlt import pipeline
 # from dlt.common.runtime.collector import LogCollector
 # from dlt.destinations import bigquery
+>>>>>>> main
 
 # from teamster.libraries.dlt.salesforce.pipeline import salesforce_source
 
@@ -41,12 +51,21 @@
 #         security_token=salesforce_security_token,
 #     )
 
+<<<<<<< HEAD
+    dlt_pipeline = pipeline(
+        pipeline_name="kippadb",
+        destination=bigquery(),
+        dataset_name=f"dagster_{code_location}_dlt_kippadb",
+        progress=LogCollector(dump_system_stats=False, log_level=logging.DEBUG),
+    )
+=======
 #     dlt_pipeline = pipeline(
 #         pipeline_name="kippadb",
 #         destination=bigquery(),
 #         dataset_name=f"dagster_{code_location}_dlt_kippadb",
 #         progress=LogCollector(dump_system_stats=False),
 #     )
+>>>>>>> main
 
 #     @dlt_assets(
 #         dlt_source=dlt_source,

src/teamster/libraries/dlt/salesforce/pipeline/__init__.py

@@ -27,13 +27,23 @@
 #     """
 #     Retrieves data from Salesforce using the Salesforce API.
 
+<<<<<<< HEAD
+    Args:
+        user_name (str): The username for authentication.
+            Defaults to the value in the `dlt.secrets` object.
+        password (str): The password for authentication.
+            Defaults to the value in the `dlt.secrets` object.
+        security_token (str): The security token for authentication.
+            Defaults to the value in the `dlt.secrets` object.
+=======
 #     Args:
 #         user_name (str): The username for authentication. Defaults to the value in the
 #             `dlt.secrets` object.
 #         password (str): The password for authentication. Defaults to the value in the
 #             `dlt.secrets` object.
 #         security_token (str): The security token for authentication. Defaults to the
 #             value in the `dlt.secrets` object.
+>>>>>>> main
 
 #     Yields:
 #         DltResource: Data resources from Salesforce.
@@ -43,6 +53,137 @@
 #         username=user_name, password=password, security_token=security_token
 #     )
 
+<<<<<<< HEAD
+    opportunity_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+    opportunity_line_item_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+    opportunity_contact_role_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+    account_last_timestamp = dlt.sources.incremental(
+        cursor_path="LastModifiedDate", initial_value=None
+    )
+    campaign_member_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+    task_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+    event_last_timestamp = dlt.sources.incremental(
+        cursor_path="SystemModstamp", initial_value=None
+    )
+
+    # define resources
+    @dlt.resource(write_disposition="replace")
+    def sf_user() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="User")
+
+    @dlt.resource(write_disposition="replace")
+    def user_role() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="UserRole")
+
+    @dlt.resource(write_disposition="merge")
+    def opportunity(
+        last_timestamp: incremental = opportunity_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="Opportunity",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+
+    @dlt.resource(write_disposition="merge")
+    def opportunity_line_item(
+        last_timestamp: incremental = opportunity_line_item_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="OpportunityLineItem",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+
+    @dlt.resource(write_disposition="merge")
+    def opportunity_contact_role(
+        last_timestamp: incremental = opportunity_contact_role_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="OpportunityContactRole",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+
+    @dlt.resource(write_disposition="merge")
+    def account(
+        last_timestamp: incremental = account_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="Account",
+            last_state=last_timestamp.last_value,
+            replication_key="LastModifiedDate",
+        )
+
+    @dlt.resource(write_disposition="replace")
+    def contact() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="Contact")
+
+    @dlt.resource(write_disposition="replace")
+    def lead() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="Lead")
+
+    @dlt.resource(write_disposition="replace")
+    def campaign() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="Campaign")
+
+    @dlt.resource(write_disposition="merge")
+    def campaign_member(
+        last_timestamp: incremental = campaign_member_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="CampaignMember",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+
+    @dlt.resource(write_disposition="replace")
+    def product_2() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="Product2")
+
+    @dlt.resource(write_disposition="replace")
+    def pricebook_2() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="Pricebook2")
+
+    @dlt.resource(write_disposition="replace")
+    def pricebook_entry() -> Iterable[TDataItem]:
+        yield get_records(sf=client, sobject="PricebookEntry")
+
+    @dlt.resource(write_disposition="merge")
+    def task(last_timestamp: incremental = task_last_timestamp) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="Task",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+
+    @dlt.resource(write_disposition="merge")
+    def event(
+        last_timestamp: incremental = event_last_timestamp,
+    ) -> Iterable[TDataItem]:
+        yield get_records(
+            sf=client,
+            sobject="Event",
+            last_state=last_timestamp.last_value,
+            replication_key="SystemModstamp",
+        )
+=======
 #     # define resources
 #     @dlt.resource(write_disposition="replace")
 #     def sf_user() -> Iterable[TDataItem]:
@@ -144,6 +285,7 @@
 #         ),
 #     ) -> Iterable[TDataItem]:
 #         yield get_records(client, "Event", last_timestamp.last_value, "SystemModstamp")
+>>>>>>> main
 
 #     return (
 #         sf_user,

src/teamster/libraries/dlt/salesforce/pipeline/helpers.py

@@ -1,10 +1,16 @@
 # """Salesforce source helpers"""
 
 # from typing import Iterable
 
+<<<<<<< HEAD
+import pendulum
+from dlt.common.typing import TDataItem
+from simple_salesforce.api import Salesforce
+=======
 # import pendulum
 # from dlt.common.typing import TDataItem
 # from simple_salesforce.api import Salesforce, SFType
+>>>>>>> main
 
 # from teamster.libraries.dlt.salesforce.pipeline.settings import IS_PRODUCTION
 
@@ -30,10 +36,15 @@
 #         Dict[TDataItem]: A dictionary representing a record from the Salesforce sObject.
 #     """
 
+<<<<<<< HEAD
+    # Get all fields for the sobject
+    desc = getattr(sf, sobject).describe()
+=======
 #     s_object: SFType = getattr(sf, sobject)
 
 #     # Get all fields for the sobject
 #     desc = s_object.describe()
+>>>>>>> main
 
 #     # Salesforce returns compound fields as separate fields, so we need to filter them
 #     # out

tests/resources/test_resource_salesforce.py

@@ -0,0 +1,22 @@
+from dagster import EnvVar
+from dagster_shared import check
+from simple_salesforce.api import Salesforce
+from simple_salesforce.login import SalesforceLogin
+
+
+def test_foo():
+    session_id, instance = SalesforceLogin(
+        username=check.not_none(
+            value=EnvVar("SALESFORCE_KIPPADB_USER_NAME").get_value()
+        ),
+        password=check.not_none(
+            value=EnvVar("SALESFORCE_KIPPADB_PASSWORD").get_value()
+        ),
+        security_token=check.not_none(
+            value=EnvVar("SALESFORCE_KIPPADB_SECURITY_TOKEN").get_value()
+        ),
+    )
+
+    client = Salesforce(session_id=session_id, instance=instance)
+
+    client.describe()