<s> import warnings <EOL> from datetime import datetime <EOL> import json <EOL> import logging <EOL> from time import sleep <EOL> import uuid <EOL> import time <EOL> import sys <EOL> import numpy as np <EOL> from distutils . version import StrictVersion <EOL> from pandas import compat <EOL> from pandas . core . api import DataFrame <EOL> from pandas . tools . merge import concat <EOL> from pandas . core . common import PandasError <EOL> from pandas . compat import lzip , bytes_to_str <EOL> def _check_google_client_version ( ) : <EOL> try : <EOL> import pkg_resources <EOL> except ImportError : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> if compat . PY3 : <EOL> google_api_minimum_version = '<STR_LIT>' <EOL> else : <EOL> google_api_minimum_version = '<STR_LIT>' <EOL> _GOOGLE_API_CLIENT_VERSION = pkg_resources . get_distribution ( <EOL> '<STR_LIT>' ) . version <EOL> if ( StrictVersion ( _GOOGLE_API_CLIENT_VERSION ) < <EOL> StrictVersion ( google_api_minimum_version ) ) : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> . format ( google_api_minimum_version , <EOL> _GOOGLE_API_CLIENT_VERSION ) ) <EOL> def _test_google_api_imports ( ) : <EOL> try : <EOL> import httplib2 <EOL> from apiclient . discovery import build <EOL> from apiclient . errors import HttpError <EOL> from oauth2client . client import AccessTokenRefreshError <EOL> from oauth2client . client import OAuth2WebServerFlow <EOL> from oauth2client . file import Storage <EOL> from oauth2client . tools import run_flow , argparser <EOL> except ImportError as e : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( str ( e ) ) ) <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> logger . setLevel ( logging . ERROR ) <EOL> class InvalidPrivateKeyFormat ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class AccessDenied ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class DatasetCreationError ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GenericGBQException ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InvalidColumnOrder ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InvalidPageToken ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class InvalidSchema ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class NotFoundException ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class StreamingInsertError ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> class TableCreationError ( PandasError , ValueError ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GbqConnector ( object ) : <EOL> scope = '<STR_LIT>' <EOL> def __init__ ( self , project_id , reauth = False , verbose = False , <EOL> private_key = None ) : <EOL> _check_google_client_version ( ) <EOL> _test_google_api_imports ( ) <EOL> self . project_id = project_id <EOL> self . reauth = reauth <EOL> self . verbose = verbose <EOL> self . private_key = private_key <EOL> self . credentials = self . get_credentials ( ) <EOL> self . service = self . get_service ( ) <EOL> def get_credentials ( self ) : <EOL> if self . private_key : <EOL> return self . get_service_account_credentials ( ) <EOL> else : <EOL> return self . get_user_account_credentials ( ) <EOL> def get_user_account_credentials ( self ) : <EOL> from oauth2client . client import OAuth2WebServerFlow <EOL> from oauth2client . file import Storage <EOL> from oauth2client . tools import run_flow , argparser <EOL> flow = OAuth2WebServerFlow ( <EOL> client_id = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> client_secret = '<STR_LIT>' , <EOL> scope = self . scope , <EOL> redirect_uri = '<STR_LIT>' ) <EOL> storage = Storage ( '<STR_LIT>' ) <EOL> credentials = storage . get ( ) <EOL> if credentials is None or credentials . invalid or self . reauth : <EOL> credentials = run_flow ( flow , storage , argparser . parse_args ( [ ] ) ) <EOL> return credentials <EOL> def get_service_account_credentials ( self ) : <EOL> oauth2client_v1 = True <EOL> oauth2client_v2 = True <EOL> try : <EOL> from oauth2client . client import SignedJwtAssertionCredentials <EOL> except ImportError : <EOL> oauth2client_v1 = False <EOL> try : <EOL> from oauth2client . service_account import ServiceAccountCredentials <EOL> except ImportError : <EOL> oauth2client_v2 = False <EOL> if not oauth2client_v1 and not oauth2client_v2 : <EOL> raise ImportError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> from os . path import isfile <EOL> try : <EOL> if isfile ( self . private_key ) : <EOL> with open ( self . private_key ) as f : <EOL> json_key = json . loads ( f . read ( ) ) <EOL> else : <EOL> json_key = json . loads ( self . private_key . replace ( '<STR_LIT:\n>' , '<STR_LIT:U+0020>' ) ) <EOL> json_key [ '<STR_LIT>' ] = json_key [ '<STR_LIT>' ] . replace ( <EOL> '<STR_LIT:U+0020>' , '<STR_LIT:\n>' ) <EOL> if compat . PY3 : <EOL> json_key [ '<STR_LIT>' ] = bytes ( <EOL> json_key [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> if oauth2client_v1 : <EOL> return SignedJwtAssertionCredentials ( <EOL> json_key [ '<STR_LIT>' ] , <EOL> json_key [ '<STR_LIT>' ] , <EOL> self . scope , <EOL> ) <EOL> else : <EOL> return ServiceAccountCredentials . from_json_keyfile_dict ( <EOL> json_key , <EOL> self . scope ) <EOL> except ( KeyError , ValueError , TypeError , AttributeError ) : <EOL> raise InvalidPrivateKeyFormat ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def _print ( self , msg , end = '<STR_LIT:\n>' ) : <EOL> if self . verbose : <EOL> sys . stdout . write ( msg + end ) <EOL> sys . stdout . flush ( ) <EOL> def _start_timer ( self ) : <EOL> self . start = time . time ( ) <EOL> def get_elapsed_seconds ( self ) : <EOL> return round ( time . time ( ) - self . start , <NUM_LIT:2> ) <EOL> def print_elapsed_seconds ( self , prefix = '<STR_LIT>' , postfix = '<STR_LIT>' , <EOL> overlong = <NUM_LIT:7> ) : <EOL> sec = self . get_elapsed_seconds ( ) <EOL> if sec > overlong : <EOL> self . _print ( '<STR_LIT>' . format ( prefix , sec , postfix ) ) <EOL> @ staticmethod <EOL> def sizeof_fmt ( num , suffix = '<STR_LIT:b>' ) : <EOL> fmt = "<STR_LIT>" <EOL> for unit in [ '<STR_LIT>' , '<STR_LIT:k>' , '<STR_LIT:M>' , '<STR_LIT>' , '<STR_LIT:T>' , '<STR_LIT:P>' , '<STR_LIT:E>' , '<STR_LIT>' ] : <EOL> if abs ( num ) < <NUM_LIT> : <EOL> return fmt % ( num , unit , suffix ) <EOL> num /= <NUM_LIT> <EOL> return fmt % ( num , '<STR_LIT:Y>' , suffix ) <EOL> def get_service ( self ) : <EOL> import httplib2 <EOL> from apiclient . discovery import build <EOL> http = httplib2 . Http ( ) <EOL> http = self . credentials . authorize ( http ) <EOL> bigquery_service = build ( '<STR_LIT>' , '<STR_LIT>' , http = http ) <EOL> return bigquery_service <EOL> @ staticmethod <EOL> def process_http_error ( ex ) : <EOL> status = json . loads ( bytes_to_str ( ex . content ) ) [ '<STR_LIT:error>' ] <EOL> errors = status . get ( '<STR_LIT>' , None ) <EOL> if errors : <EOL> for error in errors : <EOL> reason = error [ '<STR_LIT>' ] <EOL> message = error [ '<STR_LIT:message>' ] <EOL> raise GenericGBQException ( <EOL> "<STR_LIT>" . format ( reason , message ) ) <EOL> raise GenericGBQException ( errors ) <EOL> def process_insert_errors ( self , insert_errors ) : <EOL> for insert_error in insert_errors : <EOL> row = insert_error [ '<STR_LIT:index>' ] <EOL> errors = insert_error . get ( '<STR_LIT>' , None ) <EOL> for error in errors : <EOL> reason = error [ '<STR_LIT>' ] <EOL> message = error [ '<STR_LIT:message>' ] <EOL> location = error [ '<STR_LIT:location>' ] <EOL> error_message = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( row , reason , location , message ) ) <EOL> if self . verbose : <EOL> self . _print ( error_message ) <EOL> else : <EOL> raise StreamingInsertError ( error_message + <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise StreamingInsertError <EOL> def run_query ( self , query ) : <EOL> from apiclient . errors import HttpError <EOL> from oauth2client . client import AccessTokenRefreshError <EOL> _check_google_client_version ( ) <EOL> job_collection = self . service . jobs ( ) <EOL> job_data = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : query <EOL> } <EOL> } <EOL> } <EOL> self . _start_timer ( ) <EOL> try : <EOL> self . _print ( '<STR_LIT>' , end = "<STR_LIT>" ) <EOL> query_reply = job_collection . insert ( <EOL> projectId = self . project_id , body = job_data ) . execute ( ) <EOL> self . _print ( '<STR_LIT>' ) <EOL> except ( AccessTokenRefreshError , ValueError ) : <EOL> if self . private_key : <EOL> raise AccessDenied ( <EOL> "<STR_LIT>" ) <EOL> else : <EOL> raise AccessDenied ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> except HttpError as ex : <EOL> self . process_http_error ( ex ) <EOL> job_reference = query_reply [ '<STR_LIT>' ] <EOL> while not query_reply . get ( '<STR_LIT>' , False ) : <EOL> self . print_elapsed_seconds ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> query_reply = job_collection . getQueryResults ( <EOL> projectId = job_reference [ '<STR_LIT>' ] , <EOL> jobId = job_reference [ '<STR_LIT>' ] ) . execute ( ) <EOL> except HttpError as ex : <EOL> self . process_http_error ( ex ) <EOL> if self . verbose : <EOL> if query_reply [ '<STR_LIT>' ] : <EOL> self . _print ( '<STR_LIT>' ) <EOL> else : <EOL> bytes_processed = int ( query_reply . get ( <EOL> '<STR_LIT>' , '<STR_LIT:0>' ) ) <EOL> self . _print ( '<STR_LIT>' . format ( <EOL> self . sizeof_fmt ( bytes_processed ) ) ) <EOL> self . _print ( '<STR_LIT>' ) <EOL> total_rows = int ( query_reply [ '<STR_LIT>' ] ) <EOL> result_pages = list ( ) <EOL> seen_page_tokens = list ( ) <EOL> current_row = <NUM_LIT:0> <EOL> schema = query_reply [ '<STR_LIT>' ] <EOL> while '<STR_LIT>' in query_reply and current_row < total_rows : <EOL> page = query_reply [ '<STR_LIT>' ] <EOL> result_pages . append ( page ) <EOL> current_row += len ( page ) <EOL> self . print_elapsed_seconds ( <EOL> '<STR_LIT>' . format ( <EOL> len ( result_pages ) , <EOL> round ( <NUM_LIT> * current_row / total_rows ) ) ) <EOL> if current_row == total_rows : <EOL> break <EOL> page_token = query_reply . get ( '<STR_LIT>' , None ) <EOL> if not page_token and current_row < total_rows : <EOL> raise InvalidPageToken ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> . format ( current_row , total_rows ) ) <EOL> elif page_token in seen_page_tokens : <EOL> raise InvalidPageToken ( "<STR_LIT>" ) <EOL> seen_page_tokens . append ( page_token ) <EOL> try : <EOL> query_reply = job_collection . getQueryResults ( <EOL> projectId = job_reference [ '<STR_LIT>' ] , <EOL> jobId = job_reference [ '<STR_LIT>' ] , <EOL> pageToken = page_token ) . execute ( ) <EOL> except HttpError as ex : <EOL> self . process_http_error ( ex ) <EOL> if current_row < total_rows : <EOL> raise InvalidPageToken ( ) <EOL> self . _print ( '<STR_LIT>' . format ( total_rows ) ) <EOL> return schema , result_pages <EOL> def load_data ( self , dataframe , dataset_id , table_id , chunksize ) : <EOL> from apiclient . errors import HttpError <EOL> job_id = uuid . uuid4 ( ) . hex <EOL> rows = [ ] <EOL> remaining_rows = len ( dataframe ) <EOL> if self . verbose : <EOL> total_rows = remaining_rows <EOL> self . _print ( "<STR_LIT>" ) <EOL> for index , row in dataframe . reset_index ( drop = True ) . iterrows ( ) : <EOL> row_dict = dict ( ) <EOL> row_dict [ '<STR_LIT>' ] = json . loads ( row . to_json ( force_ascii = False , <EOL> date_unit = '<STR_LIT:s>' , <EOL> date_format = '<STR_LIT>' ) ) <EOL> row_dict [ '<STR_LIT>' ] = job_id + str ( index ) <EOL> rows . append ( row_dict ) <EOL> remaining_rows -= <NUM_LIT:1> <EOL> if ( len ( rows ) % chunksize == <NUM_LIT:0> ) or ( remaining_rows == <NUM_LIT:0> ) : <EOL> self . _print ( "<STR_LIT>" . format ( <EOL> ( ( total_rows - remaining_rows ) * <NUM_LIT:100> ) / total_rows ) ) <EOL> body = { '<STR_LIT>' : rows } <EOL> try : <EOL> response = self . service . tabledata ( ) . insertAll ( <EOL> projectId = self . project_id , <EOL> datasetId = dataset_id , <EOL> tableId = table_id , <EOL> body = body ) . execute ( ) <EOL> except HttpError as ex : <EOL> self . process_http_error ( ex ) <EOL> insert_errors = response . get ( '<STR_LIT>' , None ) <EOL> if insert_errors : <EOL> self . process_insert_errors ( insert_errors ) <EOL> sleep ( <NUM_LIT:1> ) <EOL> rows = [ ] <EOL> self . _print ( "<STR_LIT:\n>" ) <EOL> def verify_schema ( self , dataset_id , table_id , schema ) : <EOL> from apiclient . errors import HttpError <EOL> try : <EOL> return ( self . service . tables ( ) . get ( <EOL> projectId = self . project_id , <EOL> datasetId = dataset_id , <EOL> tableId = table_id <EOL> ) . execute ( ) [ '<STR_LIT>' ] ) == schema <EOL> except HttpError as ex : <EOL> self . process_http_error ( ex ) <EOL> def delete_and_recreate_table ( self , dataset_id , table_id , table_schema ) : <EOL> delay = <NUM_LIT:0> <EOL> if not self . verify_schema ( dataset_id , table_id , table_schema ) : <EOL> self . _print ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> delay = <NUM_LIT> <EOL> table = _Table ( self . project_id , dataset_id , <EOL> private_key = self . private_key ) <EOL> table . delete ( table_id ) <EOL> table . create ( table_id , table_schema ) <EOL> sleep ( delay ) <EOL> def _parse_data ( schema , rows ) : <EOL> dtype_map = { '<STR_LIT>' : np . dtype ( float ) , <EOL> '<STR_LIT>' : np . dtype ( float ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> fields = schema [ '<STR_LIT>' ] <EOL> col_types = [ field [ '<STR_LIT:type>' ] for field in fields ] <EOL> col_names = [ str ( field [ '<STR_LIT:name>' ] ) for field in fields ] <EOL> col_dtypes = [ dtype_map . get ( field [ '<STR_LIT:type>' ] , object ) for field in fields ] <EOL> page_array = np . zeros ( ( len ( rows ) , ) , <EOL> dtype = lzip ( col_names , col_dtypes ) ) <EOL> for row_num , raw_row in enumerate ( rows ) : <EOL> entries = raw_row . get ( '<STR_LIT:f>' , [ ] ) <EOL> for col_num , field_type in enumerate ( col_types ) : <EOL> field_value = _parse_entry ( entries [ col_num ] . get ( '<STR_LIT:v>' , '<STR_LIT>' ) , <EOL> field_type ) <EOL> page_array [ row_num ] [ col_num ] = field_value <EOL> return DataFrame ( page_array , columns = col_names ) <EOL> def _parse_entry ( field_value , field_type ) : <EOL> if field_value is None or field_value == '<STR_LIT:null>' : <EOL> return None <EOL> if field_type == '<STR_LIT>' or field_type == '<STR_LIT>' : <EOL> return float ( field_value ) <EOL> elif field_type == '<STR_LIT>' : <EOL> timestamp = datetime . utcfromtimestamp ( float ( field_value ) ) <EOL> return np . datetime64 ( timestamp ) <EOL> elif field_type == '<STR_LIT>' : <EOL> return field_value == '<STR_LIT:true>' <EOL> return field_value <EOL> def read_gbq ( query , project_id = None , index_col = None , col_order = None , <EOL> reauth = False , verbose = True , private_key = None ) : <EOL> """<STR_LIT>""" <EOL> if not project_id : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> connector = GbqConnector ( project_id , reauth = reauth , verbose = verbose , <EOL> private_key = private_key ) <EOL> schema , pages = connector . run_query ( query ) <EOL> dataframe_list = [ ] <EOL> while len ( pages ) > <NUM_LIT:0> : <EOL> page = pages . pop ( ) <EOL> dataframe_list . append ( _parse_data ( schema , page ) ) <EOL> if len ( dataframe_list ) > <NUM_LIT:0> : <EOL> final_df = concat ( dataframe_list , ignore_index = True ) <EOL> else : <EOL> final_df = _parse_data ( schema , [ ] ) <EOL> if index_col is not None : <EOL> if index_col in final_df . columns : <EOL> final_df . set_index ( index_col , inplace = True ) <EOL> else : <EOL> raise InvalidColumnOrder ( <EOL> '<STR_LIT>' <EOL> . format ( index_col ) <EOL> ) <EOL> if col_order is not None : <EOL> if sorted ( col_order ) == sorted ( final_df . columns ) : <EOL> final_df = final_df [ col_order ] <EOL> else : <EOL> raise InvalidColumnOrder ( <EOL> '<STR_LIT>' <EOL> ) <EOL> final_df . _data = final_df . _data . downcast ( dtypes = '<STR_LIT>' ) <EOL> connector . print_elapsed_seconds ( <EOL> '<STR_LIT>' , <EOL> datetime . now ( ) . strftime ( '<STR_LIT>' ) , <EOL> <NUM_LIT:0> <EOL> ) <EOL> return final_df <EOL> def to_gbq ( dataframe , destination_table , project_id , chunksize = <NUM_LIT> , <EOL> verbose = True , reauth = False , if_exists = '<STR_LIT>' , private_key = None ) : <EOL> """<STR_LIT>""" <EOL> if if_exists not in ( '<STR_LIT>' , '<STR_LIT:replace>' , '<STR_LIT>' ) : <EOL> raise ValueError ( "<STR_LIT>" . format ( if_exists ) ) <EOL> if '<STR_LIT:.>' not in destination_table : <EOL> raise NotFoundException ( <EOL> "<STR_LIT>" ) <EOL> connector = GbqConnector ( project_id , reauth = reauth , verbose = verbose , <EOL> private_key = private_key ) <EOL> dataset_id , table_id = destination_table . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> table = _Table ( project_id , dataset_id , reauth = reauth , <EOL> private_key = private_key ) <EOL> table_schema = _generate_bq_schema ( dataframe ) <EOL> if table . exists ( table_id ) : <EOL> if if_exists == '<STR_LIT>' : <EOL> raise TableCreationError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> elif if_exists == '<STR_LIT:replace>' : <EOL> connector . delete_and_recreate_table ( <EOL> dataset_id , table_id , table_schema ) <EOL> elif if_exists == '<STR_LIT>' : <EOL> if not connector . verify_schema ( dataset_id , table_id , table_schema ) : <EOL> raise InvalidSchema ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> table . create ( table_id , table_schema ) <EOL> connector . load_data ( dataframe , dataset_id , table_id , chunksize ) <EOL> def generate_bq_schema ( df , default_type = '<STR_LIT>' ) : <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , FutureWarning , stacklevel = <NUM_LIT:2> ) <EOL> return _generate_bq_schema ( df , default_type = default_type ) <EOL> def _generate_bq_schema ( df , default_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> type_mapping = { <EOL> '<STR_LIT:i>' : '<STR_LIT>' , <EOL> '<STR_LIT:b>' : '<STR_LIT>' , <EOL> '<STR_LIT:f>' : '<STR_LIT>' , <EOL> '<STR_LIT:O>' : '<STR_LIT>' , <EOL> '<STR_LIT:S>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:M>' : '<STR_LIT>' <EOL> } <EOL> fields = [ ] <EOL> for column_name , dtype in df . dtypes . iteritems ( ) : <EOL> fields . append ( { '<STR_LIT:name>' : column_name , <EOL> '<STR_LIT:type>' : type_mapping . get ( dtype . kind , default_type ) } ) <EOL> return { '<STR_LIT>' : fields } <EOL> class _Table ( GbqConnector ) : <EOL> def __init__ ( self , project_id , dataset_id , reauth = False , verbose = False , <EOL> private_key = None ) : <EOL> from apiclient . errors import HttpError <EOL> self . http_error = HttpError <EOL> self . dataset_id = dataset_id <EOL> super ( _Table , self ) . __init__ ( project_id , reauth , verbose , private_key ) <EOL> def exists ( self , table_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . service . tables ( ) . get ( <EOL> projectId = self . project_id , <EOL> datasetId = self . dataset_id , <EOL> tableId = table_id ) . execute ( ) <EOL> return True <EOL> except self . http_error as ex : <EOL> if ex . resp . status == <NUM_LIT> : <EOL> return False <EOL> else : <EOL> self . process_http_error ( ex ) <EOL> def create ( self , table_id , schema ) : <EOL> """<STR_LIT>""" <EOL> if self . exists ( table_id ) : <EOL> raise TableCreationError ( <EOL> "<STR_LIT>" ) <EOL> if not _Dataset ( self . project_id , <EOL> private_key = self . private_key ) . exists ( self . dataset_id ) : <EOL> _Dataset ( self . project_id , <EOL> private_key = self . private_key ) . create ( self . dataset_id ) <EOL> body = { <EOL> '<STR_LIT>' : schema , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : table_id , <EOL> '<STR_LIT>' : self . project_id , <EOL> '<STR_LIT>' : self . dataset_id <EOL> } <EOL> } <EOL> try : <EOL> self . service . tables ( ) . insert ( <EOL> projectId = self . project_id , <EOL> datasetId = self . dataset_id , <EOL> body = body ) . execute ( ) <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) <EOL> def delete ( self , table_id ) : <EOL> """<STR_LIT>""" <EOL> if not self . exists ( table_id ) : <EOL> raise NotFoundException ( "<STR_LIT>" ) <EOL> try : <EOL> self . service . tables ( ) . delete ( <EOL> datasetId = self . dataset_id , <EOL> projectId = self . project_id , <EOL> tableId = table_id ) . execute ( ) <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) <EOL> class _Dataset ( GbqConnector ) : <EOL> def __init__ ( self , project_id , reauth = False , verbose = False , <EOL> private_key = None ) : <EOL> from apiclient . errors import HttpError <EOL> self . http_error = HttpError <EOL> super ( _Dataset , self ) . __init__ ( project_id , reauth , verbose , <EOL> private_key ) <EOL> def exists ( self , dataset_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . service . datasets ( ) . get ( <EOL> projectId = self . project_id , <EOL> datasetId = dataset_id ) . execute ( ) <EOL> return True <EOL> except self . http_error as ex : <EOL> if ex . resp . status == <NUM_LIT> : <EOL> return False <EOL> else : <EOL> self . process_http_error ( ex ) <EOL> def datasets ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> list_dataset_response = self . service . datasets ( ) . list ( <EOL> projectId = self . project_id ) . execute ( ) . get ( '<STR_LIT>' , None ) <EOL> if not list_dataset_response : <EOL> return [ ] <EOL> dataset_list = list ( ) <EOL> for row_num , raw_row in enumerate ( list_dataset_response ) : <EOL> dataset_list . append ( raw_row [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> return dataset_list <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) <EOL> def create ( self , dataset_id ) : <EOL> """<STR_LIT>""" <EOL> if self . exists ( dataset_id ) : <EOL> raise DatasetCreationError ( <EOL> "<STR_LIT>" ) <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . project_id , <EOL> '<STR_LIT>' : dataset_id <EOL> } <EOL> } <EOL> try : <EOL> self . service . datasets ( ) . insert ( <EOL> projectId = self . project_id , <EOL> body = body ) . execute ( ) <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) <EOL> def delete ( self , dataset_id ) : <EOL> """<STR_LIT>""" <EOL> if not self . exists ( dataset_id ) : <EOL> raise NotFoundException ( <EOL> "<STR_LIT>" . format ( dataset_id ) ) <EOL> try : <EOL> self . service . datasets ( ) . delete ( <EOL> datasetId = dataset_id , <EOL> projectId = self . project_id ) . execute ( ) <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) <EOL> def tables ( self , dataset_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> list_table_response = self . service . tables ( ) . list ( <EOL> projectId = self . project_id , <EOL> datasetId = dataset_id ) . execute ( ) . get ( '<STR_LIT>' , None ) <EOL> if not list_table_response : <EOL> return [ ] <EOL> table_list = list ( ) <EOL> for row_num , raw_row in enumerate ( list_table_response ) : <EOL> table_list . append ( raw_row [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> return table_list <EOL> except self . http_error as ex : <EOL> self . process_http_error ( ex ) </s>
