feat: Add product properties streams (#82)

* Add product properties stream

* Add product property specifics stream

* Add product option specifics stream

* Add variant property specifics stream

tap_veeqo/client.py

@@ -13,7 +13,7 @@ class VeeqoStream(RESTStream):
     """Veeqo stream class."""
 
     url_base = "https://api.veeqo.com"
-    primary_keys = ("id",)
+    primary_keys: tuple[str, ...] = ("id",)
     page_size = 1000
 
     @property

tap_veeqo/schemas/product_property.py

@@ -0,0 +1,8 @@
+"""Schema definitions for product property objects."""
+
+from singer_sdk import typing as th
+
+ProductPropertyObject = th.PropertiesList(
+    th.Property("id", th.IntegerType),
+    th.Property("name", th.StringType),
+)

tap_veeqo/schemas/product_property_specific.py

@@ -0,0 +1,10 @@
+"""Schema definitions for product property specific objects."""
+
+from singer_sdk import typing as th
+
+ProductPropertySpecificObject = th.PropertiesList(
+    th.Property("id", th.IntegerType),
+    th.Property("product_property_id", th.IntegerType),
+    th.Property("product_property_name", th.StringType),
+    th.Property("value", th.StringType),
+)

tap_veeqo/schemas/variant_property_specific.py

@@ -0,0 +1,11 @@
+"""Schema definitions for variant property specific objects."""
+
+from singer_sdk import typing as th
+
+VariantPropertySpecificObject = th.PropertiesList(
+    th.Property("id", th.IntegerType),
+    th.Property("product_specific_id", th.IntegerType),
+    th.Property("product_property_id", th.IntegerType),
+    th.Property("product_property_name", th.StringType),
+    th.Property("value", th.StringType),
+)

tap_veeqo/streams.py

@@ -11,11 +11,14 @@
 from tap_veeqo.schemas.order import OrderObject
 from tap_veeqo.schemas.product import ProductObject
 from tap_veeqo.schemas.product_brand import ProductBrandObject
+from tap_veeqo.schemas.product_property import ProductPropertyObject
+from tap_veeqo.schemas.product_property_specific import ProductPropertySpecificObject
 from tap_veeqo.schemas.purchase_order import PurchaseOrderObject
 from tap_veeqo.schemas.sellable import SellableObject
 from tap_veeqo.schemas.store import StoreObject
 from tap_veeqo.schemas.supplier import SupplierObject
 from tap_veeqo.schemas.tag import TagObject
+from tap_veeqo.schemas.variant_property_specific import VariantPropertySpecificObject
 from tap_veeqo.schemas.warehouse import WarehouseObject
 
 
@@ -60,6 +63,10 @@ class ProductsStream(VeeqoStream):
     replication_key = "updated_at"
     schema = ProductObject.to_dict()
 
+    @override
+    def get_child_context(self, record, context):
+        return {"id": record["id"]}
+
 
 class ProductBrandsStream(VeeqoStream):
     """Define product brands stream."""
@@ -69,6 +76,31 @@ class ProductBrandsStream(VeeqoStream):
     schema = ProductBrandObject.to_dict()
 
 
+class ProductPropertiesStream(VeeqoStream):
+    """Define product properties stream."""
+
+    name = "product_properties"
+    path = "/product_properties"
+    schema = ProductPropertyObject.to_dict()
+
+
+class ProductPropertySpecificsStream(VeeqoStream):
+    """Define product property specifics stream."""
+
+    name = "product_property_specifics"
+    parent_stream_type = ProductsStream
+    path = "/products/{id}/product_property_specifics"
+    schema = ProductPropertySpecificObject.to_dict()
+    primary_keys = ("id", "product_property_id")
+
+
+class ProductOptionSpecificsStream(ProductPropertySpecificsStream):
+    """Define product option specifics stream."""
+
+    name = "product_option_specifics"
+    path = "/products/{id}/product_option_specifics"
+
+
 class ProductTagsStream(VeeqoStream):
     """Define product tags stream."""
 
@@ -102,6 +134,20 @@ class SellablesStream(VeeqoStream):
     path = "/sellables"
     schema = SellableObject.to_dict()
 
+    @override
+    def get_child_context(self, record, context):
+        return {"id": record["id"]}
+
+
+class VariantPropertySpecificsStream(VeeqoStream):
+    """Define variant property specifics stream."""
+
+    name = "variant_property_specifics"
+    parent_stream_type = SellablesStream
+    path = "/product_variants/{id}/variant_property_specifics"
+    schema = VariantPropertySpecificObject.to_dict()
+    primary_keys = ("id", "product_specific_id", "product_property_id")
+
 
 class StoresStream(VeeqoStream):
     """Define stores stream."""

tap_veeqo/tap.py

@@ -15,9 +15,13 @@
     streams.OrdersStream,
     streams.ProductsStream,
     streams.ProductBrandsStream,
+    streams.ProductPropertiesStream,
+    streams.ProductPropertySpecificsStream,
+    streams.ProductOptionSpecificsStream,
     streams.ProductTagsStream,
     streams.PurchaseOrdersStream,
     streams.SellablesStream,
+    streams.VariantPropertySpecificsStream,
     streams.StoresStream,
     streams.SuppliersStream,
     streams.TagsStream,