Hello everyone 🙂 I'm trying to build a python source connector. The problem is we have many cities and the same endpoints for all. So what I'm trying to do is make just one incremental stream for all cities in my list. However I can't save multiple state to a single stream. Could you help me? Here my code and my sample_state.json

{
    "historical_ocupation":{
      "city1_historical_occupation": {
        "date": "2023-01-20T00:00:00Z"
      },
      "city2_historical_occupation": {
        "date": "2023-01-18T00:00:00Z"
      },
      "city3_historical_occupation": {
        "date": "2023-01-20T00:00:00Z"
      }
    },
    "historical_daily_public":{
      "city1_historical_daily_public": {
        "date": "2023-01-20T00:00:00Z"
      },
      "city2_historical_daily_public": {
        "date": "2023-01-20T00:00:00Z"
      },
      "city3_historical_daily_public": {
        "date": "2023-01-20T00:00:00Z"
      }
    }
  }


class KeyAccessStream(HttpStream):

    def __init__(self, url, sites: List[str], **kwargs):
        super().__init__(**kwargs)
        self.url = url
        self.sites = sites
        self.localizacao = self.sites[0]

    @property
    def url_base(self) -> str:
        return f'{self.url}/dashboards/'
    def next_page_token(self,
                        response: requests.Response) -> Optional[Mapping[str,
                                                                         Any]]:
        """
        :param response: the most recent response from the API
        :return If there is another page in the result, a mapping (e.g: dict)
                containing information needed to query the next page
                in the response.
                If there are no more pages in the result, return None.
        """
        return None

    def parse_response(self,
                       response: requests.Response,
                       **kwargs) -> Iterable[Mapping]:
        """
        :return an iterable containing each record in the response
        """
        if isinstance(response.json(), list):
            for item in response.json():
                if isinstance(item, dict):
                    item.update({'site':self.localizacao})
                    yield item
        if isinstance(response.json(), dict):
            for key, item in dict(response.json()).items():
                if isinstance(item, list):
                    for value in item:
                        if isinstance(value, dict):
                            value.update({'site':self.localizacao})
                            yield value
                    return
    
    def stream_slices(self, *, sync_mode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None) -> Iterable[Optional[Mapping[str, Any]]]:
        return self.sites

class IncrementalKeyAccessStream(KeyAccessStream, IncrementalMixin):

    # Fill in to checkpoint stream reads after N records.
    # This prevents re-reading of data if the stream fails for any reason.
    state_checkpoint_interval = None
    _cursor_value = None

    def __init__(self, url, sites, start_date: datetime, **kwargs):
        super().__init__(url, sites, **kwargs)

        self.start_date = start_date
        self._cursor_value = None


    @property
    def cursor_field(self) -> str:
        """
        :return str: The name of the cursor field.
        """
        return "date"

    @property
    def state(self) -> Mapping[str, Any]:
        """
        :return: A dictionary representing the current state of the stream.
        """
        state = {}
        if self._cursor_value:
            state[casing.camel_to_snake(self.localizacao+self.__class__.__name__)] = {self.cursor_field: self._cursor_value}
        else:
            state[casing.camel_to_snake(self.localizacao+self.__class__.__name__)] = { self.cursor_field: self.start_date.strftime("%Y-%m-%dT%H:%M:%SZ")}
        return state

    @state.setter
    def state(self, value: Mapping[str, Any]):
        """
        :param value: A dictionary representing the new state of the stream.
        """
        self._cursor_value = value[casing.camel_to_snake(self.__class__.__name__)][casing.camel_to_snake(self.localizacao+self.__class__.__name__)][self.cursor_field]


    def read_records(self, *args, **kwargs) -> Iterable[Mapping[str, Any]]:
        if self._cursor_value:
            print(self._cursor_value)
        for record in super().read_records(*args, **kwargs):
            latest_record_date = datetime.strptime(record[self.cursor_field],
                                                   "%Y-%m-%dT%H:%M:%SZ")
            if self._cursor_value:
                if self._cursor_value < str(latest_record_date):
                    self._cursor_value = str(latest_record_date)
                    yield record
            else:
                self._cursor_value = str(latest_record_date)
                yield record 


class HistoricalOccupation(IncrementalKeyAccessStream):

    primary_key = None

    @property
    def http_method(self) -> str:
        return "POST"

    def request_body_json(
        self,
        stream_state: Mapping[str, Any],
        stream_slice: Mapping[str, Any] = None,
        next_page_token: Mapping[str, Any] = None,
    ) -> Optional[Mapping[str, Any]]:
        """
        Override when creating POST/PUT/PATCH requests
        to populate the body of the request with a non-JSON payload.
        If returns a ready text that it will be sent as is.
        If returns a dict that it will be converted to a urlencoded form.
        E.g. {"key1": "value1", "key2": "value2"} => "key1=value1&key2=value2"
        At the same time only one of the
        'request_body_data' and 'request_body_json'
        functions can be overridden.
        """

        if stream_state:
            print("Aqui")
            start_at = datetime.strptime(
                                         stream_state[casing.camel_to_snake(self.localizacao+self.__class__.__name__)][self.cursor_field],
                                         "%Y-%m-%dT%H:%M:%SZ")
            start_at = start_at.strftime("%Y-%m-%dT%H:%M:%SZ")
            print("Recuperei o state:",start_at)
        else:
            start_at = self.start_date.strftime("%Y-%m-%dT%H:%M:%SZ")
            print(start_at, self.localizacao)
            print(casing.camel_to_snake(self.localizacao+self.__class__.__name__))

        return {"companies": None,
                "category": None,
                "scale": "HOURS",
                "endAt": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
                "startAt": start_at}

    def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
        self.localizacao = stream_slice
        return stream_slice + "/historical-occupation"

I think you would need to dynamically create a new stream for each city. An example of dynamically creating new streams based on the values returned in another stream is available at: https://airbyte.com/tutorials/extract-data-from-the-webflow-api

Thanks for your answer Alex! I already tried this and worked, but I have a problem we can't dynamically join this table after and it will be very complicated do this manually (we use dbt as transformation tool). The way I did make tables like

city1_historical_occupation

but the bussiness area just need

historical_occupation