def to_url()

in azure/durable_functions/models/RpcManagementOptions.py [0:0]

• 22 lines of code
• 7 McCabe index (conditional complexity)

    def to_url(self, base_url: Optional[str]) -> str:
        """Get the url based on the options selected.

        Parameters
        ----------
        base_url: str
            The base url to prepend to the url path

        Raises
        ------
        ValueError
            When the `base_url` argument is None

        Returns
        -------
        str
            The Url used to get orchestration status information
        """
        if base_url is None:
            raise ValueError("orchestration bindings has not RPC base url")

        if self.entity_Id:
            url = f'{base_url}{EntityId.get_entity_id_url_path(self.entity_Id)}'
        else:
            url = f"{base_url}instances/{self._instance_id if self._instance_id else ''}"

        query: List[str] = []

        self._add_arg(query, 'taskHub', self._task_hub_name)
        self._add_arg(query, 'connectionName', self._connection_name)
        self._add_arg(query, 'showInput', self._show_input)
        self._add_arg(query, 'showHistory', self._show_history)
        self._add_arg(query, 'showHistoryOutput', self._show_history_output)
        self._add_date_arg(query, 'createdTimeFrom', self._created_time_from)
        self._add_date_arg(query, 'createdTimeTo', self._created_time_to)
        self._add_arg(query, 'op', self.operation_name)
        if self._runtime_status is not None and len(self._runtime_status) > 0:
            runtime_status = ",".join(r.value for r in self._runtime_status)
            self._add_arg(query, 'runtimeStatus', runtime_status)

        if len(query) > 0:
            url += "?" + "&".join(query)

        return url