openslides_backend/services/datastore/http_engine.py
def retrieve(
        self, endpoint: str, data: Optional[str]
    ) -> Tuple[Union[bytes, str], int]:
        """
        Throws 2 kinds of DatastoreConnectionException:
        1. If there is no valid endpoint given to build a URL
        2. If the datastore cannot be reached for unknown reasons
        Other exceptions from request.post are passed thru
        """
        # TODO: Check and test this error handling.
        if endpoint in self.READER_ENDPOINTS:
            base_url = self.datastore_reader_url
        elif endpoint in self.WRITER_ENDPOINTS:
            base_url = self.datastore_writer_url
        else:
            raise DatastoreConnectionException(f"Endpoint {endpoint} does not exist.")
        url = "/".join((base_url, endpoint))

        try:
            response = requests.post(url=url, data=data, headers=self.headers)
        except requests.exceptions.ConnectionError as e:
            error_message = f"Cannot reach the datastore service on {url}. Error: {e}"
            raise DatastoreConnectionException(error_message)
        return response.content, response.status_code

        Throws 2 kinds of DatastoreConnectionException:
        1. If there is no valid endpoint given to build a URL
        2. If the datastore cannot be reached for unknown reasons
        Other exceptions from request.post are passed thru
        
Throws 2 kinds of DatastoreConnectionException:
1. If there is no valid endpoint given to build a URL
2. If the datastore cannot be reached for unknown reasons
Other exceptions from request.post are passed thru
Throws 2 kinds of DatastoreConnectionException : 1 . If there is no valid endpoint given to build a URL 2 . If the datastore cannot be reached for unknown reasons Other exceptions from request . post are passed thru
def retrieve(
        self, endpoint: str, data: Optional[str]
    ) -> Tuple[Union[bytes, str], int]:
        if endpoint in self.READER_ENDPOINTS:
            base_url = self.datastore_reader_url
        elif endpoint in self.WRITER_ENDPOINTS:
            base_url = self.datastore_writer_url
        else:
            raise DatastoreConnectionException(f"Endpoint {endpoint} does not exist.")
        url = "/".join((base_url, endpoint))
        try:
            response = requests.post(url=url, data=data, headers=self.headers)
        except requests.exceptions.ConnectionError as e:
            error_message = f"Cannot reach the datastore service on {url}. Error: {e}"
            raise DatastoreConnectionException(error_message)
        return response.content, response.status_code
def retrieve ( self , endpoint : str , data : Optional [ str ] ) -> Tuple [ Union [ bytes , str ] , int ] : if endpoint in self . READER_ENDPOINTS : base_url = self . datastore_reader_url elif endpoint in self . WRITER_ENDPOINTS : base_url = self . datastore_writer_url else : raise DatastoreConnectionException ( f"Endpoint {endpoint} does not exist." ) url = "/" . join ( ( base_url , endpoint ) ) try : response = requests . post ( url = url , data = data , headers = self . headers ) except requests . exceptions . ConnectionError as e : error_message = f"Cannot reach the datastore service on {url}. Error: {e}" raise DatastoreConnectionException ( error_message ) return response . content , response . status_code
Throws 2 kinds of DatastoreConnectionException:
1.
Throws 2 kinds of DatastoreConnectionException : 1 .
"""
        Throws 2 kinds of DatastoreConnectionException:
        1. If there is no valid endpoint given to build a URL
        2. If the datastore cannot be reached for unknown reasons
        Other exceptions from request.post are passed thru
        """ # TODO: Check and test this error handling.
self
endpoint
str
data
Optional[str]