fist commit

requirements.txt

@@ -0,0 +1,4 @@
+flask==1.1.1
+flask-sqlalchemy==2.3.2
+sqlalchemy==1.3.3
+werkzeug==0.15.3

setup.py

@@ -0,0 +1,39 @@
+# coding: utf-8
+
+# Do not import unicode_literals it generate an error when install module with pip
+from __future__ import (print_function,
+                        absolute_import, division)
+
+import re
+import setuptools
+
+
+def get_version(path="./VERSION"):
+    """ Return the version of by with regex intead of importing it"""
+    version_number = open(path, "rt").read()
+    return version_number
+
+
+setuptools.setup(
+    name='utils-flask-sqlalchemy',
+    version=get_version(),
+    description="Python lib of tools for Flask and SQLAlchemy",
+    long_description=open('README.md', encoding="utf-8").read().strip(),
+    author="Les parcs nationaux de France",
+    url='https://github.com/PnX-SI/Nomenclature-api-module',
+    packages=setuptools.find_packages('src'),
+    package_dir={'': 'src'},
+    install_requires=list(open('requirements.txt', 'r')),
+    include_package_data=True,
+    zip_safe=False,
+    keywords='ww',
+    classifiers=['Development Status :: 1 - Planning',
+                 'Intended Audience :: Developers',
+                 'Natural Language :: English',
+                 'Programming Language :: Python :: 3.3',
+                 'Programming Language :: Python :: 3.4',
+                 'Programming Language :: Python :: 3.5',
+                 'Programming Language :: Python :: 3.6',
+                 'License :: OSI Approved :: GNU Affero General Public License v3'
+                 'Operating System :: OS Independent'],
+)

src/response.py

@@ -0,0 +1,86 @@
+import json
+import csv
+import io
+from functools import wraps
+
+from flask import Response
+from werkzeug.datastructures import Headers
+
+
+def json_resp(fn):
+    """
+    Décorateur transformant le résultat renvoyé par une vue
+    en objet JSON
+    """
+
+    @wraps(fn)
+    def _json_resp(*args, **kwargs):
+        res = fn(*args, **kwargs)
+        if isinstance(res, tuple):
+            return to_json_resp(*res)
+        else:
+            return to_json_resp(res)
+
+    return _json_resp
+
+
+def to_json_resp(
+    res, status=200, filename=None, as_file=False, indent=None, extension="json"
+):
+    if not res:
+        status = 404
+        res = {"message": "not found"}
+
+    headers = None
+    if as_file:
+        headers = Headers()
+        headers.add("Content-Type", "application/json")
+        headers.add(
+            "Content-Disposition",
+            "attachment",
+            filename="export_{}.{}".format(filename, extension),
+        )
+    return Response(
+        json.dumps(res, ensure_ascii=False, indent=indent),
+        status=status,
+        mimetype="application/json",
+        headers=headers,
+    )
+
+
+def csv_resp(fn):
+    """
+    Décorateur transformant le résultat renvoyé en un fichier csv
+    """
+
+    @wraps(fn)
+    def _csv_resp(*args, **kwargs):
+        res = fn(*args, **kwargs)
+        filename, data, columns, separator = res
+        return to_csv_resp(filename, data, columns, separator)
+
+    return _csv_resp
+
+
+def to_csv_resp(filename, data, columns, separator=";"):
+
+    headers = Headers()
+    headers.add("Content-Type", "text/plain")
+    headers.add(
+        "Content-Disposition", "attachment", filename="export_%s.csv" % filename
+    )
+    out = generate_csv_content(columns, data, separator)
+    return Response(out, headers=headers)
+
+
+def generate_csv_content(columns, data, separator):
+    fp = io.StringIO()
+    writer = csv.DictWriter(
+        fp, columns, delimiter=separator, quoting=csv.QUOTE_ALL, extrasaction="ignore"
+    )
+    writer.writeheader()  # ligne d'entête
+
+    for line in data:
+        writer.writerow(line)
+    fp.seek(0)  # Rembobinage du "fichier"
+    return fp.read()  # Retourne une chaine

src/serializers.py

@@ -0,0 +1,92 @@
+"""
+  Serialize function for SQLAlchemy models
+"""
+
+"""
+    List of data type who need a particular serialization
+    @TODO MISSING FLOAT
+"""
+SERIALIZERS = {
+    "date": lambda x: str(x) if x else None,
+    "datetime": lambda x: str(x) if x else None,
+    "time": lambda x: str(x) if x else None,
+    "timestamp": lambda x: str(x) if x else None,
+    "uuid": lambda x: str(x) if x else None,
+    "numeric": lambda x: str(x) if x else None,
+}
+
+
+def serializable(cls):
+    """
+        Décorateur de classe pour les DB.Models
+        Permet de rajouter la fonction as_dict
+        qui est basée sur le mapping SQLAlchemy
+    """
+
+    """
+        Liste des propriétés sérialisables de la classe
+        associées à leur sérializer en fonction de leur type
+    """
+    cls_db_columns = [
+        (
+            db_col.key,
+            SERIALIZERS.get(
+                db_col.type.__class__.__name__.lower(), lambda x: x),
+        )
+        for db_col in cls.__mapper__.c
+        if not db_col.type.__class__.__name__ == "Geometry"
+    ]
+
+    """
+        Liste des propriétés de type relationship
+        uselist permet de savoir si c'est une collection de sous objet
+        sa valeur est déduite du type de relation
+        (OneToMany, ManyToOne ou ManyToMany)
+    """
+    cls_db_relationships = [
+        (db_rel.key, db_rel.uselist) for db_rel in cls.__mapper__.relationships
+    ]
+
+    def serializefn(self, recursif=False, columns=(), relationships=()):
+        """
+        Méthode qui renvoie les données de l'objet sous la forme d'un dict
+
+        Parameters
+        ----------
+            recursif: boolean
+                Spécifie si on veut que les sous objet (relationship)
+                soit également sérialisé
+            columns: liste
+                liste des colonnes qui doivent être prises en compte
+            relationships: liste
+                liste des relationships qui doivent être prise en compte
+        """
+        if columns:
+            fprops = list(filter(lambda d: d[0] in columns, cls_db_columns))
+        else:
+            fprops = cls_db_columns
+        if relationships:
+            selected_relationship = list(
+                filter(lambda d: d[0] in relationships, cls_db_relationships)
+            )
+        else:
+            selected_relationship = cls_db_relationships
+        out = {item: _serializer(getattr(self, item))
+               for item, _serializer in fprops}
+        if recursif is False:
+            return out
+
+        for (rel, uselist) in selected_relationship:
+            if getattr(self, rel):
+                if uselist is True:
+                    out[rel] = [
+                        x.as_dict(recursif, relationships=relationships)
+                        for x in getattr(self, rel)
+                    ]
+                else:
+                    out[rel] = getattr(self, rel).as_dict(recursif)
+
+        return out
+
+    cls.as_dict = serializefn
+    return cls