nino_mueller
12/16/2022, 2:41 PMvisch
12/16/2022, 2:43 PMnino_mueller
12/16/2022, 2:45 PMvisch
12/16/2022, 2:56 PMnino_mueller
12/16/2022, 3:29 PMclass InvoicesStream(bexioStream):
"""Invoices stream."""
name = "kb_invoices"
path = "2.0/kb_invoice"
data_key = "kb_invoice"
primary_keys = ["id"]
replication_method = "INCREMENTAL"
replication_key = "id"
schema_filepath = SCHEMAS_DIR / "kb_invoice.json"
def get_child_context(self, record: dict, context: Optional[dict]) -> dict:
"""Return a context dictionary for child streams."""
return {
"invoice_id": record["id"],
}
class InvoiceStream(bexioStream):
"""Invoice stream."""
name = "kb_invoice"
path = "2.0/kb_invoice/{invoice_id}"
data_key = "kb_invoice"
primary_keys = ["id"]
replication_method = "INCREMENTAL"
replication_key = "id"
schema_filepath = SCHEMAS_DIR / "kb_invoice.json"
# EpicIssues streams should be invoked once per parent epic:
parent_stream_type = InvoicesStream
# Assume epics don't have `updated_at` incremented when issues are changed:
ignore_parent_replication_keys = True
visch
12/16/2022, 3:31 PMdata_key
but š¤·nino_mueller
12/16/2022, 3:32 PMnino_mueller
12/16/2022, 3:34 PMvisch
12/16/2022, 3:34 PMvisch
12/16/2022, 3:35 PMnino_mueller
12/16/2022, 3:45 PM