Changeset View

Standalone View

	Show All 38 Lines
	def jsonize(value):			def jsonize(value):
	"""Convert a value to a psycopg2 JSON object if necessary"""			"""Convert a value to a psycopg2 JSON object if necessary"""
	if isinstance(value, dict):			if isinstance(value, dict):
	return psycopg2.extras.Json(value)			return psycopg2.extras.Json(value)

	return value			return value


	def entry_to_bytes(entry):
	"""Convert an entry coming from the database to bytes"""
	if isinstance(entry, memoryview):
	return entry.tobytes()
	if isinstance(entry, list):
	return [entry_to_bytes(value) for value in entry]
	return entry


	def line_to_bytes(line):
	"""Convert a line coming from the database to bytes"""
	if not line:
	return line
	if isinstance(line, dict):
	return {k: entry_to_bytes(v) for k, v in line.items()}
	return line.__class__(entry_to_bytes(entry) for entry in line)


	def cursor_to_bytes(cursor):
	"""Yield all the data from a cursor as bytes"""
	yield from (line_to_bytes(line) for line in cursor)


	def execute_values_to_bytes(args, *kwargs):
	for line in execute_values_generator(args, *kwargs):
	yield line_to_bytes(line)


	def _paginate(seq, page_size):			def _paginate(seq, page_size):
	"""Consume an iterable and return it in chunks.			"""Consume an iterable and return it in chunks.
	Every chunk is at most `page_size`. Never return an empty chunk.			Every chunk is at most `page_size`. Never return an empty chunk.
	"""			"""
	page = []			page = []
	it = iter(seq)			it = iter(seq)
	while 1:			while 1:
	try:			try:
	▲ Show 20 Lines • Show All 95 Lines • Show Last 20 Lines