Eng 1086 create syntax sugar for sql queries (#103)

* first commit

* integration test

* test

* style

* update on test and enable regex filtering out tag

* stlye

* change variable name to tag

* make regex constant and add doc

* style fix

* style fix

Co-authored-by: Ubuntu <[email protected]>

integration_tests/sdk/sql_integration_test.py

@@ -37,3 +37,13 @@ def test_invalid_destination_integration(client):
         output_artifact.save(
             config=db.config(table=generate_table_name(), update_mode=LoadUpdateMode.REPLACE)
         )
+
+
+def test_sql_today_tag(client):
+    db = client.integration(name=get_integration_name())
+    sql_artifact_today = db.sql(query="select * from hotel_reviews where review_date = {{today}}")
+    assert sql_artifact_today.get().empty
+    sql_artifact_not_today = db.sql(
+        query="select * from hotel_reviews where review_date < {{today}}"
+    )
+    assert len(sql_artifact_not_today.get()) == 100

src/python/aqueduct_executor/operators/connectors/tabular/relational.py

@@ -1,11 +1,18 @@
 from typing import List
 
 import pandas as pd
+import re
+from datetime import date
 from sqlalchemy import engine, inspect
 from sqlalchemy.exc import SQLAlchemyError
 
 from aqueduct_executor.operators.connectors.tabular import connector, extract, load
 
+# Regular Expression that matches any substring apperance with
+# "{{ }}" and a word inside with optional space in front or after
+# Potential Matches: "{{today}}", "{{ today  }}""
+TAG_PATTERN = r"{{[\s+]*\w+[\s+]*}}"
+
 
 class RelationalConnector(connector.TabularConnector):
     def __init__(self, conn_engine: engine.Engine):
@@ -24,7 +31,15 @@ def discover(self) -> List[str]:
         return inspect(self.engine).get_table_names()
 
     def extract(self, params: extract.RelationalParams) -> pd.DataFrame:
-        df = pd.read_sql(params.query, con=self.engine)
+        query = params.query
+        matches = re.findall(TAG_PATTERN, query)
+        for match in matches:
+            tag = match.strip(" " "{}")
+            if tag == "today":
+                today_python = date.today()
+                today_sql = "'" + today_python.strftime("%Y-%m-%d") + "'"
+                query = query.replace("{{today}}", today_sql)
+        df = pd.read_sql(query, con=self.engine)
         return df
 
     def load(self, params: load.RelationalParams, df: pd.DataFrame) -> None: