Skip to content

Commit

Permalink
Merge pull request #51 from archdotdev/accounting-streams
Browse files Browse the repository at this point in the history 
feat: Accounting streams coverage
  • Loading branch information
@cjohnhanson
cjohnhanson authored Nov 15, 2024
2 parents d955d33 + faa371b commit 8a43ebb
Show file tree
Hide file tree
Showing 3 changed files with 327 additions and 1 deletion.
2 changes: 1 addition & 1 deletion scripts/autogenerate.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ def get_prompts(
"pbcopy", env={"LANG": "en_US.UTF-8"}, stdin=subprocess.PIPE
)
process.communicate(text.encode("utf-8"))
print(f"Path: {path}")
print(path)
click.pause()
except KeyError:
pass
Expand Down
320 changes: 320 additions & 0 deletions tap_service_titan/streams/accounting.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from functools import cached_property

from singer_sdk import typing as th # JSON Schema typing helpers
import logging

from tap_service_titan.client import (
ServiceTitanExportStream,
Expand Down Expand Up @@ -553,3 +554,322 @@ class InventoryBillsStream(ServiceTitanExportStream):
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/export/inventory-bills"


class ApCreditsStream(ServiceTitanStream):
"""Define ap credits stream."""

name = "ap_credits"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"
schema = th.PropertiesList(
th.Property("id", th.IntegerType),
th.Property("inventoryReturnId", th.IntegerType),
th.Property("jobId", th.IntegerType),
th.Property("active", th.BooleanType),
th.Property("createdOn", th.DateTimeType),
th.Property("modifiedOn", th.DateTimeType),
th.Property("date", th.DateTimeType),
th.Property("canceledOn", th.DateTimeType),
th.Property("number", th.StringType),
th.Property("referenceNumber", th.StringType),
th.Property("memo", th.StringType),
th.Property("amount", th.NumberType),
th.Property("appliedAmount", th.NumberType),
th.Property("status", th.StringType),
th.Property("syncStatus", th.StringType),
th.Property(
"batch",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("number", th.StringType),
th.Property("name", th.StringType),
),
),
th.Property(
"businessUnit",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"remittanceVendor",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"vendor",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property("paymentStatus", th.StringType),
th.Property(
"splits",
th.ArrayType(
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("active", th.BooleanType),
th.Property("createdOn", th.DateTimeType),
th.Property("inventoryBillId", th.IntegerType),
th.Property("vendorCreditId", th.IntegerType),
th.Property("amount", th.NumberType),
)
),
),
).to_dict()

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/ap-credits"


class ApPaymentsStream(ServiceTitanStream):
"""Define ap payment stream."""

name = "ap_payments"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"
schema = th.PropertiesList(
th.Property("id", th.IntegerType),
th.Property("active", th.BooleanType),
th.Property("createdOn", th.DateTimeType),
th.Property("modifiedOn", th.DateTimeType),
th.Property("date", th.DateTimeType),
th.Property("approvedDate", th.DateTimeType),
th.Property("method", th.StringType),
th.Property("name", th.StringType),
th.Property("printCheck", th.BooleanType),
th.Property("amount", th.NumberType),
th.Property("errorMessage", th.StringType),
th.Property("status", th.StringType),
th.Property("syncStatus", th.StringType),
th.Property(
"batch",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("number", th.StringType),
th.Property("name", th.StringType),
),
),
th.Property(
"glAccount",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
th.Property("number", th.StringType),
th.Property("type", th.StringType),
th.Property("detailType", th.StringType),
),
),
th.Property(
"businessUnit",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"vendor",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"remittanceVendor",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"splits",
th.ArrayType(
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("active", th.BooleanType),
th.Property("createdOn", th.DateTimeType),
th.Property("documentId", th.IntegerType),
th.Property("inventoryBillId", th.IntegerType),
th.Property("amount", th.NumberType),
)
),
),
).to_dict()

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/ap-payments"


class PaymentTermsStream(ServiceTitanStream):
"""Define payment terms stream."""

name = "payment_terms"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"

schema = th.PropertiesList(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
th.Property("dueDayType", th.StringType),
th.Property("dueDay", th.IntegerType),
th.Property("isCustomerDefault", th.BooleanType),
th.Property("isVendorDefault", th.BooleanType),
th.Property("active", th.BooleanType),
th.Property("inUse", th.BooleanType),
th.Property("modifiedOn", th.DateTimeType),
th.Property(
"paymentTermDiscountModel",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("discountApplyTo", th.StringType),
th.Property("discount", th.NumberType),
th.Property("discountType", th.StringType),
th.Property("account", th.StringType),
th.Property("applyBy", th.StringType),
th.Property("applyByValue", th.IntegerType),
),
),
th.Property(
"interestSettings",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("rateType", th.StringType),
th.Property("flatRateValue", th.NumberType),
th.Property("percentageRateValue", th.NumberType),
th.Property("chargeMethod", th.StringType),
th.Property("frequency", th.StringType),
th.Property("gracePeriod", th.IntegerType),
th.Property("targetInvoices", th.ArrayType(th.StringType)),
th.Property("taskId", th.IntegerType),
th.Property("taskDisplayName", th.StringType),
),
),
).to_dict()

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/payment-terms"


class PaymentTypesStream(ServiceTitanStream):
"""Define payment types stream."""

name = "payment_types"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"

schema = th.PropertiesList(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
th.Property("modifiedOn", th.DateTimeType),
).to_dict()

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/payment-types"


class TaxZonesStream(ServiceTitanStream):
"""Define tax zones stream."""

name = "tax_zones"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"

schema = th.PropertiesList(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
th.Property("color", th.IntegerType),
th.Property("isTaxRateSeparated", th.BooleanType),
th.Property("isMultipleTaxZone", th.BooleanType),
th.Property(
"rates",
th.ArrayType(
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("taxName", th.StringType),
th.Property("taxBaseType", th.ArrayType(th.StringType)),
th.Property("taxRate", th.NumberType),
th.Property("salesTaxItem", th.StringType),
)
),
),
th.Property("createdOn", th.DateTimeType),
th.Property("modifiedOn", th.DateTimeType),
th.Property("active", th.BooleanType),
).to_dict()

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/tax-zones"


class JournalEntriesStream(ServiceTitanStream):
"""Define journal entries stream."""

name = "journal_entries"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key: str = "modifiedOn"
schema = th.PropertiesList(
th.Property("id", th.StringType),
th.Property("createdOn", th.DateTimeType),
th.Property("modifiedOn", th.DateTimeType),
th.Property("number", th.IntegerType),
th.Property("name", th.StringType),
th.Property("status", th.StringType),
th.Property("postDate", th.DateTimeType),
th.Property("exportedOn", th.DateTimeType),
th.Property(
"exportedBy",
th.ObjectType(
th.Property("id", th.IntegerType),
th.Property("name", th.StringType),
),
),
th.Property(
"customFields",
th.ArrayType(
th.ObjectType(
th.Property("name", th.StringType),
th.Property("value", th.StringType),
)
),
),
th.Property("url", th.StringType),
).to_dict()

def get_url_params(
self,
context: dict | None,
next_page_token: Any | None, # noqa: ANN401
) -> dict[str, Any]:
"""Return a dictionary of values to be used in URL parameterization.
Args:
context: The stream context.
next_page_token: The next page index or value.
Returns:
A dictionary of URL query parameters.
"""
params = super().get_url_params(context, next_page_token)
# This endpoint has an undocumented max page size of 500
params["pageSize"] = 500
return params

@cached_property
def path(self) -> str:
"""Return the API path for the stream."""
return f"/accounting/v2/tenant/{self._tap.config['tenant_id']}/journal-entries"
6 changes: 6 additions & 0 deletions tap_service_titan/tap.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,12 @@ def discover_streams(self) -> list[ServiceTitanStream]:
streams.accounting.InvoicesStream(self),
streams.accounting.PaymentsStream(self),
streams.accounting.InventoryBillsStream(self),
streams.accounting.ApCreditsStream(self),
streams.accounting.ApPaymentsStream(self),
streams.accounting.PaymentTermsStream(self),
streams.accounting.PaymentTypesStream(self),
streams.accounting.TaxZonesStream(self),
streams.accounting.JournalEntriesStream(self),
streams.crm.BookingsStream(self),
streams.crm.CustomerContactsStream(self),
streams.crm.CustomersStream(self),
Expand Down

0 comments on commit 8a43ebb

Please sign in to comment.