<commit_before>#-------------------------------------------------------
# Asyn PostgresSQL Dispatcher
# <NAME> (<EMAIL>)
# 2015.6.9
#-------------------------------------------------------
		
DEBUG = False

try:
	import psycopg2
	
except ImportError:	
	class AsynConnect:
		def __init__ (self, address, params = None, lock = None, logger = None):
			logger ("[warn] cannot import psycopg2")
			raise ImportError ("cannot import psycopg2")

else:	
	import asyncore
	from . import dbconnect

	from psycopg2.extensions import POLL_OK, POLL_WRITE, POLL_READ
	_STATE_OK = (POLL_OK, POLL_WRITE, POLL_READ)
		
	class AsynConnect (dbconnect.AsynDBConnect, asyncore.dispatcher):
		def __init__ (self, address, params = None, lock = None, logger = None):
			dbconnect.AsynDBConnect.__init__ (self, address, params, lock, logger)			
			self.conn = None
			self.cur = None			
			asyncore.dispatcher.__init__ (self)
			
		def check_state (self, state):
			if state not in (_STATE_OK):				
				self.logger ("[warn] psycopg2.poll() returned %s" % state)
				self.handle_close ()
		
		def poll (self):
			# 2 cases, if on requesting raise immediatly, else handle silently
			try:
				return self.socket.poll ()
			except psycopg2.OperationalError:				
				if self.request:
					return self.handle_error ()					
				# else usually timeout	
			except:
				if self.request:
					return self.handle_error ()
				# logging
				elf.logger.trace ()
			return -1
			
		def writable (self):			
			return self.out_buffer or not self.connected
			
		def readable (self):
			return self.connected and not self.out_buffer
		
		def add_channel (self, map = None):
			return asyncore.dispatcher.add_channel (self, map)
				
		def del_channel (self, map=None):
			fd = self._fileno
			if map is None:
				map = self._map
			if fd in map:
				del map[fd]
		
		def handle_expt_event (self):
			self.handle_expt ()
			
		def handle_connect_event (self):
			state = self.poll ()			
			if state == POLL_OK:	
				self.handle_connect ()
				self.connected = True
				self.connecting = False		
			else:				
				self.check_state (state)
		
		def handle_write_event (self):		
			if not self.connected:
				self.handle_connect_event ()
			else:	
				self.handle_write ()
		
		def handle_expt (self):
			self.handle_close (psycopg2.OperationalError, "socket panic")
			
		def handle_connect (self):
			self.del_channel ()
			self.conn = self.socket
			self.cur = self.conn.cursor()		
			self.set_socket (self.cur.connection)			
					
		def handle_read (self):
			state = self.poll ()
			if self.cur and state == POLL_OK:
				self.set_event_time ()
				self.has_result = True
				self.close_case_with_end_tran ()
			else:
				self.check_state (state)
				
		def handle_write (self):
			state = self.poll ()
			if self.cur and state == POLL_OK:
				self.set_event_time ()				
				self.cur.execute (*self.out_buffer)
				self.out_buffer = ()
			else:				
				self.check_state (state)
				
		#-----------------------------------
		# Overriden
		#-----------------------------------
		def close_case (self):			
			if self.request:
				if self.has_result and self.cur.description:					
					self.request.handle_result (self.cur.description, self.exception_class, self.exception_str, self.fetchall ())					
				else:
					self.request.handle_result (None, self.exception_class, self.exception_str, None)
					self.has_result = False
				self.request = None
			self.set_active (False)
			
		def empty_cursor (self):
			if self.has_result:
				try:
					self.fetchall ()
				except psycopg2.ProgrammingError:
					pass				
		
		def fetchall (self):
			result = self.cur.fetchall ()
			self.has_result = False
			return result
							
		def close (self, deactive = 1):
			if self.cur:
				try:
					self.cur.close ()
				except psycopg2.ProgrammingError:
					pass						
				self.cur = None
			if self.conn:	
				self.conn.close ()
				self.conn = None	
				
			dbconnect.AsynDBConnect.close (self, deactive)
			asyncore.dispatcher.close (self)
			
		def connect (self, force = 0):
			host, port = self.address		
			sock = psycopg2.connect (
				dbname = self.dbname,
				user = self.user,
				password = <PASSWORD>,
				host = host,
				port = port,
				async = 1
			)
			self.set_socket (sock)
		
		def end_tran (self):
			self.del_channel ()

		def begin_tran (self, request):
			dbconnect.AsynDBConnect.begin_tran (self, request)
			self.out_buffer = request.params
								
		def execute (self, request):			
			self.begin_tran (request)			
			if not self.connected:
				self.connect ()				
			else:
				state = self.poll ()
				if state != POLL_OK:
					self.reconnect ()
				else:
					self.add_channel ()<commit_msg>fix psycopg2.poll()<commit_after>#-------------------------------------------------------
# Asyn PostgresSQL Dispatcher
# <NAME> (<EMAIL>)
# 2015.6.9
#-------------------------------------------------------
		
DEBUG = False

try:
	import psycopg2
	
except ImportError:	
	class AsynConnect:
		def __init__ (self, address, params = None, lock = None, logger = None):
			logger ("[warn] cannot import psycopg2")
			raise ImportError ("cannot import psycopg2")

else:	
	import asyncore
	from . import dbconnect

	from psycopg2.extensions import POLL_OK, POLL_WRITE, POLL_READ
	_STATE_OK = (POLL_OK, POLL_WRITE, POLL_READ)
		
	class AsynConnect (dbconnect.AsynDBConnect, asyncore.dispatcher):
		def __init__ (self, address, params = None, lock = None, logger = None):
			dbconnect.AsynDBConnect.__init__ (self, address, params, lock, logger)			
			self.conn = None
			self.cur = None			
			asyncore.dispatcher.__init__ (self)
			
		def check_state (self, state):
			if state not in (_STATE_OK):				
				self.logger ("[warn] psycopg2.poll() returned %s" % state)
				self.handle_close ()
		
		def poll (self):
			# 2 cases, if on requesting raise immediatly, else handle silently
			try:
				return self.socket.poll ()
			except psycopg2.OperationalError:				
				if self.request:
					raise
				# else usually timeout	
			except:
				if self.request:
					raise
				# logging
				self.logger.trace ()
			return -1
			
		def writable (self):			
			return self.out_buffer or not self.connected
			
		def readable (self):
			return self.connected and not self.out_buffer
		
		def add_channel (self, map = None):
			return asyncore.dispatcher.add_channel (self, map)
				
		def del_channel (self, map=None):
			fd = self._fileno
			if map is None:
				map = self._map
			if fd in map:
				del map[fd]
		
		def handle_expt_event (self):
			self.handle_expt ()
			
		def handle_connect_event (self):
			state = self.poll ()			
			if state == POLL_OK:	
				self.handle_connect ()
				self.connected = True
				self.connecting = False		
			else:				
				self.check_state (state)
		
		def handle_write_event (self):		
			if not self.connected:
				self.handle_connect_event ()
			else:	
				self.handle_write ()
		
		def handle_expt (self):
			self.handle_close (psycopg2.OperationalError, "socket panic")
			
		def handle_connect (self):
			self.del_channel ()
			self.conn = self.socket
			self.cur = self.conn.cursor()		
			self.set_socket (self.cur.connection)			
					
		def handle_read (self):
			state = self.poll ()
			if self.cur and state == POLL_OK:
				self.set_event_time ()
				self.has_result = True
				self.close_case_with_end_tran ()
			else:
				self.check_state (state)
				
		def handle_write (self):
			state = self.poll ()
			if self.cur and state == POLL_OK:
				self.set_event_time ()				
				self.cur.execute (self.out_buffer)
				self.out_buffer = ""
			else:				
				self.check_state (state)
				
		#-----------------------------------
		# Overriden
		#-----------------------------------
		def close_case (self):			
			if self.request:
				if self.has_result and self.cur.description:					
					self.request.handle_result (self.cur.description, self.exception_class, self.exception_str, self.fetchall ())					
				else:
					self.request.handle_result (None, self.exception_class, self.exception_str, None)
					self.has_result = False
				self.request = None
			self.set_active (False)
			
		def empty_cursor (self):
			if self.has_result:
				try:
					self.fetchall ()
				except psycopg2.ProgrammingError:
					pass				
		
		def fetchall (self):
			result = self.cur.fetchall ()
			self.has_result = False
			return result
							
		def close (self, deactive = 1):
			if self.cur:
				try:
					self.cur.close ()
				except psycopg2.ProgrammingError:
					pass						
				self.cur = None
			if self.conn:	
				self.conn.close ()
				self.conn = None	
				
			dbconnect.AsynDBConnect.close (self, deactive)
			asyncore.dispatcher.close (self)
			
		def connect (self, force = 0):
			host, port = self.address		
			sock = psycopg2.connect (
				dbname = self.dbname,
				user = self.user,
				password = <PASSWORD>,
				host = host,
				port = port,
				async = 1
			)
			self.set_socket (sock)
		
		def end_tran (self):
			self.del_channel ()

		def begin_tran (self, request):
			dbconnect.AsynDBConnect.begin_tran (self, request)
			self.out_buffer = request.params [0]
								
		def execute (self, request):			
			self.begin_tran (request)			
			if not self.connected:
				self.connect ()				
			else:
				state = self.poll ()
				if state != POLL_OK:
					self.reconnect ()
				else:
					self.add_channel ()
{'fraction_non_alphanumeric': 0.13743413270244112, 'fraction_numerical': 0.004731691579739757, 'mean_word_length': 2.3107867568529725, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2202972', 'n_tokens_mistral': 3958, 'n_tokens_neox': 3013, 'n_words': 758}