Merge pull request #99 from mindsdb/add-last

LAST support

Author: ea-rus

mindsdb_sdk/jobs.py

@@ -98,9 +98,15 @@ def get(self, name: str) -> Job:
         else:
             raise RuntimeError("Several jobs with the same name")
 
-    def create(self, name: str, query_str: str,
-                   start_at: dt.datetime = None, end_at: dt.datetime = None,
-                   repeat_str: str = None) -> Union[Job, None]:
+    def create(
+            self,
+            name: str,
+            query_str: str,
+            start_at: dt.datetime = None,
+            end_at: dt.datetime = None,
+            repeat_str: str = None,
+            repeat_min: int = None,
+        ) -> Union[Job, None]:
         """
         Create new job in project and return it.
 
@@ -114,6 +120,7 @@ def create(self, name: str, query_str: str,
         :param start_at: datetime, first start of job,
         :param end_at: datetime, when job have to be stopped,
         :param repeat_str: str, optional, how to repeat job (e.g. '1 hour', '2 weeks', '3 min')
+        :param repeat_min: int, optional, period to repeat the job in minutes
         :return: Job object or None
         """
 
@@ -126,6 +133,10 @@ def create(self, name: str, query_str: str,
             end_str = end_at.strftime("%Y-%m-%d %H:%M:%S")
         else:
             end_str = None
+
+        if repeat_min is not None:
+            repeat_str = f'{repeat_min} minutes'
+
         ast_query = CreateJob(
             name=Identifier(name),
             query_str=query_str,

mindsdb_sdk/tables.py

@@ -5,9 +5,9 @@
 import pandas as pd
 
 from mindsdb_sql.parser.ast import DropTables
-from mindsdb_sql.parser.ast import Select, Star, Identifier, Constant, Delete, Insert, Update
+from mindsdb_sql.parser.ast import Select, Star, Identifier, Constant, Delete, Insert, Update, Last, BinaryOperation
 
-from mindsdb_sdk.utils.sql import dict_to_binary_op
+from mindsdb_sdk.utils.sql import dict_to_binary_op, add_condition
 from mindsdb_sdk.utils.objects_collection import CollectionBase
 
 from .query import Query
@@ -20,6 +20,7 @@ def __init__(self, db, name):
         self.db = db
         self._filters = {}
         self._limit = None
+        self._track_column = None
         self._update_query()
 
     def _filters_repr(self):
@@ -45,6 +46,7 @@ def filter(self, **kwargs):
         'select * from table1 where a=1 and b=2'
 
         :param kwargs: filter
+        :return: Table object
         """
         # creates new object
         query = copy.deepcopy(self)
@@ -57,17 +59,46 @@ def limit(self, val: int):
         Applies limit condition to table query
 
         :param val: limit size
+        :return: Table object
         """
         query = copy.deepcopy(self)
         query._limit = val
         query._update_query()
         return query
 
+    def track(self, column):
+        """
+        Apply tracking column to table. ('LAST' keyword in mindsdb)
+        First call returns nothing
+        The next calls return new records since previous call (where value of track_column is greater)
+
+        Example:
+
+        >>> query = con.databases.my_db.tables.sales.filter(type='house').track('created_at')
+        >>> # first call returns no records
+        >>> df = query.fetch()
+        >>> # second call returns rows with created_at is greater since previous fetch
+        >>> df = query.fetch()
+
+        :param column: column to track new data from table.
+        :return: Table object
+        """
+        query = copy.deepcopy(self)
+        query._track_column = column
+
+        query._update_query()
+        return query
+
     def _update_query(self):
+        where = dict_to_binary_op(self._filters)
+        if self._track_column is not None:
+            condition = BinaryOperation(op='>', args=[Identifier(self._track_column), Last()])
+            where = add_condition(where, condition)
+
         ast_query = Select(
             targets=[Star()],
             from_table=Identifier(self.name),
-            where=dict_to_binary_op(self._filters)
+            where=where
         )
         if self._limit is not None:
             ast_query.limit = Constant(self._limit)

mindsdb_sdk/utils/sql.py

@@ -1,19 +1,23 @@
-from mindsdb_sql.parser.ast import *
+from mindsdb_sql.parser.ast import BinaryOperation, Identifier, Constant
 
 
 def dict_to_binary_op(filters):
     where = None
     for name, value in filters.items():
-        where1 = BinaryOperation('=', args=[Identifier(name), Constant(value)])
-        if where is None:
-            where = where1
-        else:
-            where = BinaryOperation(
-                'and',
-                args=[where, where1]
-            )
+        condition = BinaryOperation('=', args=[Identifier(name), Constant(value)])
+
+        where = add_condition(where, condition)
+
     return where
 
 
+def add_condition(where, condition):
+    if where is None:
+        return condition
+    else:
+        return BinaryOperation(
+            'and',
+            args=[where, condition]
+        )
 
 

tests/test_sdk.py

@@ -169,11 +169,15 @@ def check_model(self, model, database, mock_post):
     def check_table(self, table, mock_post):
         response_mock(mock_post, pd.DataFrame([{'x': 'a'}]))
 
-        table = table.filter(a=3, b='2')
-        table = table.limit(3)
-        table.fetch()
-        str(table)
-        check_sql_call(mock_post, f'SELECT * FROM {table.name} WHERE a = 3 AND b = \'2\' LIMIT 3')
+        table2 = table.filter(a=3, b='2').limit(3)
+        table2.fetch()
+        str(table2)
+        check_sql_call(mock_post, f'SELECT * FROM {table2.name} WHERE a = 3 AND b = \'2\' LIMIT 3')
+
+        # last
+        table2 = table.filter(a=3).track('type')
+        table2.fetch()
+        check_sql_call(mock_post, f'SELECT * FROM {table2.name} WHERE a = 3 AND type > last')
 
 
 class Test(BaseFlow):
@@ -1040,14 +1044,14 @@ def check_project_jobs(self, project, mock_post):
         project.jobs.create(
             name='job2',
             query_str='retrain m1',
-            repeat_str='1 min',
+            repeat_min=1,
             start_at=dt.datetime(2025, 2, 5, 11, 22),
             end_at=dt.date(2030, 1, 2)
         )
 
         check_sql_call(
             mock_post,
-            f"CREATE JOB job2 (retrain m1) START '2025-02-05 11:22:00' END '2030-01-02 00:00:00' EVERY 1 min",
+            f"CREATE JOB job2 (retrain m1) START '2025-02-05 11:22:00' END '2030-01-02 00:00:00' EVERY 1 minutes",
             call_stack_num=-2
         )