<s> from collections import namedtuple <EOL> import httplib <EOL> import json <EOL> import logging <EOL> import Queue <EOL> import random <EOL> import re <EOL> import socket <EOL> import textwrap <EOL> from thunderdome . exceptions import ThunderdomeException <EOL> from thunderdome . spec import Spec <EOL> logger = logging . getLogger ( __name__ ) <EOL> class ThunderdomeConnectionError ( ThunderdomeException ) : <EOL> """<STR_LIT>""" <EOL> class ThunderdomeQueryError ( ThunderdomeException ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , message , full_response = { } ) : <EOL> """<STR_LIT>""" <EOL> super ( ThunderdomeQueryError , self ) . __init__ ( message ) <EOL> self . _full_response = full_response <EOL> @ property <EOL> def raw_response ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _full_response <EOL> class ThunderdomeGraphMissingError ( ThunderdomeException ) : <EOL> """<STR_LIT>""" <EOL> Host = namedtuple ( '<STR_LIT>' , [ '<STR_LIT:name>' , '<STR_LIT:port>' ] ) <EOL> _hosts = [ ] <EOL> _host_idx = <NUM_LIT:0> <EOL> _graph_name = None <EOL> _username = None <EOL> _password = None <EOL> _index_all_fields = True <EOL> _existing_indices = None <EOL> _statsd = None <EOL> def create_key_index ( name ) : <EOL> """<STR_LIT>""" <EOL> global _existing_indices <EOL> _existing_indices = _existing_indices or execute_query ( '<STR_LIT>' ) <EOL> if name not in _existing_indices : <EOL> execute_query ( <EOL> "<STR_LIT>" , <EOL> { '<STR_LIT>' : name } , transaction = False ) <EOL> _existing_indices = None <EOL> def create_unique_index ( name , data_type ) : <EOL> """<STR_LIT>""" <EOL> global _existing_indices <EOL> _existing_indices = _existing_indices or execute_query ( '<STR_LIT>' ) <EOL> if name not in _existing_indices : <EOL> execute_query ( <EOL> "<STR_LIT>" . format ( data_type ) , <EOL> { '<STR_LIT:name>' : name } , transaction = False ) <EOL> _existing_indices = None <EOL> def setup ( hosts , graph_name , username = None , password = None , index_all_fields = False , statsd = None ) : <EOL> """<STR_LIT>""" <EOL> global _hosts <EOL> global _graph_name <EOL> global _username <EOL> global _password <EOL> global _index_all_fields <EOL> global _statsd <EOL> _graph_name = graph_name <EOL> _username = username <EOL> _password = password <EOL> _index_all_fields = index_all_fields <EOL> if statsd : <EOL> try : <EOL> sd = statsd <EOL> import statsd <EOL> tmp = sd . split ( '<STR_LIT::>' ) <EOL> if len ( tmp ) == <NUM_LIT:1> : <EOL> tmp . append ( '<STR_LIT>' ) <EOL> _statsd = statsd . StatsClient ( tmp [ <NUM_LIT:0> ] , int ( tmp [ <NUM_LIT:1> ] ) , prefix = '<STR_LIT>' ) <EOL> except ImportError : <EOL> logging . warning ( "<STR_LIT>" ) <EOL> except : <EOL> raise <EOL> for host in hosts : <EOL> host = host . strip ( ) <EOL> host = host . split ( '<STR_LIT::>' ) <EOL> if len ( host ) == <NUM_LIT:1> : <EOL> _hosts . append ( Host ( host [ <NUM_LIT:0> ] , <NUM_LIT> ) ) <EOL> elif len ( host ) == <NUM_LIT:2> : <EOL> _hosts . append ( Host ( * host ) ) <EOL> else : <EOL> raise ThunderdomeConnectionError ( "<STR_LIT>" . format ( '<STR_LIT>' . join ( host ) ) ) <EOL> if not _hosts : <EOL> raise ThunderdomeConnectionError ( "<STR_LIT>" ) <EOL> random . shuffle ( _hosts ) <EOL> create_unique_index ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> from thunderdome . models import vertex_types <EOL> for klass in vertex_types . values ( ) : <EOL> klass . _create_indices ( ) <EOL> def execute_query ( query , params = { } , transaction = True , context = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if transaction : <EOL> query = "<STR_LIT>" + query <EOL> if len ( _hosts ) <= <NUM_LIT:0> : <EOL> raise ThunderdomeConnectionError ( '<STR_LIT>' ) <EOL> host = _hosts [ <NUM_LIT:0> ] <EOL> data = json . dumps ( { '<STR_LIT>' : query , '<STR_LIT>' : params } ) <EOL> headers = { '<STR_LIT:Content-Type>' : '<STR_LIT:application/json>' , '<STR_LIT>' : '<STR_LIT:application/json>' , '<STR_LIT>' : '<STR_LIT:utf-8>' } <EOL> import time <EOL> try : <EOL> start_time = time . time ( ) <EOL> conn = httplib . HTTPConnection ( host . name , host . port ) <EOL> conn . request ( "<STR_LIT:POST>" , '<STR_LIT>' . format ( _graph_name ) , data , headers ) <EOL> response = conn . getresponse ( ) <EOL> content = response . read ( ) <EOL> total_time = int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) <EOL> if context and _statsd : <EOL> _statsd . timing ( "<STR_LIT>" . format ( context ) , total_time ) <EOL> _statsd . incr ( "<STR_LIT>" . format ( context ) ) <EOL> except socket . error as sock_err : <EOL> if _statsd : <EOL> total_time = int ( ( time . time ( ) - start_time ) * <NUM_LIT:1000> ) <EOL> _statsd . incr ( "<STR_LIT>" . format ( context ) , total_time ) <EOL> raise ThunderdomeQueryError ( '<STR_LIT>' . format ( sock_err ) ) <EOL> except : <EOL> raise <EOL> logger . info ( json . dumps ( data ) ) <EOL> logger . info ( content ) <EOL> try : <EOL> response_data = json . loads ( content ) <EOL> except ValueError as ve : <EOL> raise ThunderdomeQueryError ( '<STR_LIT>' . format ( ve ) ) <EOL> if response . status != <NUM_LIT:200> : <EOL> if '<STR_LIT:message>' in response_data and len ( response_data [ '<STR_LIT:message>' ] ) > <NUM_LIT:0> : <EOL> graph_missing_re = r"<STR_LIT>" <EOL> if re . search ( graph_missing_re , response_data [ '<STR_LIT:message>' ] ) : <EOL> raise ThunderdomeGraphMissingError ( response_data [ '<STR_LIT:message>' ] ) <EOL> else : <EOL> raise ThunderdomeQueryError ( <EOL> response_data [ '<STR_LIT:message>' ] , <EOL> response_data <EOL> ) <EOL> else : <EOL> if _statsd : <EOL> _statsd . incr ( "<STR_LIT>" . format ( context ) ) <EOL> raise ThunderdomeQueryError ( <EOL> response_data [ '<STR_LIT:error>' ] , <EOL> response_data <EOL> ) <EOL> return response_data [ '<STR_LIT>' ] <EOL> def sync_spec ( filename , host , graph_name , dry_run = False ) : <EOL> """<STR_LIT>""" <EOL> Spec ( filename ) . sync ( host , graph_name , dry_run = dry_run ) </s>
