Auto-detectors operational.

tablib/core.py

@@ -277,6 +277,16 @@ class Databook(object):
 		return len(self._datasets)
 
 
+def detect(stream):
+	"""Return (format, stream) of given stream."""
+	for fmt in formats:
+		try:
+			if fmt.detect(stream):
+				return (fmt, stream) 
+		except AttributeError:
+			pass 
+	return (None, stream)
+		
 
 class InvalidDatasetType(Exception):
 	"Only Datasets can be added to a DataBook"

tablib/formats/__init__.py

@@ -8,4 +8,4 @@ import _json as json
 import _xls as xls
 import _yaml as yaml
 
-FORMATS = (csv, json, xls, yaml)
+FORMATS = (json, xls, yaml, csv)

tablib/formats/_csv.py

@@ -40,3 +40,12 @@ def import_set(dset, in_stream, headers=True):
 			dset.headers = row
 		else:
 			dset.append(row)
+
+
+def detect(stream):
+	"""Returns True if given stream is valid CSV."""
+	try:
+		rows = dialect = csv.Sniffer().sniff(stream)
+		return True
+	except csv.Error:
+		return False

tablib/formats/_json.py

@@ -36,3 +36,12 @@ def import_book(dbook, in_stream):
 		data.title = sheet['title']
 		data.dict = sheet['data']
 		dbook.add_sheet(data)
+
+
+def detect(stream):
+	"""Returns True if given stream is valid JSON."""
+	try:
+		json.loads(stream)
+		return True
+	except json.decoder.JSONDecodeError:
+		return False

tablib/formats/_yaml.py

@@ -39,4 +39,15 @@ def import_book(dbook, in_stream):
 		data = tablib.core.Dataset()
 		data.title = sheet['title']
 		data.dict = sheet['data']
-		dbook.add_sheet(data)
+		dbook.add_sheet(data)
+		
+def detect(stream):
+	"""Returns True if given stream is valid YAML."""
+	try:
+		_yaml = yaml.load(stream)
+		if isinstance(_yaml, (list, tuple, dict)):
+			return True
+		else:
+			return False
+	except yaml.parser.ParserError:
+		return False

test_tablib.py

@@ -269,9 +269,42 @@ class TablibTestCase(unittest.TestCase):
 		self.assertEqual(_csv, data.csv)
 
 	def test_csv_format_detect(self):
-		"""Test format detection."""
+		"""Test CSV format detection."""
 		
-		pass
+		_csv = (
+			'1,2,3\n'
+			'4,5,6\n'
+			'7,8,9\n'
+		)
+		_bunk = (
+			'¡¡¡¡¡¡¡¡£™∞¢£§∞§¶•¶ª∞¶•ªº••ª–º§•†•§º¶•†¥ª–º•§ƒø¥¨©πƒø†ˆ¥ç©¨√øˆ¥≈†ƒ¥ç©ø¨çˆ¥ƒçø¶'
+		)
+		
+		self.assertTrue(tablib.formats.csv.detect(_csv))
+		self.assertFalse(tablib.formats.csv.detect(_bunk))
+
+	def test_json_format_detect(self):
+		"""Test JSON format detection."""
+
+		_json = '[{"last_name": "Adams","age": 90,"first_name": "John"}]'
+		_bunk = (
+			'¡¡¡¡¡¡¡¡£™∞¢£§∞§¶•¶ª∞¶•ªº••ª–º§•†•§º¶•†¥ª–º•§ƒø¥¨©πƒø†ˆ¥ç©¨√øˆ¥≈†ƒ¥ç©ø¨çˆ¥ƒçø¶'
+		)
+
+		self.assertTrue(tablib.formats.json.detect(_json))
+		self.assertFalse(tablib.formats.json.detect(_bunk))
+
+
+	def test_yaml_format_detect(self):
+		"""Test YAML format detection."""
+
+		_yaml = '- {age: 90, first_name: John, last_name: Adams}'
+		_bunk = (
+			'¡¡¡¡¡¡---///\n\n\n¡¡£™∞¢£§∞§¶•¶ª∞¶•ªº••ª–º§•†•§º¶•†¥ª–º•§ƒø¥¨©πƒø†ˆ¥ç©¨√øˆ¥≈†ƒ¥ç©ø¨çˆ¥ƒçø¶'
+		)
+
+		self.assertTrue(tablib.formats.yaml.detect(_yaml))
+		self.assertFalse(tablib.formats.yaml.detect(_bunk))
 
 
 	def test_wipe(self):