def submit_export_request()

in infra-as-code/modules/export-to-bq-incremental/function-source-code/lib.py [0:0]

• 23 lines of code
• 1 McCabe index (conditional complexity)

    def submit_export_request(self, filter):
        """
        Submits an export request to BigQuery.

        Args:
            filter (str): The filter to apply to the export request.

        Returns:
            dict: The response from the export request.
        """
        headers = {
            'charset': 'utf-8',
            'Content-type': 'application/json',
            'Authorization': f'Bearer {self.get_token()}'
        }

        request_data = {
            'parent': f'projects/{self.ccai_insights_project_id}/locations/{self.ccai_insights_location_id}',
            'writeDisposition':'WRITE_TRUNCATE',
            'bigQueryDestination':{
                'projectId':self.bigquery_project_id,
                'dataset':self.bigquery_staging_dataset,
                'table':self.bigquery_staging_table
            },
            'filter':filter
        }
        print('BQ Export Request Data:')
        print(request_data)

        url = f'{self.insights_url_with_location}/insightsdata:export'

        response = requests.post(url, headers=headers, json=request_data)
        response.raise_for_status()

        response_json = response.json()
        
        return response_json