Repository: innobi/pantab (python)

Issue:
Append Mode Support
After discussing with @rhelenius on the Tableau forums I think need to add support for Append Mode with Hyper Extracts. This can support delta processes and potential chunking of datasets into Hyper.

Some consideration points:

  * The meaning of "Append" can be ambiguous; in some contexts append may refer to adding a Table to an Extract / Schema and in other contexts this may mean appending to an existing table
  * Need to be cognizant of what the Hyper API exposes as terminology at the database level; those terms are:
    * CREATE: Create the database. Will fail if it already exists
    * CREATE_AND_REPLACE: Create the database. If already exists, drop the old one first
    * CREATE_IF_NOT_EXISTS: Create the database if it does not exist
    * NONE: Do not create the database. Will fail if it already exists

I'll jot some more thoughts down later, but we may want to offer similar enumerations for Table creation and add as a keyword argument to the existing functions, or maybe even just re-purpose the enum already exposed by Hyper

Pull request:
Add Support for Appending to Existing Tables
closes #14 

Patch:
diff --git a/doc/source/examples.rst b/doc/source/examples.rst
index 54e3eb0..a074232 100644
--- a/doc/source/examples.rst
+++ b/doc/source/examples.rst
@@ -80,6 +80,26 @@ Reading and Writing Multiple Tables
 
    While you can write using ``str``, ``tableauhyperapi.Name`` or ``tableauhyperapi.TableName`` instances, the keys of the dict returned by ``frames_from_hyper`` will always be ``tableauhyperapi.TableName`` instances
 
-.. warning::
+Appending Data to Existing Tables
+---------------------------------
 
-   pantab currently supports only complete read and writes of a hyper extract. Partial updates are not implemented. See :doc:`caveats`
+By default, ``frame_to_hyper`` and ``frames_to_hyper`` will fully drop and reloaded targeted tables. However, you can also append records to existing tables by supplying ``table_mode="a"`` as a keyword argument.
+
+.. code-block:: python
+
+   import pandas as pd
+   import pantab
+
+   df = pd.DataFrame([
+       ["dog", 4],
+       ["cat", 4],
+   ], columns=["animal", "num_of_legs"])
+
+   pantab.frame_to_hyper(df, "example.hyper", table="animals")
+
+   new_data = pd.DataFrame([["moose", 4]], columns=["animal", "num_of_legs"])
+
+   # Instead of overwriting the animals table, we can append via table_mode
+   pantab.frame_to_hyper(df, "example.hyper", table="animals", table_mode="a")
+
+Please note that ``table_mode="a"`` will create the table(s) if they do not already exist.
diff --git a/pantab/_pantab.py b/pantab/_pantab.py
index fa0de9a..1967d6d 100644
--- a/pantab/_pantab.py
+++ b/pantab/_pantab.py
@@ -1,9 +1,10 @@
 import collections
+import itertools
 import pathlib
 import shutil
 import tempfile
 import uuid
-from typing import Dict, List, Union
+from typing import Dict, List, Sequence, Union
 
 import numpy as np
 import pandas as pd
@@ -93,27 +94,95 @@ def _interval_to_timedelta(interval: tab_api.Interval) -> pd.Timedelta:
     return pd.Timedelta(days=interval.days, microseconds=interval.microseconds)
 
 
+def _validate_table_mode(table_mode: str) -> None:
+    if table_mode not in {"a", "w"}:
+        raise ValueError("'table_mode' must be either 'w' or 'a'")
+
+
+def _assert_columns_equal(
+    left: Sequence[tab_api.TableDefinition.Column],
+    right: Sequence[tab_api.TableDefinition.Column],
+) -> None:
+    """
+    Helper function to validate if sequences of columns are equal.
+
+    The TableauHyperAPI as of 0.0.8953 does not implement equality operations
+    for Column instances, hence the need for this.
+    """
+
+    class DummyColumn:
+        """Dummy class to match items needed for str repr of columns."""
+
+        @property
+        def name(self):
+            return None
+
+        @property
+        def type(self):
+            return None
+
+        @property
+        def nullability(self):
+            return None
+
+    for c1, c2 in itertools.zip_longest(left, right, fillvalue=DummyColumn()):
+        if c1.name != c2.name or c1.type != c2.type or c1.nullability != c2.nullability:
+            break  # go to error handler
+    else:
+        return None  # everything matched up, so bail out
+
+    c1_str = ", ".join(
+        f"(Name={x.name}, Type={x.type}, Nullability={x.nullability})" for x in left
+    )
+    c2_str = ", ".join(
+        f"(Name={x.name}, Type={x.type}, Nullability={x.nullability})" for x in right
+    )
+
+    raise TypeError(f"Mismatched column definitions: {c1_str} != {c2_str}")
+
+
 def _insert_frame(
-    df: pd.DataFrame, *, connection: tab_api.Connection, table: TableType
+    df: pd.DataFrame,
+    *,
+    connection: tab_api.Connection,
+    table: TableType,
+    table_mode: str,
 ) -> None:
+    _validate_table_mode(table_mode)
+
     if isinstance(table, str):
         table = tab_api.TableName(table)
 
-    table_def = tab_api.TableDefinition(table)
+    # Populate insertion mechanisms dependent on column types
     insert_funcs: List[str] = []
-
+    column_types: List[_ColumnType] = []
+    columns: List[tab_api.TableDefinition.Column] = []
     for col_name, dtype in df.dtypes.items():
         column_type = _pandas_to_tableau_type(dtype.name)
-        col = tab_api.TableDefinition.Column(
-            name=col_name, type=column_type.type_, nullability=column_type.nullability
-        )
+        column_types.append(column_type)
         insert_funcs.append(_insert_functions[column_type.type_])
-        table_def.add_column(col)
+        columns.append(
+            tab_api.TableDefinition.Column(
+                name=col_name,
+                type=column_type.type_,
+                nullability=column_type.nullability,
+            )
+        )
+
+    # Sanity check for existing table structures
+    if table_mode == "a" and connection.catalog.has_table(table):
+        table_def = connection.catalog.get_table_definition(table)
+        _assert_columns_equal(columns, table_def.columns)
+    else:  # New table, potentially new schema
+        table_def = tab_api.TableDefinition(table)
+
+        for column, column_type in zip(columns, column_types):
+            table_def.add_column(column)
 
-    if isinstance(table, tab_api.TableName) and table.schema_name:
-        connection.catalog.create_schema_if_not_exists(table.schema_name)
+        if isinstance(table, tab_api.TableName) and table.schema_name:
+            connection.catalog.create_schema_if_not_exists(table.schema_name)
 
-    connection.catalog.create_table_if_not_exists(table_def)
+        connection.catalog.create_table_if_not_exists(table_def)
 
     # Special handling for conversions
     df = df.copy()
@@ -166,29 +235,25 @@ def _read_table(*, connection: tab_api.Connection, table: TableType) -> pd.DataF
 
 
 def frame_to_hyper(
-    df: pd.DataFrame, database: Union[str, pathlib.Path], *, table: TableType
+    df: pd.DataFrame,
+    database: Union[str, pathlib.Path],
+    *,
+    table: TableType,
+    table_mode: str = "w",
 ) -> None:
-    """
-    Convert a DataFrame to a .hyper extract.
-
-    Parameters
-    ----------
-    df : DataFrame
-        Data to be written out.
-    database : str
-        tab_api.Name / location of the Hyper file to be written to.
-    table : str, tab_api.Name or tab_api.TableName
-        tab_api.Name of the table to write to. Must be supplied as a keyword argument.
-    """
+    """See api.rst for documentation"""
     with tab_api.HyperProcess(
         tab_api.Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU
     ) as hpe:
         tmp_db = pathlib.Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.hyper"
 
+        if table_mode == "a" and pathlib.Path(database).exists():
+            shutil.move(database, tmp_db)
+
         with tab_api.Connection(
-            hpe.endpoint, tmp_db, tab_api.CreateMode.CREATE
+            hpe.endpoint, tmp_db, tab_api.CreateMode.CREATE_IF_NOT_EXISTS
         ) as connection:
-            _insert_frame(df, connection=connection, table=table)
+            _insert_frame(df, connection=connection, table=table, table_mode=table_mode)
 
         shutil.move(tmp_db, database)
 
@@ -196,20 +261,7 @@ def frame_to_hyper(
 def frame_from_hyper(
     database: Union[str, pathlib.Path], *, table: TableType
 ) -> pd.DataFrame:
-    """
-    Extracts a DataFrame from a .hyper extract.
-
-    Parameters
-    ----------
-    database : str
-        tab_api.Name / location of the Hyper file to be read.
-    table : str
-        tab_api.Name of the table to read. Must be supplied as a keyword argument.
-
-    Returns
-    -------
-    DataFrame
-    """
+    """See api.rst for documentation"""
     with tempfile.TemporaryDirectory() as tmp_dir, tab_api.HyperProcess(
         tab_api.Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU
     ) as hpe:
@@ -219,19 +271,28 @@ def frame_from_hyper(
 
 
 def frames_to_hyper(
-    dict_of_frames: Dict[TableType, pd.DataFrame], database: Union[str, pathlib.Path]
+    dict_of_frames: Dict[TableType, pd.DataFrame],
+    database: Union[str, pathlib.Path],
+    table_mode: str = "w",
 ) -> None:
     """See api.rst for documentation."""
+    _validate_table_mode(table_mode)
+
     with tab_api.HyperProcess(
         tab_api.Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU
     ) as hpe:
         tmp_db = pathlib.Path(tempfile.gettempdir()) / f"{uuid.uuid4()}.hyper"
 
+        if table_mode == "a" and pathlib.Path(database).exists():
+            shutil.move(database, tmp_db)
+
         with tab_api.Connection(
-            hpe.endpoint, tmp_db, tab_api.CreateMode.CREATE
+            hpe.endpoint, tmp_db, tab_api.CreateMode.CREATE_IF_NOT_EXISTS
         ) as connection:
             for table, df in dict_of_frames.items():
-                _insert_frame(df, connection=connection, table=table)
+                _insert_frame(
+                    df, connection=connection, table=table, table_mode=table_mode
+                )
 
         shutil.move(tmp_db, database)
 
diff --git a/pantab/newsfragments/14.feature b/pantab/newsfragments/14.feature
new file mode 100644
index 0000000..693f2b1
--- /dev/null
+++ b/pantab/newsfragments/14.feature
@@ -0,0 +1,1 @@
+frame_to_hyper and frames_to_hyper now support a table_mode keyword argument. ``table_mode="a"`` will append data to existing tables, or create them if they do not exist. The default operation of ``table_mode="w"`` will continue to fully drop / reload tables.
\ No newline at end of file


Test patch:
diff --git a/pantab/tests/test_pantab.py b/pantab/tests/test_pantab.py
index 2c3ccda..32e8042 100644
--- a/pantab/tests/test_pantab.py
+++ b/pantab/tests/test_pantab.py
@@ -86,6 +86,12 @@ def tmp_hyper(tmp_path):
     return tmp_path / "test.hyper"
 
 
+@pytest.fixture(params=["w", "a"])
+def table_mode(request):
+    """Write or append markers for table handling."""
+    return request.param
+
+
 @pytest.fixture(
     params=[
         "table",
@@ -100,34 +106,54 @@ def table_name(request):
     return request.param
 
 
-def test_roundtrip(df, tmp_hyper, table_name):
-    pantab.frame_to_hyper(df, tmp_hyper, table=table_name)
+def test_roundtrip(df, tmp_hyper, table_name, table_mode):
+    # Write twice; depending on mode this should either overwrite or duplicate entries
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode=table_mode)
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode=table_mode)
     result = pantab.frame_from_hyper(tmp_hyper, table=table_name)
+
     expected = df.copy()
+
+    if table_mode == "a":
+        expected = pd.concat([expected, expected]).reset_index(drop=True)
+
     expected["float32"] = expected["float32"].astype(np.float64)
 
     tm.assert_frame_equal(result, expected)
 
 
-def test_roundtrip_missing_data(tmp_hyper, table_name):
+def test_roundtrip_missing_data(tmp_hyper, table_name, table_mode):
     df = pd.DataFrame([[np.nan], [1]], columns=list("a"))
     df["b"] = pd.Series([None, np.nan], dtype=object)  # no inference
     df["c"] = pd.Series([np.nan, "c"])
 
-    pantab.frame_to_hyper(df, tmp_hyper, table=table_name)
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode=table_mode)
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode=table_mode)
 
     result = pantab.frame_from_hyper(tmp_hyper, table=table_name)
     expected = pd.DataFrame(
         [[np.nan, np.nan, np.nan], [1, np.nan, "c"]], columns=list("abc")
     )
-    tm.assert_frame_equal(result, expected)
+    if table_mode == "a":
+        expected = pd.concat([expected, expected]).reset_index(drop=True)
 
+    tm.assert_frame_equal(result, expected)
 
-def test_roundtrip_multiple_tables(df, tmp_hyper, table_name):
-    pantab.frames_to_hyper({table_name: df, "table2": df}, tmp_hyper)
 
+def test_roundtrip_multiple_tables(df, tmp_hyper, table_name, table_mode):
+    # Write twice; depending on mode this should either overwrite or duplicate entries
+    pantab.frames_to_hyper(
+        {table_name: df, "table2": df}, tmp_hyper, table_mode=table_mode
+    )
+    pantab.frames_to_hyper(
+        {table_name: df, "table2": df}, tmp_hyper, table_mode=table_mode
+    )
     result = pantab.frames_from_hyper(tmp_hyper)
+
     expected = df.copy()
+    if table_mode == "a":
+        expected = pd.concat([expected, expected]).reset_index(drop=True)
+
     expected["float32"] = expected["float32"].astype(np.float64)
 
     # some test trickery here
@@ -141,6 +167,33 @@ def test_roundtrip_multiple_tables(df, tmp_hyper, table_name):
         tm.assert_frame_equal(val, expected)
 
 
+def test_bad_table_mode_raises(df, tmp_hyper):
+    msg = "'table_mode' must be either 'w' or 'a'"
+    with pytest.raises(ValueError, match=msg):
+        pantab.frame_to_hyper(df, tmp_hyper, table="test", table_mode="x")
+
+    with pytest.raises(ValueError, match=msg):
+        pantab.frames_to_hyper({"a": df}, tmp_hyper, table_mode="x")
+
+
+def test_append_mode_raises_column_mismatch(df, tmp_hyper, table_name):
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name)
+
+    df = df.drop("object", axis=1)
+    msg = "^Mismatched column definitions:"
+    with pytest.raises(TypeError, match=msg):
+        pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode="a")
+
+
+def test_append_mode_raises_column_dtype_mismatch(df, tmp_hyper, table_name):
+    pantab.frame_to_hyper(df, tmp_hyper, table=table_name)
+
+    df["int16"] = df["int16"].astype(np.int64)
+    msg = "^Mismatched column definitions:"
+    with pytest.raises(TypeError, match=msg):
+        pantab.frame_to_hyper(df, tmp_hyper, table=table_name, table_mode="a")
+
+
 def test_read_doesnt_modify_existing_file(df, tmp_hyper):
     pantab.frame_to_hyper(df, tmp_hyper, table="test")
     last_modified = tmp_hyper.stat().st_mtime