remove JobConfigTable and register_key_source

Author: Thinh Nguyen

datajoint/autopopulate.py

@@ -6,7 +6,7 @@
 import inspect
 from tqdm import tqdm
 from .hash import key_hash
-from .expression import QueryExpression, AndList, _SQLExpression
+from .expression import QueryExpression, AndList
 from .errors import DataJointError, LostConnectionError
 from .settings import config
 from .utils import user_choice, to_camel_case
@@ -90,18 +90,6 @@ def _rename_attributes(table, props):
             for q in parents[1:]:
                 self._key_source *= _rename_attributes(*q)
 
-        jobs_config = self.connection.schemas[self.target.database].jobs_config
-        key_source_query = jobs_config & {"table_name": self.target.table_name}
-        if key_source_query:
-            key_source_uuid = key_hash({"sql": self._key_source.make_sql()})
-            if not (key_source_query & {"key_source_uuid": key_source_uuid}):
-                # different key_source stored in jobs_config
-                return _SQLExpression(
-                    self._key_source.proj(), key_source_query.fetch1("key_source")
-                )
-        else:
-            self.register_key_source(self._key_source.proj().make_sql())
-
         return self._key_source
 
     def make(self, key):
@@ -146,7 +134,7 @@ def _jobs_to_do(self, restrictions):
         if inspect.isclass(todo) and issubclass(todo, QueryExpression):
             todo = todo()
 
-        if not isinstance(todo, (QueryExpression, _SQLExpression)):
+        if not isinstance(todo, QueryExpression):
             raise DataJointError("Invalid key_source value")
 
         try:
@@ -403,66 +391,12 @@ def _Jobs(self):
     def jobs(self):
         return self._Jobs & {"table_name": self.target.table_name}
 
-    def register_key_source(self, sql=None, safemode=None):
-        key_source_sql = sql or self.key_source.proj().make_sql()
-        key_source_uuid = key_hash({"sql": key_source_sql})
-
-        jobs_config = self.connection.schemas[self.target.database].jobs_config
-
-        entry = {
-            "table_name": self.target.table_name,
-            "key_source": key_source_sql,
-            "key_source_uuid": key_source_uuid,
-            "last_refresh_time": datetime.datetime.utcnow(),
-        }
-
-        if jobs_config & {"table_name": self.target.table_name}:
-            if jobs_config & {
-                "table_name": self.target.table_name,
-                "key_source_uuid": key_source_uuid,
-            }:
-                return
-
-            safemode = config["safemode"] if safemode is None else safemode
-            if (
-                not safemode
-                or user_choice(
-                    f"Modified key_source for table `{to_camel_case(self.target.table_name)}`,  Re-register?",
-                    default="no",
-                )
-                == "yes"
-            ):
-                jobs_config.insert1(entry, replace=True)
-        else:
-            jobs_config.insert1(entry)
-
     def schedule_jobs(self, purge_invalid_jobs=True):
         """
         Schedule new jobs for this autopopulate table
         :param purge_invalid_jobs: if True, remove invalid entry from the jobs table (potentially expensive operation)
         :return:
         """
-        jobs_config = self.connection.schemas[self.target.database].jobs_config
-        key_source_query = jobs_config & {"table_name": self.target.table_name}
-        if not key_source_query:
-            self.register_key_source()
-
-        if key_source_query.fetch1("refresh_reserved"):
-            # scheduling is ongoing
-            return
-
-        last_refresh_time, refresh_rate = key_source_query.fetch1(
-            "last_refresh_time", "refresh_rate"
-        )
-        new_refresh_time = last_refresh_time + datetime.timedelta(seconds=refresh_rate)
-        if datetime.datetime.utcnow() < new_refresh_time:
-            # not yet time to refresh jobs
-            return
-
-        # add new scheduled jobs to JobTable
-        jobs_config.update1(
-            {"table_name": self.target.table_name, "refresh_reserved": 1}
-        )
         try:
             with self.connection.transaction:
                 schedule_count = 0
@@ -477,9 +411,6 @@ def schedule_jobs(self, purge_invalid_jobs=True):
         finally:
             if purge_invalid_jobs:
                 self.purge_invalid_jobs()
-            jobs_config.update1(
-                {"table_name": self.target.table_name, "refresh_reserved": 0}
-            )
 
     def purge_invalid_jobs(self):
         """

datajoint/expression.py

@@ -931,61 +931,3 @@ def aggr(self, group, **named_attributes):
         )
 
     aggregate = aggr  # alias for aggr
-
-
-class _SQLExpression:
-    def __init__(self, expression, sql):
-        self._expression = expression
-        self.make_sql = lambda: sql
-        self._expression.cursor = self.cursor
-
-        self._select, self._from_where = re.match(
-            r"SELECT (.*) FROM (.*)", sql
-        ).groups()
-
-        assert set(self._expression.heading.names) == set(
-            v.strip("`") for v in self._select.split(",")
-        )
-
-    def cursor(self, offset=0, limit=None, order_by=None, as_dict=False):
-        """
-        See expression.fetch() for input description.
-        :return: query cursor
-        """
-        if offset and limit is None:
-            raise DataJointError("limit is required when offset is set")
-        sql = self.make_sql()
-        if order_by is not None:
-            sql += " ORDER BY " + ", ".join(order_by)
-        if limit is not None:
-            sql += " LIMIT %d" % limit + (" OFFSET %d" % offset if offset else "")
-        logger.debug(sql)
-        return self._expression.connection.query(sql, as_dict=as_dict)
-
-    @property
-    def heading(self):
-        return self._expression.heading
-
-    @property
-    def fetch1(self):
-        return Fetch1(self._expression)
-
-    @property
-    def fetch(self):
-        return Fetch(self._expression)
-
-    def __len__(self):
-        """:return: number of elements in the result set e.g. ``len(q1)``."""
-        return self._expression.connection.query(
-            "SELECT {select_} FROM {from_where}".format(
-                select_=f"count(DISTINCT {self._select})",
-                from_where=self._from_where,
-            )
-        ).fetchone()[0]
-
-    def __bool__(self):
-        """
-        :return: True if the result is not empty. Equivalent to len(self) > 0 but often
-            faster e.g. ``bool(q1)``.
-        """
-        return bool(self._expression.connection.query(self.make_sql()).fetchone()[0])

datajoint/jobs.py

@@ -233,45 +233,3 @@ def error(self, table_name, key, error_message, error_stack=None):
                 replace=True,
                 ignore_extra_fields=True,
             )
-
-
-class JobConfigTable(Table):
-    """
-    A base table with no definition. Allows job configuration for imported/computed tables
-    """
-
-    def __init__(self, conn, database):
-        self.database = database
-        self._connection = conn
-        self._heading = Heading(
-            table_info=dict(
-                conn=conn, database=database, table_name=self.table_name, context=None
-            )
-        )
-        self._support = [self.full_table_name]
-
-        self._definition = """    # job configuration table for `{database}`
-        table_name  :varchar(255)  # className of the table
-        ---
-        key_source_uuid: UUID # hash of the key_source
-        key_source  :mediumblob  # sql statement for the key_source of the table - from make_sql()
-        unique index (table_name, key_source_uuid)
-        unique index (key_source_uuid)
-        refresh_rate=1: int unsigned # (second) how often should the jobs for the table be refreshed
-        refresh_reserved=0: bool  # is the jobs for the table currently being refreshed
-        last_refresh_time: datetime  # timestamp (UTC) of the last refresh time for the table
-        additional_config=null: longblob  # (TODO: change to json) any additional configuration (e.g. retries, docker image, etc.)
-        """.format(
-            database=database
-        )
-        if not self.is_declared:
-            self.declare()
-        self._user = self.connection.get_user()
-
-    @property
-    def definition(self):
-        return self._definition
-
-    @property
-    def table_name(self):
-        return "~jobs_config"