Added: custom date serializer (#73)

libreforms_fastapi/app/__init__.py

@@ -413,6 +413,7 @@ async def api_form_create(
     # # Here we validate and coerce data into its proper type
     form_data = FormModel.model_validate(body)
     json_data = form_data.model_dump_json()
+    data_dict = form_data.model_dump()
 
     # Ugh, I'd like to find a more efficient way to get the user data. But alas, that
     # the sqlalchemy-signing table is not optimized alongside the user model...
@@ -443,7 +444,8 @@ async def api_form_create(
         # doc_db.create_document,
     d = doc_db.create_document(
         form_name=form_name, 
-        json_data=json_data, 
+        # json_data=json_data, 
+        data_dict=data_dict, 
         metadata=metadata,
     )
 
@@ -626,6 +628,7 @@ async def api_form_update(
     # # Here we validate and coerce data into its proper type
     form_data = FormModel.model_validate(body)
     json_data = form_data.model_dump_json()
+    data_dict = form_data.model_dump()
 
     # print("\n\n\n", json_data)
 
@@ -660,7 +663,8 @@ async def api_form_update(
         d = doc_db.update_document(
             form_name=form_name, 
             document_id=document_id,
-            json_data=json_data, 
+            # json_data=json_data,
+            updated_data_dict=data_dict, 
             metadata=metadata,
             limit_users=limit_query_to,
         )

libreforms_fastapi/utils/certificates.py

@@ -113,29 +113,6 @@ def sign_data(self, data):
         )
         return signature
 
-    # def verify_signature(self, data, signature):
-    #     """
-    #     Verifies the signature of the data using the public key.
-    #     """
-    #     with open(self.get_public_key_file(), "rb") as key_file:
-    #         public_key = serialization.load_pem_public_key(
-    #             key_file.read(),
-    #             backend=default_backend()
-    #         )
-
-    #     try:
-    #         public_key.verify(
-    #             signature,
-    #             data,
-    #             padding.PSS(
-    #                 mgf=padding.MGF1(hashes.SHA256()),
-    #                 salt_length=padding.PSS.MAX_LENGTH
-    #             ),
-    #             hashes.SHA256()
-    #         )
-    #         return True
-    #     except Exception as e:
-    #         return False
     def verify_signature(self, data, signature, public_key=None):
         """
         Verifies the signature of the data using the provided public key.

libreforms_fastapi/utils/custom_tinydb.py

@@ -0,0 +1,144 @@
+import json
+from datetime import date
+from json import JSONEncoder
+from bson import ObjectId
+
+from tinydb import (
+    TinyDB, 
+    Query, 
+    Storage
+)
+from tinydb.table import (
+    Table as TinyTable, 
+    Document
+)
+
+from typing import (
+    Mapping,
+    Union,
+    Iterable,
+    List,
+)
+
+
+class DateEncoder(JSONEncoder):
+    """We need to convert date objects to 'YYYY-MM-DD' format"""
+    def default(self, obj):
+        if isinstance(obj, date):
+            return obj.isoformat()
+        # Fall back to the superclass method for other types
+        return JSONEncoder.default(self, obj)
+
+# We want to modify TinyDB use use string representations of bson 
+# ObjectIDs. As such, we will need to modify some underlying behavior, 
+# see https://github.com/signebedi/libreforms-fastapi/issues/15.
+class CustomTable(TinyTable):
+    document_id_class = str  # Use string IDs instead of integers
+
+    def _get_next_id(self, document_id=str(ObjectId())):
+        """
+        Generate a new BSON ObjectID string to use as the TinyDB document ID.
+        """
+        return document_id
+
+
+    def insert(self, document: Mapping, document_id:Union[str, bool]=False) -> int:
+        """
+        Insert a new document into the table.
+
+        :param document: the document to insert
+        :returns: the inserted document's ID
+        """
+
+        if not document_id:
+            document_id = str(ObjectId())
+
+        # Make sure the document implements the ``Mapping`` interface
+        if not isinstance(document, Mapping):
+            raise ValueError('Document is not a Mapping')
+
+        # First, we get the document ID for the new document
+        if isinstance(document, Document):
+            # For a `Document` object we use the specified ID
+            doc_id = document.doc_id
+
+            # We also reset the stored next ID so the next insert won't
+            # re-use document IDs by accident when storing an old value
+            self._next_id = None
+        else:
+            # In all other cases we use the next free ID
+            doc_id = self._get_next_id(document_id=document_id)
+
+        # Now, we update the table and add the document
+        def updater(table: dict):
+            if doc_id in table:
+                raise ValueError(f'Document with ID {str(doc_id)} '
+                                 f'already exists')
+
+            # By calling ``dict(document)`` we convert the data we got to a
+            # ``dict`` instance even if it was a different class that
+            # implemented the ``Mapping`` interface
+            table[doc_id] = dict(document)
+
+        # See below for details on ``Table._update``
+        self._update_table(updater)
+
+        return doc_id
+
+    def insert_multiple(self, documents: Iterable[Mapping], document_ids:Union[List, bool]=False) -> List[int]:
+        """
+        Insert multiple documents into the table.
+
+        :param documents: an Iterable of documents to insert
+        :returns: a list containing the inserted documents' IDs
+        """
+        doc_ids = []
+
+        if document_ids and len(document_ids) != len(documents):
+            raise Exception("When inserting multiple and passing your own document_ids," \
+                "the list must be the same length as the document list")
+
+        def updater(table: dict):
+            # for document in documents:
+            for i, document in enumerate(documents):
+
+                # Make sure the document implements the ``Mapping`` interface
+                if not isinstance(document, Mapping):
+                    raise ValueError('Document is not a Mapping')
+
+                if isinstance(document, Document):
+                    # Check if document does not override an existing document
+                    if document.doc_id in table:
+                        raise ValueError(
+                            f'Document with ID {str(document.doc_id)} '
+                            f'already exists'
+                        )
+
+                    # Store the doc_id, so we can return all document IDs
+                    # later. Then save the document with its doc_id and
+                    # skip the rest of the current loop
+                    doc_id = document.doc_id
+                    doc_ids.append(doc_id)
+                    table[doc_id] = dict(document)
+                    continue
+
+                # Generate new document ID for this document
+                # Store the doc_id, so we can return all document IDs
+                # later, then save the document with the new doc_id
+                if not document_ids:
+                    document_id = str(ObjectId())
+                else:
+                    document_id = document_ids[i]
+                doc_id = self._get_next_id()
+                doc_ids.append(doc_id)
+                table[doc_id] = dict(document)
+
+        # See below for details on ``Table._update``
+        self._update_table(updater)
+
+        return doc_ids
+
+# Subclass TinyDB and override the table_class attribute with our new logic
+class CustomTinyDB(TinyDB):
+    table_class = CustomTable
+