Export check_timestamps method (#47)

* Export check_timestamps method

This gives applications more explicit control on when timestamps are
checked.

* Update Spark download link

Author: Daniel Thom

.github/workflows/ci.yml

@@ -30,7 +30,7 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install ".[dev,spark]"
-        wget https://dlcdn.apache.org/spark/spark-3.5.4/spark-3.5.4-bin-hadoop3.tgz
+        wget https://archive.apache.org/dist/spark/spark-3.5.4/spark-3.5.4-bin-hadoop3.tgz
         tar -xzf spark-3.5.4-bin-hadoop3.tgz
         export SPARK_HOME=$(pwd)/spark-3.5.4-bin-hadoop3
         export PATH=$SPARK_HOME/sbin:$PATH

src/chronify/csv_time_series_parser.py

@@ -124,7 +124,7 @@ def _ingest_data(self, data: pd.DataFrame, table_name: str, year: int, length: i
                 assert src_schema is not None
                 self._store.ingest_pivoted_table(data, src_schema, dst_schema)
             case fmt if fmt in UNPIVOTED_TABLES:
-                self._store.ingest_table(data, dst_schema, bypass_time_checks=True)
+                self._store.ingest_table(data, dst_schema, skip_time_checks=True)
 
     @staticmethod
     def _create_schemas(

src/chronify/schema_manager.py

@@ -52,7 +52,7 @@ def __init__(self, engine: Engine, metadata: MetaData):
                 table = Table(
                     self.SCHEMAS_TABLE,
                     self._metadata,
-                    Column("name", String),
+                    Column("name", String, nullable=False, unique=True),
                     Column("schema", String),  # schema encoded as JSON
                 )
                 self._metadata.create_all(self._engine, tables=[table])

src/chronify/store.py

@@ -259,6 +259,30 @@ def schema_manager(self) -> SchemaManager:
         """Return the store's schema manager."""
         return self._schema_mgr
 
+    def check_timestamps(self, name: str, connection: Connection | None = None) -> None:
+        """Check the timestamps in the table.
+
+        This is useful if you call a :meth:`ingest_table` many times with skip_time_checks=True
+        and then want to check the final table.
+
+        Parameters
+        ----------
+        name
+            Name of the table to check.
+
+        Raises
+        ------
+        InvalidTable
+            Raised if the timestamps do not match the schema.
+        """
+        table = self.get_table(name)
+        schema = self._schema_mgr.get_schema(name)
+        if connection is None:
+            with self._engine.connect() as conn:
+                check_timestamps(conn, table, schema)
+        else:
+            check_timestamps(connection, table, schema)
+
     def create_view_from_parquet(
         self, path: Path, schema: TableSchema, bypass_checks: bool = False
     ) -> None:
@@ -771,13 +795,13 @@ def _ingest_tables(
         conn: Connection,
         data: Iterable[pd.DataFrame | DuckDBPyRelation],
         schema: TableSchema,
-        bypass_time_checks: bool = False,
+        skip_time_checks: bool = False,
     ) -> bool:
         created_table = False
         for table in data:
             if self._ingest_table(conn, table, schema):
                 created_table = True
-        if not bypass_time_checks:
+        if not skip_time_checks:
             check_timestamps(conn, Table(schema.name, self._metadata), schema)
         return created_table
 

tests/test_mapper_column_representative_to_datetime.py

@@ -146,7 +146,7 @@ def test_NYMDPV_mapper(time_series_NYMDPV, iter_store: Store):
     )
 
     data = pd.read_csv(time_series_NYMDPV)
-    iter_store.ingest_table(data, from_schema, bypass_time_checks=True)
+    iter_store.ingest_table(data, from_schema, skip_time_checks=True)
 
     metadata = MetaData()
     metadata.reflect(iter_store.engine, views=True)

tests/test_store.py

@@ -770,3 +770,12 @@ def test_read_raw_query(iter_stores_by_engine: Store, one_week_per_month_by_hour
     with store.engine.connect() as conn:
         df2 = store.read_raw_query(query, params=params, connection=conn)
     assert df2.equals(df[df["id"] == 2].reset_index(drop=True))
+
+
+def test_check_timestamps(iter_stores_by_engine: Store, one_week_per_month_by_hour_table) -> None:
+    store = iter_stores_by_engine
+    df, _, schema = one_week_per_month_by_hour_table
+    store.ingest_table(df, schema)
+    store.check_timestamps(schema.name)
+    with store.engine.begin() as conn:
+        store.check_timestamps(schema.name, connection=conn)