Add basic functionality for live dbs

pydbhub/dbhub.py

@@ -127,7 +127,7 @@ def __prepareVals(self, dbOwner: str = None, dbName: str = None, ident: Identifi
                 data['tag'] = (None, ident.tag)
         return data
 
-    def Databases(self) -> Tuple[List[str], str]:
+    def Databases(self, live: bool = False) -> Tuple[List[str], str]:
         """
         Returns the list of databases in the requesting users account.
         Ref: https://api.dbhub.io/#databases
@@ -142,6 +142,8 @@ def Databases(self) -> Tuple[List[str], str]:
         data = {
             'apikey': (None, self._connection.api_key),
         }
+        if live:
+            data['live'] = "true"
         return httphub.send_request_json(self._connection.server + "/v1/databases", data)
 
     def Columns(self, db_owner: str, db_name: str, table: str, ident: Identifier = None) -> Tuple[List[Dict], str]:
@@ -356,6 +358,35 @@ def Download(self, db_owner: str, db_name: str) -> Tuple[List[bytes], str]:
         data = self.__prepareVals(db_owner, db_name)
         return httphub.send_request(self._connection.server + "/v1/download", data)
 
+    def Execute(self, db_owner: str, db_name: str, sql: str) -> Tuple[int, str]:
+        """
+        Execute a SQLite statement (INSERT, UPDATE, DELETE) on the chosen database, returning the rows changed.
+        Ref: https://api.dbhub.io/#execute
+
+        Parameters
+        ----------
+        db_owner : str
+            The owner of the database
+        db_name : str
+            The name of the live database
+        sql : str
+            The SQLite statement (INSERT, UPDATE, DELETE)
+
+        Returns
+        -------
+        Tuple[int, str]
+            The returned data is
+                - an integer corresponding to the number of rows changed
+                - a string describing the error, if one occurs
+        """
+        data = self.__prepareVals(db_owner, db_name)
+        data['sql'] = base64.b64encode(sql.encode('ascii'))
+        res, err = httphub.send_request_json(self._connection.server + "/v1/execute", data)
+        if err:
+            return None, res
+
+        return res['rows_changed'], None
+
     def Indexes(self, db_owner: str, db_name: str) -> Tuple[List[Dict], str]:
         """
         Returns the details of all indexes in a SQLite database

tests/test_dbhub.py

@@ -35,13 +35,22 @@ def test_configuration():
 
 # https://api.dbhub.io/#databases
 def test_databases(connection):
-    databases, err = connection.Databases()
+    known_dbs = ("Marine Litter Survey (Keep Northern Ireland Beautiful).sqlite", "Join Testing.sqlite")
+    known_live_dbs = ("DB4S daily users by country-live.sqlite", "Join Testing-live.sqlite")
+
+    databases, err = connection.Databases(live=False)
     assert err is None, err
     assert databases is not None, 'No data result'
     assert len(databases) >= 1, 'Missing data result'
-    matching = ("Marine Litter Survey (Keep Northern Ireland Beautiful).sqlite") in databases
-    matching |= ("Join Testing.sqlite") in databases
-    assert matching
+    assert any(db in databases for db in known_dbs), 'Missing data result'
+    assert not any(live_db in databases for live_db in known_live_dbs), 'Incorrect data result'
+
+    live_databases, err = connection.Databases(live=True)
+    assert err is None, err
+    assert live_databases is not None, 'No data result'
+    assert len(live_databases) >= 1, 'Missing data result'
+    assert any(live_db in live_databases for live_db in known_live_dbs), 'Missing data result'
+    assert not any(db in live_databases for db in known_dbs), 'Incorrect data result'
 
 
 # https://api.dbhub.io/#columns
@@ -94,6 +103,20 @@ def test_download(connection):
     assert len(buf) == 12288
 
 
+# https://api.dbhub.io/#execute
+def test_execute(connection):
+    rows_changed, err = connection.Execute(
+        db_owner='justinclift',
+        db_name="Join Testing-live.sqlite",
+        sql='''
+        UPDATE table1 SET Name = 'Foo' WHERE id = 1
+        '''
+    )
+    assert err is None, err
+    assert rows_changed is not None, 'No data result'
+    assert rows_changed == 1, 'Incorrect data result'
+
+
 # https://api.dbhub.io/#indexes
 def test_indexes(connection):
     indexes, err = connection.Indexes(db_name="DB4S daily users by country.sqlite", db_owner="justinclift")