Hi All I have built a custom tap to extract data from an API. However, I have hit an issue trying to make an API Call that requires results from two other API calls. For example the structure of the three APIs are v/1/Companies v/1/LeaveTypes?companyid={Company_id '/1/leavebalances?company_id={Company_Id}&leave_type_ids={LeaveType_Id}' I can return the LeaveTypes successfully by using the _get_child_context_ in Companies and define the _parent_stream_type_ to Companies in LeaveTypes The issue I am having is the LeaveBalances requires the CompanyId from Company and LeaveTypeId from LeaveTypes. is it possible to define two _parent_stream_type_ sources in a call? Below is the definition of my streams class Companies(LiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/companies' name = "Companies" schema_filepath = SCHEMAS_DIR / "Companies_Schema.json" def get_child_context(self, record: dict, context: dict) -> dict: """Return a context dictionary for child streams.""" return { "Company_Id": record["Id"], } class LeaveTypes(EasiPayLiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/leavetypes?companyid={Company_Id}' name = "LeaveTypes" parent_stream_type = Companies schema_filepath = SCHEMAS_DIR / "LeaveTypes_Schema.json" def get_child_context(self, record: dict, context: dict) -> dict: """Return a context dictionary for child streams.""" return { "LeaveType_Id": record["Id"], } class LeaveBalances(EasiPayLiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/leavebalances?company_id={Company_Id}&leave_type_ids={LeaveType_Id}' name = "EasiPayLive_LeaveBalances" parent_stream_type = [Companies, LeaveTypes] schema_filepath = SCHEMAS_DIR / "LeaveBalances_Schema.json" Thank you for any help Tim

Does this not already work if you make the

parent_stream_type

of

LeaveBalances

stream

LeaveTypes

only, and modify

LeaveTypes.get_child_context

to be

def get_child_context(self, record: dict, context: dict) -> dict:
        """Return a context dictionary for child streams."""
        return {
            **context,  # or `"Company_Id": context["Company_Id"]` if you want to be more specific
            "LeaveType_Id": record["Id"],
        }

?

Hi Ruben Thank you that worked perfectly. If I am may ask the next question what would me stream need to be if I wanted to use two parameters that weren't linked. Is it even possible? v/1/payruns v/1/LeaveTypes?companyid={Company_id} v/1/leavebalances?payrun_id={payrun_Id}&leave_type_ids={LeaveType_Id}' class PayRuns(LiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/PayRuns' name = "PayRuns" schema_filepath = SCHEMAS_DIR / "PayRuns_Schema.json" def get_child_context(self, record: dict, context: dict) -> dict: """Return a context dictionary for child streams.""" return { "PayRuns_Id": record["Id"], } class LeaveTypes(EasiPayLiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/leavetypes?companyid={Company_Id}' name = "LeaveTypes" parent_stream_type = Companies schema_filepath = SCHEMAS_DIR / "LeaveTypes_Schema.json" def get_child_context(self, record: dict, context: dict) -> dict: """Return a context dictionary for child streams.""" return { "LeaveType_Id": record["Id"], } class LeaveBalances(EasiPayLiveStream): primary_keys = ["id"] rest_method = "Get" path = '/1/leavebalances?payrun_id={payrun_Id}&&leave_type_ids={LeaveType_Id}' name = "EasiPayLive_LeaveBalances" parent_stream_type = [Companies, PayRuns] schema_filepath = SCHEMAS_DIR / "LeaveBalances_Schema.json"

Is the idea here that you would make an auxiliary request for

Company_id

for the

LeaveTypes

stream?

yes sorry, the

LeaveTypes

stream would still use the

Company_id

. However

LeaveBalances

would use

LeaveType_Id

from

LeaveTypes

stream and

PayRun_Id

from

PayRun

stream. can that be done?

I think you can still do it with another child stream, although it is a bit inefficient as in this example a request for pay runs will be made for every leave type of every company:

class Companies(EasiPayLiveStream):
    primary_keys = ["id"]
    rest_method = "Get"
    path = '/1/companies'
    name = "Companies"
    schema_filepath = SCHEMAS_DIR / "Companies_Schema.json"

    def get_child_context(self, record: dict, context: dict) -> dict:
        """Return a context dictionary for child streams."""
        return {
            "Company_Id": record["Id"],
        }

class LeaveTypes(EasiPayLiveStream):
    primary_keys = ["id"]
    rest_method = "Get"
    path = '/1/leavetypes?companyid={Company_Id}'
    name = "LeaveTypes"
    parent_stream_type = Companies
    schema_filepath = SCHEMAS_DIR / "LeaveTypes_Schema.json"

    def get_child_context(self, record: dict, context: dict) -> dict:
        """Return a context dictionary for child streams."""
        return {
            **context,
            "LeaveType_Id": record["Id"],
        }

class PayRuns(EasiPayLiveStream):
    primary_keys = ["id"]
    rest_method = "Get"
    path = '/1/PayRuns'
    name = "PayRuns"
    parent_stream_type = LeaveTypes
    schema_filepath = SCHEMAS_DIR / "PayRuns_Schema.json"

    def get_child_context(self, record: dict, context: dict) -> dict:
        """Return a context dictionary for child streams."""
        return {
            **context,
            "PayRun_Id": record["Id"],
        }

class LeaveBalances(EasiPayLiveStream):
    primary_keys = ["id"]
    rest_method = "Get"
    path = '/1/leavebalances?payrun_id={PayRun_Id}&&leave_type_ids={LeaveType_Id}'
    name = "EasiPayLive_LeaveBalances"
    parent_stream_type = PayRuns
    schema_filepath = SCHEMAS_DIR / "LeaveBalances_Schema.json"

I saw what you were trying to do with a composite

parent_stream_type

which is a nice idea but not supported (I'm not sure of the implications of that design). You can get pretty custom with streams though, so there might be a more efficient way that doesn't use parent-child streams. This might be related: https://github.com/meltano/sdk/issues/93

Hi Rubin thank you for such a detailed and timely response. I shall test your scenarios and update the tread with my findings. Tim