Added some validation functions

Author: dhondta

README.md

@@ -36,9 +36,34 @@ pip install --user dsff
 ```python
 >>> import dsff
 >>> with dsff.DSFF("/path/to/my-dataset.dsff") as f:
-    f.to_dataset()  # creates ./[dsff-title] with data.csv, features.json and metadata.json
+    f.to_dataset()  # creates ./my-dataset with data.csv, features.json and metadata.json
 ```
 
+**
+
+## :heavy_plus_sign: Extensions
+
+**Install all available extensions**
+
+```sh
+pip install --user dsff[all]
+```
+
+**Dealing with [Apache Arrow](https://arrow.apache.org/) formats**
+
+```sh
+pip install --user dsff[arrow]
+```
+
+```python
+>>> import dsff
+>>> with dsff.DSFF("/path/to/my-dataset.dsff") as f:
+    f.to_feather()  # creates ./my-dataset.feather
+    f.to_orc()      # creates ./my-dataset.orc
+    f.to_parquet()  # creates ./my-dataset.parquet
+```
+
+
 ## :star: Related Projects
 
 You may also like these:

pyproject.toml

@@ -34,9 +34,8 @@ dependencies = [
 dynamic = ["version"]
 
 [project.optional-dependencies]
-extra = [
-  "pyarrow",
-]
+all = ["dsff[arrow]"]
+arrow = ["pyarrow"]
 
 [project.readme]
 file = "README.md"

src/dsff/VERSION.txt

@@ -1 +1 @@
-1.2.0
+1.2.1

src/dsff/formats/__common__.py

@@ -5,7 +5,7 @@
 from ast import literal_eval
 from io import BytesIO, StringIO
 from os import makedirs, remove
-from os.path import basename, expanduser, isfile, isdir, join, splitext
+from os.path import basename, exists, expanduser, isfile, isdir, join, splitext
 try:  # pragma: no cover
     import pyarrow
     import pyarrow.feather as feather
@@ -17,11 +17,11 @@
     PYARROW = False
 
 
-__all__ = ["basename", "csvmod", "expanduser", "isfile", "isdir", "join", "json", "literal_eval", "makedirs",
-           "remove", "splitext", "re", "BytesIO", "StringIO",
+__all__ = ["basename", "csvmod", "ensure_str", "exists", "expanduser", "isfile", "isdir", "join", "json",
+           "literal_eval", "makedirs", "re", "remove", "splitext", "text_or_path", "BytesIO", "StringIO",
            "CSV_DELIMITER", "DEFAULT_EXCL", "INMEMORY", "META_EXCL", "MISSING_TOKEN", "PYARROW", "TARGET_NAME"]
 if PYARROW:
-    __all__ += ["feather", "orc", "pandas", "pyarrow", "parquet"]
+    __all__ += ["feather", "orc", "pandas", "parquet", "pyarrow"]
 
 
 CSV_DELIMITER = ";"
@@ -31,3 +31,26 @@
 MISSING_TOKEN = "?"
 TARGET_NAME   = "label"
 
+
+def ensure_str(string):
+    try:
+        return string.decode()
+    except AttributeError:
+        return str(string)
+
+
+def text_or_path(f):
+    def _wrapper(txt_path, *args, **kwargs):
+        if not isinstance(txt_path, bytes):
+            if isinstance(txt_path, str):
+                if not re.search(r"[<>:\'\"/\\|?*\x00-\x1f]", txt_path, re.I) and exists(txt_path):
+                    if not isdir(txt_path):
+                        with open(txt_path, "rb") as fh:
+                            txt_path = fh.read()
+                else:
+                    txt_path = txt_path.encode()
+            else:
+                raise ValueError(f"Bad input text or path ({type(txt_path)})")
+        return f(txt_path, *args, **kwargs)
+    return _wrapper
+

src/dsff/formats/__init__.py

@@ -22,6 +22,9 @@
 
 
 __all__ = ["DSFF"]
+for k in list(globals().keys()):
+    if k.startswith("is_"):
+        __all__.append(k)
 
 _FORMAT_TEXT_ALIAS = {'arff': "ARFF", 'csv': "CSV", 'db': "SQL", 'orc': "ORC"}
 _NO_EXT_FORMAT     = ["dataset"]
@@ -343,4 +346,16 @@ def path(self):
         if p != INMEMORY and not p.endswith(".dsff"):
             p += ".dsff"
         return p
+    
+    @staticmethod
+    @text_or_path
+    def format(text):
+        for k, func in globals().items():
+            if k.startswith("is_") and func(text):
+                return k.split("_", 1)[1]
+    
+    @staticmethod
+    @text_or_path
+    def is_valid_format(text):
+        return DSFF.format(text) is not None
 

src/dsff/formats/arff.py

@@ -2,68 +2,66 @@
 from .__common__ import *
 
 
-__all__ = ["from_arff", "to_arff"]
+__all__ = ["from_arff", "is_arff", "to_arff"]
 
 
-def from_arff(dsff, path=None, target=TARGET_NAME, missing=MISSING_TOKEN):
-    """ Populate the DSFF file from an ARFF file. """
-    d, features = [], {}
-    with open(path) as f:
-        relation, attributes, data = False, [False, False], False
-        for n, l in enumerate(f, 1):
-            l, pf = l.strip(), f"Line {n}: "
-            # the file shall start with "@RELATION"
-            if not relation:
-                if l.startswith("@RELATION "):
-                    relation = True
-                    try:
-                        dsff['title'] = re.match(r"@RELATION\s+('[^']*'|\"[^\"]*\")$", l).group(1).strip("'\"")
-                        continue
-                    except Exception as e:
-                        raise BadInputData(f"{pf}failed on @RELATION ({e})")
-                else:
-                    raise BadInputData(f"{pf}did not find @RELATION")
-            # get metadata and feature descriptions from comments
-            if l.startswith("%"):
-                if re.match(r"^\%\s+metadata\s*\:\s*\{.*\}$", l):
-                    dsff.write(metadata=literal_eval(l.split(":", 1)[1]))
-                elif (m := re.match(r"^\%\s+(.*?)\s*\:\s*(.*?)$", l)):
-                    name, descr = m.groups()
-                    features[name] = descr
-                continue
-            # then ignore blank lines
-            if l == "":
-                if attributes[0] and not attributes[1]:
-                    # close the atributes block
-                    attributes[1] = True
-                    n_cols = len(d[0])
-                continue
-            if l.startswith("@ATTRIBUTE "):
-                if not attributes[0]:
-                    attributes[0] = True
-                if len(d) == 0:
-                    # start the attributes block
-                    d.append([])
-                if attributes[1]:
-                    raise BadInputData(f"{pf}found @ATTRIBUTE out of the attributes block)")
+def _parse(text_or_fh, target=TARGET_NAME, missing=MISSING_TOKEN):
+    d, features, metadata, title = [], {}, {}, ""
+    relation, attributes, data = False, [False, False], False
+    for n, l in enumerate(t.splitlines() if isinstance(t := text_or_fh, str) else t):
+        l, pf = l.strip(), f"Line {n}: "
+        # the file shall start with "@RELATION"
+        if not relation:
+            if l.startswith("@RELATION "):
+                relation = True
                 try:
-                    header = re.match(r"@ATTRIBUTE\s+([^\s]+)\s+[A-Z]+$", l).group(1)
-                    if header == "class":
-                        header = target
-                    d[0].append(header)
-                    continue
-                except AttributeError:
-                    raise BadInputData(f"{pf}failed on @ATTRIBUTE (bad type)")
-            if not data:
-                if l == "@DATA":
-                    data = True
+                    title = re.match(r"@RELATION\s+('[^']*'|\"[^\"]*\")$", l).group(1).strip("'\"")
                     continue
-                else:
-                    raise BadInputData(f"{pf}did not find @DATA where expected")
-            row = list(map(lambda x: x.strip("'\""), re.split(r",\s+", l)))
-            if len(row) != n_cols:
-                raise BadInputData(f"{pf}this row does not match the number of columns")
-            d.append(row)
+                except Exception as e:
+                    raise BadInputData(f"{pf}failed on @RELATION ({e})")
+            else:
+                raise BadInputData(f"{pf}did not find @RELATION")
+        # get metadata and feature descriptions from comments
+        if l.startswith("%"):
+            if re.match(r"^\%\s+metadata\s*\:\s*\{.*\}$", l):
+                metadata = literal_eval(l.split(":", 1)[1])
+            elif (m := re.match(r"^\%\s+(.*?)\s*\:\s*(.*?)$", l)):
+                name, descr = m.groups()
+                features[name] = descr
+            continue
+        # then ignore blank lines
+        if l == "":
+            if attributes[0] and not attributes[1]:
+                # close the atributes block
+                attributes[1] = True
+                n_cols = len(d[0])
+            continue
+        if l.startswith("@ATTRIBUTE "):
+            if not attributes[0]:
+                attributes[0] = True
+            if len(d) == 0:
+                # start the attributes block
+                d.append([])
+            if attributes[1]:
+                raise BadInputData(f"{pf}found @ATTRIBUTE out of the attributes block)")
+            try:
+                header = re.match(r"@ATTRIBUTE\s+([^\s]+)\s+[A-Z]+$", l).group(1)
+                if header == "class":
+                    header = target
+                d[0].append(header)
+                continue
+            except AttributeError:
+                raise BadInputData(f"{pf}failed on @ATTRIBUTE (bad type)")
+        if not data:
+            if l == "@DATA":
+                data = True
+                continue
+            else:
+                raise BadInputData(f"{pf}did not find @DATA where expected")
+        row = list(map(lambda x: x.strip("'\""), re.split(r",\s+", l)))
+        if len(row) != n_cols:
+            raise BadInputData(f"{pf}this row does not match the number of columns")
+        d.append(row)
     for i in range(n_cols):
         values = []
         for j, row in enumerate(d):
@@ -76,9 +74,26 @@ def from_arff(dsff, path=None, target=TARGET_NAME, missing=MISSING_TOKEN):
         if "".join(sorted(set(values))) in ["0", "1", "01"]:
             for j, row in enumerate(d):
                 if j > 0:
-                    row[i] = {'0': "False", '1': "True"}[row[i]]
-    dsff.write(data=d)
-    dsff.write(features=features)
+                    row[i] = {'0': "False", '1': "True"}[row[i]]    
+    return d, features, metadata, title
+
+
+def from_arff(dsff, path=None, target=TARGET_NAME, missing=MISSING_TOKEN):
+    """ Populate the DSFF file from an ARFF file. """
+    with open(path) as f:
+        d, ft, md, t = _parse(f, target, missing)
+    dsff['title'] = t
+    dsff.write(data=d, features=ft, metadata=md)
+
+
+@text_or_path
+def is_arff(text):
+    """ Check if the input text or path is a valid ARFF. """
+    try:
+        _parse(ensure_str(text))
+        return True
+    except (BadInputData, UnicodeDecodeError):
+        return False
 
 
 def to_arff(dsff, path=None, target=TARGET_NAME, exclude=DEFAULT_EXCL, missing=MISSING_TOKEN, text=False):

src/dsff/formats/csv.py

@@ -2,7 +2,7 @@
 from .__common__ import *
 
 
-__all__ = ["from_csv", "to_csv"]
+__all__ = ["from_csv", "is_csv", "to_csv"]
 
 
 def from_csv(dsff, path=None, exclude=DEFAULT_EXCL):
@@ -18,6 +18,17 @@ def from_csv(dsff, path=None, exclude=DEFAULT_EXCL):
     dsff.write(features=features)
 
 
+@text_or_path
+def is_csv(text):
+    """ Check if the input text or path is a valid CSV. """
+    try:
+        dialect = csvmod.Sniffer().sniff(text := ensure_str(text))
+        csvmod.Sniffer().has_header(text)
+        return True
+    except (csvmod.Error, UnicodeDecodeError):
+        return False
+
+
 def to_csv(dsff, path=None, text=False):
     """ Create a CSV from the data worksheet, saved as a file or output as a string. """
     with (StringIO() if text else open(path, 'w+')) as f:

src/dsff/formats/dataset.py

@@ -2,19 +2,32 @@
 from .__common__ import *
 
 
-__all__ = ["from_dataset", "to_dataset"]
+__all__ = ["from_dataset", "is_dataset", "to_dataset"]
 
 
-def from_dataset(dsff, path=None):
-    """ Populate the DSFF file from a Dataset structure. """
+def _parse(path):
     if not isdir(path):
         raise BadInputData("Not a folder")
     else:
         if len(missing := [f for f in ["data.csv", "features.json", "metadata.json"] if not isfile(join(path, f))]) > 0:
             raise BadInputData(f"Not a valid dataset folder (missing: {', '.join(missing)})")
+
+
+def from_dataset(dsff, path=None):
+    """ Populate the DSFF file from a Dataset structure. """
+    _parse(path)
     dsff.write(path)
 
 
+def is_dataset(path):
+    """ Check if the input path is a valid Dataset. """
+    try:
+        _parse(path)
+        return True
+    except BadInputData:
+        return False
+
+
 def to_dataset(dsff, path=None):
     """ Create a dataset folder according to the following structure ;
     name

src/dsff/formats/db.py

@@ -2,7 +2,7 @@
 from .__common__ import *
 
 
-__all__ = ["from_db", "to_db"]
+__all__ = ["from_db", "is_db", "to_db"]
 
 
 def from_db(dsff, path=None, exclude=DEFAULT_EXCL):
@@ -30,6 +30,36 @@ def from_db(dsff, path=None, exclude=DEFAULT_EXCL):
     conn.close()
 
 
+@text_or_path
+def is_db(data):
+    """ Check if the input data or path is a valid SQL database. """
+    from sqlite3 import connect, Error
+    from sys import version_info
+    if not data.startswith(b"SQLite format 3\x00"):
+        return False
+    if version_info.minor > 10:
+        try:
+            with connect(":memory:") as c:
+                c.deserialize(data)
+                c.execute("PRAGMA schema_version;")
+            return True
+        except Error:
+            return False
+    else:  # pragma: no cover
+        from os import close, write
+        from tempfile import mkstemp
+        fd, path = mkstemp(suffix=".db")
+        try:
+            write(fd, data)
+            close(fd)
+            fd = None
+            with connect(f"file:{path}?mode=ro&immutable=1", uri=True) as c:
+                c.execute("PRAGMA schema_version;").fetchone()
+            return True
+        except (Error, OSError):
+            return False
+
+
 def to_db(dsff, path=None, text=False, primary_index=0):
     """ Create a SQLDB from the data worksheet, saved as a file or output as a string. """
     from json import dumps