feat: Add QuickBooks verified source

mypy.ini

@@ -40,6 +40,12 @@ ignore_missing_imports=true
 [mypy-pyarrow.*]
 ignore_missing_imports=true
 
+[mypy-quickbooks.*]
+ignore_missing_imports=true
+
+[mypy-intuitlib.*]
+ignore_missing_imports=true
+
 [mypy-tests.*]
 disallow_untyped_defs=false
 disallow_any_generics=false

sources/quickbooks_online/README.md

@@ -0,0 +1,59 @@
+# Quickbooks
+
+QuickBooks is a cloud-based accounting software designed for small to medium-sized businesses. This QuickBooks `dlt` verified source and pipeline example offers the capability to load QuickBooks endpoints such as "Customer" to a destination of your choosing. It enables you to conveniently load the following endpoint as a start:
+
+### Single loading endpoints (replace mode)
+
+| Endpoint | Mode | Description |
+| --- | --- | --- |
+| Customer | replace | A customer is a consumer of the service or product that your business offers. An individual customer can have an underlying nested structure, with a parent customer (the top-level object) having zero or more sub-customers and jobs associated with it. |
+
+
+## Initialize the pipeline with Quickbooks verified source
+```bash
+dlt init quickbooks_online duckdb
+```
+
+Here, we chose DuckDB as the destination. Alternatively, you can also choose redshift, snowflake, or any of the other [destinations.](https://dlthub.com/docs/dlt-ecosystem/destinations/)
+
+## Setup verified source and pipeline example
+
+To grab credentials and initialize the verified source, please refer to the [full documentation here.](https://dlthub.com/docs/dlt-ecosystem/verified-sources/salesforce)
+
+## Add credentials
+
+1. Open `.dlt/secrets.toml`.
+2. Put these credentials in, these can be sourced from quickbooks developer portal and quickbooks oauth playground:
+    ```toml
+    # put your secret values and credentials here. do not share this file and do not push it to github
+    [sources.quickbooks_online]
+    environment=""
+    company_id=""
+    client_id=""
+    client_secret=""
+    access_token=""
+    refresh_token=""
+    redirect_uri=""
+    ```
+
+3. Enter credentials for your chosen destination as per the [docs.](https://dlthub.com/docs/dlt-ecosystem/destinations/)
+
+## Run the pipeline example
+
+1. Install the necessary dependencies by running the following command:
+    ```bash
+    pip install -r requirements.txt
+    ```
+
+2. Now the pipeline can be run by using the command:
+    ```bash
+    python3 quickbooks_online_pipeline.py
+    ```
+
+3. To make sure that everything is loaded as expected, use the command:
+    ```bash
+    dlt pipeline <pipeline_name> show
+    ```
+
+    For example, the pipeline_name for the above pipeline is `quickbooks_online`, you may also use any custom name instead.
+

sources/quickbooks_online/__init__.py

@@ -0,0 +1,71 @@
+"""Source for Quickbooks depending on the quickbooks_online-python python package.
+
+Quickbooks-python docs: https://github.com/ej2/python-quickbooks
+Quickbooks api docs: https://developer.intuit.com/app/developer/qbo/docs/api/accounting/all-entities/
+Quickbooks company id: https://quickbooks.intuit.com/learn-support/en-uk/help-article/customer-company-settings/find-quickbooks-online-company-id/L7lp8O9yU_GB_en_GB
+To get API credentials: https://developer.intuit.com/app/developer/qbo/docs/get-started/start-developing-your-app
+Get oAuth Authorization code from: https://developer.intuit.com/app/developer/playground
+"""
+
+from dlt.sources import DltResource
+
+from typing import Iterable, Sequence
+
+import dlt
+from dlt.common.typing import TDataItem
+from intuitlib.client import AuthClient
+from quickbooks import QuickBooks
+from quickbooks.objects.customer import Customer
+from quickbooks.objects.invoice import Invoice
+
+
+@dlt.source(name="quickbooks_online")
+def quickbooks_online(
+    environment: str = dlt.secrets.value,
+    client_id: str = dlt.secrets.value,
+    client_secret: str = dlt.secrets.value,
+    access_token: str = dlt.secrets.value,
+    refresh_token: str = dlt.secrets.value,
+    company_id: str = dlt.secrets.value,
+    redirect_uri: str = dlt.secrets.value,
+) -> Sequence[DltResource]:
+    """
+    Retrieves data from Quickbooks using the Quickbooks API.
+
+    Args:
+        environment (str): The environment used for authentication ('sandbox' or 'production')
+        client_id (str): The client id provided by quickbooks for authentication. Defaults to the value in the `dlt.secrets` object.
+        client_secret (str): The client secret provided by quickbooks for authentication. Defaults to the value in the `dlt.secrets` object.
+        access_token (str): The access token provided by quickbooks oAuth playground for authentication. Defaults to the value in the `dlt.secrets` object.
+        refresh_token (str): The refresh token provided by quickbooks oAuth playground for authentication. Defaults to the value in the `dlt.secrets` object.
+        company_id (str): The company id / realm id provided by quickbooks. Defaults to the value in the `dlt.secrets` object.
+        redirect_uri (str): The redirect uri provided by quickbooks, found in the developer application created. Defaults to the value in the `dlt.secrets` object.
+    Yields:
+        DltResource: Data resources from Quickbooks.
+    """
+    auth_client = AuthClient(
+        client_id=client_id,
+        client_secret=client_secret,
+        environment=environment,
+        redirect_uri=redirect_uri,
+        access_token=access_token,
+    )
+
+    client = QuickBooks(
+        auth_client=auth_client, refresh_token=refresh_token, company_id=company_id
+    )
+
+    # define resources
+    @dlt.resource
+    def customer() -> Iterable[TDataItem]:
+        customer = Customer.all(qb=client)  # returns a list of iterables
+        for record in customer:
+            yield record.to_dict()
+
+    @dlt.resource
+    def invoice() -> Iterable[TDataItem]:
+        invoice = Invoice.all(qb=client)
+        for record in invoice:
+            yield record.to_dict()
+
+    return [customer, invoice]

sources/quickbooks_online/requirements.txt

@@ -0,0 +1,3 @@
+dlt==1.9.0
+python-quickbooks>=0.9.12
+intuit-oauth==1.2.6

sources/quickbooks_online_pipeline.py

@@ -0,0 +1,16 @@
+import dlt
+from quickbooks_online import quickbooks_online
+
+
+def load_customer() -> None:
+    pipeline = dlt.pipeline(
+        pipeline_name="quickbooks_customer",
+        destination="duckdb",
+        dataset_name="quickbooks_online",
+    )
+    load_info = pipeline.run(quickbooks_online())
+    print(load_info)
+
+
+if __name__ == "__main__":
+    load_customer()

tests/quickbooks_online/test_quickbooks_online_source.py

@@ -0,0 +1,28 @@
+from tests.utils import ALL_DESTINATIONS, assert_load_info, load_table_counts
+import pytest
+import dlt
+from sources.quickbooks_online import quickbooks_online
+
+
+@pytest.mark.parametrize("destination_name", ALL_DESTINATIONS)
+def test_quickbooks_online(destination_name: str) -> None:
+    pipeline = dlt.pipeline(
+        pipeline_name="quickbooks_customer",
+        destination=destination_name,
+        dataset_name="duckdb_customer",
+        dev_mode=True,
+    )
+    data = quickbooks_online()
+    load_info = pipeline.run(data)
+    assert_load_info(load_info)
+
+    expected_tables = ["customer", "invoice"]
+    # only those tables in the schema
+    assert set(t["name"] for t in pipeline.default_schema.data_tables()) == set(
+        expected_tables
+    )
+    # get counts
+    table_counts = load_table_counts(pipeline, *expected_tables)
+    # all tables loaded
+    assert set(table_counts.keys()) == set(expected_tables)
+    assert all(c > 0 for c in table_counts.values())