Added transaction{}, option to autocommit

Author: erezsh

preql/__main__.py

@@ -21,6 +21,7 @@
                     help="database url (postgres://user:password@host:port/db_name")
 parser.add_argument('--python-traceback', action='store_true',
                     help="Show the Python traceback when an exception causes the interpreter to quit")
+parser.add_argument('--autocommit', action='store_true')
 
 
 def find_dot_preql():
@@ -59,7 +60,7 @@ def main():
 
 
 
-    kw = {'print_sql': args.print_sql}
+    kw = {'print_sql': args.print_sql, 'autocommit': args.autocommit}
     if args.database:
         kw['db_uri'] = args.database
         kw['auto_create'] = True

preql/api.py

@@ -126,7 +126,9 @@ class Preql:
 
     __name__ = "Preql"
 
-    def __init__(self, db_uri: str='sqlite://:memory:', print_sql: bool=settings.print_sql, auto_create: bool = False):
+    def __init__(self, db_uri: str='sqlite://:memory:', print_sql: bool=settings.print_sql,
+                 auto_create: bool = False, autocommit: bool = False
+                 ):
         """Initialize a new Preql instance
 
         Parameters:
@@ -137,6 +139,7 @@ def __init__(self, db_uri: str='sqlite://:memory:', print_sql: bool=settings.pri
         self._print_sql = print_sql
         self._auto_create = auto_create
         self._display = display.RichDisplay()
+        self._autocommit = autocommit
         # self.engine.ping()
 
         engine = create_engine(self._db_uri, print_sql=self._print_sql, auto_create=auto_create)
@@ -160,7 +163,7 @@ def set_output_format(self, fmt):
     def _reset_interpreter(self, engine=None):
         if engine is None:
             engine = self._interp.state.db
-        self._interp = Interpreter(engine, self._display, _preql_inst=self)
+        self._interp = Interpreter(engine, self._display, _preql_inst=self, autocommit=self._autocommit)
 
     def close(self):
         self._interp.state.db.close()

preql/core/autocomplete.py

@@ -31,6 +31,10 @@ def eval_autocomplete(t: ast.Try, go_inside):
     with use_scope(scope):
         eval_autocomplete(t.catch_block, go_inside)
 
+@dsp
+def eval_autocomplete(t: ast.Transaction, go_inside):
+    eval_autocomplete(t.do, go_inside)
+
 @dsp
 def eval_autocomplete(a: ast.InsertRows, go_inside):
     eval_autocomplete(a.value, go_inside)

preql/core/evaluate.py

@@ -102,12 +102,17 @@ def resolve(type_: ast.Type):
 
 
 
-def db_query(sql_code, subqueries=None):
+def db_query(sql_code, subqueries=None, *, modifies=True):
     try:
-        return get_db().query(sql_code, subqueries)
+        res = get_db().query(sql_code, subqueries)
     except exc.DatabaseQueryError as e:
         raise Signal.make(T.DbQueryError, None, e.args[0]) from e
 
+    if modifies and context.state.autocommit:
+        get_db().commit()
+
+    return res
+
 def drop_table(table_type):
     name = table_type.options['name']
     code = sql.compile_drop_table(name)
@@ -346,6 +351,19 @@ def _execute(f: ast.For):
         with use_scope({f.var: objects.from_python(i)}):
             execute(f.do)
 
+@method
+def _execute(t: ast.Transaction):
+    db_query(sql.BeginTransaction())
+    with context(state=context.state.set_autocommit(False)):
+        try:
+            res = execute(t.do)
+        except Signal as e:
+            get_db().rollback()
+            raise
+
+    get_db().commit()
+    return res
+
 @method
 def _execute(t: ast.Try):
     try:
@@ -410,7 +428,8 @@ def _execute(t: ast.Throw):
     e = evaluate(t.value)
     if isinstance(e, ast.Ast):
         raise exc.InsufficientAccessLevel()
-    assert isinstance(e, Exception), e
+    if not isinstance(e, Exception):
+        raise Signal.make(T.TypeError, t, f"Can only throw an exception, not {e.type}")
     raise e
 
 def execute(stmt):
@@ -870,7 +889,7 @@ def _new_row(new_ast, table, matched):
         raise Signal.make(T.TypeError, new_ast, f"'new' expects a persistent table. Instead got a table expression.")
 
     if get_db().target == sql.bigquery:
-        rowid = db_query(sql.FuncCall(T.string, 'GENERATE_UUID', []))
+        rowid = db_query(sql.FuncCall(T.string, 'GENERATE_UUID', []), modifies=False)
         keys += ['id']
         values += [sql.make_value(rowid)]
     elif get_db().target == sql.snowflake:
@@ -885,7 +904,7 @@ def _new_row(new_ast, table, matched):
     db_query(q)
 
     if get_db().target not in (sql.bigquery, sql.snowflake):
-        rowid = db_query(sql.LastRowId())
+        rowid = db_query(sql.LastRowId(), modifies=False)
 
     d = SafeDict({'id': objects.pyvalue_inst(rowid)})
     d.update({p.name:v for p, v in matched})
@@ -1015,7 +1034,7 @@ def localize(inst: objects.Instance):
     if inst.code is sql.null:
         return None
 
-    return db_query(inst.code, inst.subqueries)
+    return db_query(inst.code, inst.subqueries, modifies=False)
 
 @method
 def localize(inst: objects.ValueInstance):

preql/core/interpreter.py

@@ -49,11 +49,11 @@ def __getattr__(self, attr):
 
 
 class Interpreter:
-    def __init__(self, sqlengine, display, use_core=True, _preql_inst=None):
+    def __init__(self, sqlengine, display, use_core=True, _preql_inst=None, autocommit=False):
         assert _preql_inst
         self._preql_inst = _preql_inst  # XXX temporary hack
 
-        self.state = ThreadState.from_components(self, sqlengine, display, initial_namespace())
+        self.state = ThreadState.from_components(self, sqlengine, display, initial_namespace(), autocommit=autocommit)
         if use_core:
             mns = import_module(self.state, ast.Import('__builtins__', use_core=False)).namespace
             bns = self.state.get_var('__builtins__').namespace

preql/core/parser.py

@@ -278,6 +278,7 @@ def name_path(self, path, name):
     while_stmt = ast.While
     for_stmt = ast.For
     try_catch = ast.Try
+    transaction = ast.Transaction
     one = lambda self, nullable, expr: ast.One(expr, nullable is not None)
 
     def marker(self, _marker):

preql/core/pql_ast.py

@@ -290,6 +290,10 @@ class Try(Statement):
     catch_expr: Expr
     catch_block: CodeBlock
 
+@dataclass
+class Transaction(Statement):
+    do: CodeBlock
+
 @dataclass
 class If(Statement):
     cond: Object

preql/core/preql.lark

@@ -13,6 +13,7 @@ module: _NL? stmt+   -> as_list
      | import_stmt
      | throw
      | try_catch
+     | transaction
      | expr _NL
 
 pql_dict: "{" _ml_sep? ml_list{named_expr} _ml_sep? "}"
@@ -52,6 +53,7 @@ func_def: "func" name func_params "=" expr [_NL string_raw] _NL -> func_def_shor
         | "func" name func_params codeblock _NL
 
 try_catch: "try" codeblock _NL? "catch" "(" [name ":"] expr ")" codeblock _NL
+transaction: "transaction" codeblock _NL
 if_stmt: "if" "(" expr ")" codeblock (_NL? "else" (codeblock _NL|if_stmt) | _NL)
 for_stmt: "for" "(" name "in" expr ")" codeblock _NL
 while_stmt: "while" "(" expr ")" codeblock _NL

preql/core/sql.py

@@ -490,6 +490,13 @@ def _compile(self, qb):
 class SqlStatement(SqlTree):
     type = T.nulltype
 
+@dataclass
+class BeginTransaction(SqlStatement):
+    def _compile(self, qb):
+        if qb.target == mysql:
+            return ["START TRANSACTION"]
+        return ["BEGIN TRANSACTION"]
+
 @dataclass
 class AddIndex(SqlStatement):
     index_name: Id

preql/core/state.py

@@ -127,18 +127,19 @@ def unique_name(self, obj):
 
 
 class ThreadState:
-    def __init__(self, state, ns=None):
+    def __init__(self, state, ns=None, *, autocommit=False):
         self.state = state
         # Add logger?
 
         self.ns = Namespace(ns)
         self.access_level = AccessLevels.WRITE_DB
         self.stacktrace = []
+        self.autocommit = autocommit
 
     @classmethod
-    def from_components(cls, interp, db, display, ns=None):
+    def from_components(cls, interp, db, display, ns=None, *, autocommit):
         state = State(interp, db, display)
-        return cls(state, ns)
+        return cls(state, ns, autocommit=autocommit)
 
     @property
     def interp(self):
@@ -159,6 +160,7 @@ def clone(cls, inst):
         s.ns = copy(inst.ns)
         s.access_level = inst.access_level
         s.stacktrace = copy(inst.stacktrace)
+        s.autocommit = inst.autocommit
         return s
 
 
@@ -171,9 +173,15 @@ def reduce_access(self, new_level):
         s.access_level = new_level
         return s
 
+    def set_autocommit(self, autocommit):
+        s = copy(self)
+        s.autocommit = autocommit
+        return s
+
     def require_access(self, level):
         if self.access_level < level:
             raise InsufficientAccessLevel(level)
+
     def catch_access(self, level):
         if self.access_level < level:
             raise Exception("Bad access. Security risk.")

tests/test_basic.py

@@ -1564,7 +1564,50 @@ def test_threading(self):
         if p._interp.state.db.target != mysql:   # Not supported
             assert p('a{item} - [..100]') == []
 
+    @uses_tables("a")
+    def test_transaction1(self):
+        p = self.Preql()
+        p('''
+            table a {
+                x: int
+            }
+
+            try{
+                transaction {
+                    new a(4)
+                    throw new Exception("Some Error")
+                }
+            } catch(Exception) {
+            }
+
+            table a {
+                x: int
+            }
+        ''')
+
+        assert not p.a
+
+    def test_transaction2(self):
+        p = self.Preql()
+        p.rollback()
+        p('''
+            table a {
+                x: int
+            }
+
+            transaction {
+                try{
+                    new a(5)
+                    throw new Exception("A")
+                } catch(Exception) {
+                }
+            }
+
+        ''')
 
+        self.assertEqual( p(r'list(a{x})'), [5] )
+        p.run_statement('DROP TABLE a')
+        p.commit()
 
 class TestFlow(PreqlTests):
     def test_new_freezes_values(self):