Add timestamp filter

dynamicannotationdb/database.py

@@ -1,6 +1,7 @@
 import logging
 from contextlib import contextmanager
 from typing import List
+import datetime
 
 from sqlalchemy import create_engine, func, inspect, or_
 from sqlalchemy.ext.automap import automap_base
@@ -350,7 +351,8 @@ def _check_table_is_unique(self, table_name):
             )
         return existing_tables
 
-    def _get_existing_table_names(self, filter_valid: bool = False) -> List[str]:
+    def _get_existing_table_names(self, filter_valid: bool = False,
+                                  filter_timestamp: datetime.datetime = None) -> List[str]:
         """Collects table_names keys of existing tables
 
         Returns
@@ -362,6 +364,10 @@ def _get_existing_table_names(self, filter_valid: bool = False) -> List[str]:
             stmt = session.query(AnnoMetadata)
             if filter_valid:
                 stmt = stmt.filter(AnnoMetadata.valid == True)
+            if filter_timestamp:
+                stmt = stmt.filter(AnnoMetadata.created <= filter_timestamp)
+                stmt = stmt.filter(or_(AnnoMetadata.deleted > filter_timestamp,
+                                       AnnoMetadata.deleted == None))
             metadata = stmt.all()
             return [m.table_name for m in metadata]
 

pyproject.toml

@@ -0,0 +1,49 @@
+[build-system]
+requires = ["setuptools>=45", "wheel", "setuptools_scm[toml]>=6.2"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "DynamicAnnotationDB"
+version = "5.13.1"
+description = "Annotation Database pendant to the chunkedgraph"
+readme = "README.md"
+authors = [
+    {name = "Sven Dorkenwald", email = "sdorkenw@princeton.edu"},
+    {name = "Derrick Brittain"}
+]
+license = {text = "MIT"}
+classifiers = [
+    "Programming Language :: Python :: 3",
+    "License :: OSI Approved :: MIT License",
+    "Operating System :: POSIX :: Linux",
+]
+requires-python = ">=3.8"
+dependencies = [
+    "marshmallow==3.5.1",
+    "emannotationschemas>=5.4.0",
+    "sqlalchemy<1.4",
+    "psycopg2-binary",
+    "geoalchemy2",
+    "pytz",
+    "alembic",
+    "shapely",
+    "jsonschema<4.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/seung-lab/DynamicAnnotationDB"
+Repository = "https://github.com/seung-lab/DynamicAnnotationDB"
+
+[dependency-groups]
+dev = [
+    "pytest",
+    "pytest-cov",
+    "docker"
+]
+
+[tool.pytest.ini_options]
+addopts = "--cov dynamicannotationdb --cov-report html --junitxml=test-reports/test.xml"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["dynamicannotationdb*"]

tests/test_database.py

@@ -6,7 +6,19 @@
 from sqlalchemy.ext.declarative.api import DeclarativeMeta
 
 from emannotationschemas import type_mapping
-
+from dynamicannotationdb.errors import TableAlreadyExists
+
+@pytest.fixture(scope="session")
+def deleted_table_metadata():
+    yield {
+        "aligned_volume": "test_volume",
+        "table_name": "deleted_table_test",
+        "schema_type": "synapse",
+        "pcg_table_name": "test_pcg",
+        "voxel_resolution_x": 4.0,
+        "voxel_resolution_y": 4.0,
+        "voxel_resolution_z": 40.0,
+    }
 
 def test_get_table_metadata(dadb_interface, annotation_metadata):
     table_name = annotation_metadata["table_name"]
@@ -65,6 +77,44 @@ def test__get_existing_table_ids(dadb_interface):
     table_names = dadb_interface.database._get_existing_table_names()
     assert isinstance(table_names, list)
 
+def test__get_existing_table_ids_with_filters(dadb_interface, deleted_table_metadata):
+    # without filters
+    before_creation = datetime.datetime.utcnow()
+    dadb_interface.annotation.create_table(
+        deleted_table_metadata["table_name"],
+        deleted_table_metadata["schema_type"],
+        description="some description",
+        user_id="foo@bar.com",
+        voxel_resolution_x=deleted_table_metadata["voxel_resolution_x"],
+        voxel_resolution_y=deleted_table_metadata["voxel_resolution_y"],
+        voxel_resolution_z=deleted_table_metadata["voxel_resolution_z"])
+
+    table_names = dadb_interface.database._get_existing_table_names(filter_timestamp=before_creation)
+    assert isinstance(table_names, list)
+    assert deleted_table_metadata["table_name"] not in table_names
+
+    after_creation = datetime.datetime.utcnow()
+    table_names = dadb_interface.database._get_existing_table_names(filter_timestamp=after_creation)
+    assert deleted_table_metadata["table_name"] in table_names
+
+    dadb_interface.annotation.delete_table(deleted_table_metadata["table_name"])
+    after_deletion = datetime.datetime.utcnow()
+    table_names = dadb_interface.database._get_existing_table_names(filter_timestamp=after_deletion)
+    assert deleted_table_metadata["table_name"] not in table_names
+
+    table_names = dadb_interface.database._get_existing_table_names(filter_timestamp=after_creation)
+    assert deleted_table_metadata["table_name"] in table_names
+
+    # make sure i can't create it after i delete it
+    with pytest.raises(TableAlreadyExists):
+        dadb_interface.annotation.create_table(
+            deleted_table_metadata["table_name"],
+            deleted_table_metadata["schema_type"],
+            description="some description",
+            user_id="foo@bar.com",
+            voxel_resolution_x=deleted_table_metadata["voxel_resolution_x"],
+            voxel_resolution_y=deleted_table_metadata["voxel_resolution_y"],
+            voxel_resolution_z=deleted_table_metadata["voxel_resolution_z"])
 
 def test_get_table_row_count(dadb_interface, annotation_metadata):
     table_name = annotation_metadata["table_name"]