Align default rendered SQL masking contract

.gitattributes

@@ -0,0 +1 @@
+test-vectors/*.tsv whitespace=-blank-at-eol

examples/conformance/app/main.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 from pathlib import Path
 import sys
 
@@ -12,9 +13,11 @@
 from teaql.data_service import SQLiteTeaQLClient
 from teaql.core import EntityKey, EntityRoot
 from teaql.runtime import CheckException, UserContext
+from app.masking_lifecycle import verify_masking_lifecycle
 
 
 async def main() -> None:
+    await verify_masking_lifecycle()
     order_key = EntityKey("Order", 1)
     execution_key = EntityKey("InferenceExecution", 1)
     target_ledger = EntityRoot()
@@ -27,7 +30,7 @@ async def main() -> None:
     assert target_ledger.original_version(execution_key) == 9
     print("PASS Mutation ledger identity (same ID, different entity types keep versions 3/9)")
 
-    database = ROOT / ".local" / "conformance.sqlite"
+    database = Path(os.environ.get("TEAQL_CONFORMANCE_DB", ROOT / ".local" / "conformance.sqlite"))
     database.parent.mkdir(parents=True, exist_ok=True)
     database.unlink(missing_ok=True)
     client = SQLiteTeaQLClient(str(database))

examples/conformance/app/masking_lifecycle.py

@@ -0,0 +1,148 @@
+"""Runtime-owned fixture; does not modify generated domain-library code."""
+from contextlib import aclosing
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from types import SimpleNamespace
+
+from teaql.core.expr import Expr
+from teaql.core.meta import EntityDescriptor, PropertyDescriptor, RelationDescriptor
+from teaql.core.mutation import InsertCommand, MutationRequest, TraceNode
+from teaql.core.query import SelectQuery
+from teaql.core.value import DataType
+from teaql.data_service import QueryRequest
+from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service
+from teaql.runtime import RuntimeModule
+from teaql.runtime.context import TextDiagnosticSqlLogSink
+from teaql.sql.executor import TransportError
+from teaql.sql.types import CompiledQuery
+
+
+async def verify_masking_lifecycle():
+    with TemporaryDirectory(prefix='teaql-mask-lifecycle-') as directory:
+        entity = (EntityDescriptor('MaskCustomer').table_name('mask_customer_data')
+                  .property(PropertyDescriptor('id', DataType.I64).is_id())
+                  .property(PropertyDescriptor('version', DataType.I64).is_version())
+                  .property(PropertyDescriptor('display_name', DataType.Text))
+                  .audit_mask_fields(['display_name']))
+        entity.relation(RelationDescriptor('children', 'MaskChild').foreign('parent_id').many())
+        child = (EntityDescriptor('MaskChild').table_name('mask_child_data')
+                 .property(PropertyDescriptor('id', DataType.I64).is_id())
+                 .property(PropertyDescriptor('version', DataType.I64).is_version())
+                 .property(PropertyDescriptor('parent_id', DataType.I64)))
+        provider = SimpleSchemaProvider()
+        provider.register_entity(entity)
+        provider.register_entity(child)
+        service = create_sqlite_service(str(Path(directory) / 'mask.sqlite'), provider)
+        context = RuntimeModule.new().entity(entity).entity(child).into_context().with_schema_provider(service)
+        await context.ensure_schema()
+        output, entries = [], []
+        log_file = Path(directory) / 'sql.log'
+        def write_line(text):
+            output.append(text)
+            with log_file.open('a', encoding='utf-8') as destination:
+                destination.write(text + '\n')
+        sink = TextDiagnosticSqlLogSink(write_line)
+        def capture(entry):
+            entries.append(entry)
+            sink.write(entry)
+        context.set_diagnostic_sql_log_sink(SimpleNamespace(write=capture))
+        async def insert(entity_id, target=None):
+            command = (InsertCommand('MaskCustomer').value('id', entity_id).value('version', 1)
+                       .value('display_name', 'Riverside'))
+            command.trace_chain = [TraceNode(comment='what: seed Riverside lifecycle fixture')]
+            return await (target or service).mutate(context, MutationRequest(command))
+        try:
+            for entity_id in [1, 2, 3]:
+                await insert(entity_id)
+            entries.clear()
+            output.clear()
+            request = QueryRequest(SelectQuery('MaskCustomer')
+                .filter(Expr.eq('display_name', 'Riverside')).limit(3)
+            ).comment('what: inspect customers').purpose('why: verify stream lifecycle')
+            # async-for break alone does not promise immediate generator close
+            # in Python. aclosing makes ownership explicit and deterministic.
+            async with aclosing(service.query_stream(context, request, 1)) as stream:
+                async for chunk in stream:
+                    assert chunk.rows[0]['display_name'] == 'Riverside'
+                    break
+            assert len(entries) == 1 and entries[0].execution_outcome == 'cancelled'
+            assert entries[0].result_count == 1
+            try:
+                await insert(1)
+                raise AssertionError('duplicate key unexpectedly succeeded')
+            except TransportError:
+                pass
+            assert entries[-1].execution_outcome == 'failure'
+            assert entries[-1].affected_rows is None
+            result = await service.query(context, request)
+            assert len(result.rows) == 3
+            assert all(row['display_name'] == 'Riverside' for row in result.rows)
+            assert entries[-1].execution_outcome == 'success'
+            text = '\n'.join(output)
+            assert 'Riverside' not in text and 'Ri*****de' in text
+            assert 'what: inspect customers' in text and 'why: verify stream lifecycle' in text
+
+            await service.transport.execute_sql(CompiledQuery(
+                'CREATE TRIGGER remove_mask_probe AFTER INSERT ON mask_customer_data '
+                'WHEN NEW.id = 777 BEGIN DELETE FROM mask_customer_data WHERE id = NEW.id; END', []))
+            entries.clear()
+            output.clear()
+            try:
+                await insert(777)
+                raise AssertionError('missing snapshot unexpectedly succeeded')
+            except TransportError:
+                pass
+            assert len(entries) == 2
+            assert entries[0].execution_outcome == 'success' and entries[0].affected_rows == 1
+            assert entries[1].execution_outcome == 'success' and entries[1].result_count == 0
+            assert 'what: seed' in entries[1].audit_reason
+            assert 'Riverside' not in repr(entries) and 'Riverside' not in '\n'.join(output)
+
+            context.insert_resource('dataService', service)
+            entries.clear()
+            async def partial_graph():
+                target = context.require_resource('dataService')
+                await insert(30, target)
+                await insert(777, target)
+                await insert(31, target)
+            try:
+                await context.execute_graph_save(partial_graph)
+                raise AssertionError('partial graph unexpectedly committed')
+            except TransportError:
+                pass
+            assert [entry.execution_outcome for entry in entries] == ['success','success','success']
+            assert entries[-1].result_count == 0
+            assert not await service.transport.fetch_all_sql(CompiledQuery(
+                'SELECT id FROM mask_customer_data WHERE id IN (30,31,777)', []))
+            assert 'Riverside' not in repr(entries)
+            await insert(32)
+            assert len(await service.transport.fetch_all_sql(CompiledQuery('SELECT id FROM mask_customer_data', []))) == 4
+
+            command = (InsertCommand('MaskChild').value('id', 1).value('version', 1).value('parent_id', 1))
+            command.trace_chain = [TraceNode(comment='what: seed child for relation verification')]
+            await service.mutate(context, MutationRequest(command))
+            graph_request = QueryRequest(SelectQuery('MaskCustomer').project('id')
+                .filter(Expr.eq('id', 1)).and_filter(Expr.eq('display_name', 'Riverside'))
+                .relation_query('children', SelectQuery('MaskChild').project('id').limit(2)).limit(1)
+            ).comment('what: load Riverside graph').purpose('why: verify inherited relation masking')
+            entries.clear()
+            await service.transport.execute_sql(CompiledQuery(
+                'ALTER TABLE mask_child_data RENAME TO mask_child_unavailable', []))
+            try:
+                try:
+                    await service.query(context, graph_request)
+                    raise AssertionError('missing relation table unexpectedly succeeded')
+                except TransportError:
+                    pass
+            finally:
+                await service.transport.execute_sql(CompiledQuery(
+                    'ALTER TABLE mask_child_unavailable RENAME TO mask_child_data', []))
+            assert [entry.execution_outcome for entry in entries] == ['success', 'failure']
+            assert 'what: load' in entries[-1].comment
+            assert 'mask_child_data' in entries[-1].debug_sql
+            assert 'Riverside' not in repr(entries) + log_file.read_text(encoding='utf-8')
+            restored = await service.query(context, graph_request)
+            assert restored.rows[0]['children'][0]['id'] == 1
+        finally:
+            await service.close()
+    print('PASS Python masked SQL lifecycle: stream, readback intent, SQLite failure, partial graph rollback, relation file/custom sinks and reuse')

examples/conformance/models/platform.py

@@ -246,4 +246,4 @@ def update_version(self, value):
         return self
     def work_item_list(self) -> list:
         self._loaded_fields.add("work_item_list")
-        return self._work_item_list
+        return self._work_item_list

examples/conformance/models/work_item.py

@@ -258,4 +258,3 @@ def update_platform(self, value):
         self._loaded_fields.add("platform")
         self._entity_root.set(self._teaql_entity_key(), "platform", Value.from_any(self.platform))
         return self
-

examples/conformance/pyproject.toml

@@ -2,7 +2,7 @@
 name = "runtime-example-conformance-service-lib"
 version = "1.0.0"
 description = "Generated python library"
-dependencies = ["teaql==0.2.5", "aiosqlite>=0.22.1"]
+dependencies = ["teaql==0.2.7", "aiosqlite>=0.22.1"]
 
 [tool.setuptools]
 py-modules = ["Q", "E"]

examples/conformance/requests/platform_request.py

@@ -1,4 +1,4 @@
-from teaql.core.query import SelectQuery
+from teaql.core.query import RelationAggregate, SelectQuery
 from teaql.core.list import SmartList, TeaQLPage
 from teaql.runtime import EntityRoot
 from teaql.data_service import QueryRequest
@@ -65,15 +65,11 @@ def offset(self, n: int):
         return self
 
     def with_deleted_rows(self):
-        self.query._filters = [
-            expression for expression in self.query._filters
-            if expression.get("field") != "version"
-        ]
+        self.query.with_deleted_rows()
         return self
 
     def deleted_rows_only(self):
-        self.with_deleted_rows()
-        self.query.and_filter(lte("version", -1))
+        self.query.deleted_rows_only()
         return self
 
     def select_self_fields(self):
@@ -290,21 +286,21 @@ def group_by_id(self):
         return self
 
     def group_by_id_as(self, ret_name: str):
-        self.query.group_by("id") 
+        self.query.group_by("id")
         return self
     def group_by_name(self):
         self.query.group_by("name")
         return self
 
     def group_by_name_as(self, ret_name: str):
-        self.query.group_by("name") 
+        self.query.group_by("name")
         return self
     def group_by_version(self):
         self.query.group_by("version")
         return self
 
     def group_by_version_as(self, ret_name: str):
-        self.query.group_by("version") 
+        self.query.group_by("version")
         return self
     def select_work_item_list(self):
         from requests.work_item_request import WorkItemRequest
@@ -322,13 +318,13 @@ def have_no_work_items(self):
         return self.without_work_item_list_matching(WorkItemRequest())
 
     def with_work_item_list_matching(self, child_request):
+        child_request.query.projection = ["platform"]
         self.query.and_filter(in_subquery(column("id"), "WorkItem", child_request.query))
-        child_request.query._projection = ["platform"]
         return self
 
     def without_work_item_list_matching(self, child_request):
+        child_request.query.projection = ["platform"]
         self.query.and_filter(not_in_subquery(column("id"), "WorkItem", child_request.query))
-        child_request.query._projection = ["platform"]
         return self
     def count_work_items(self):
         return self.count_work_items_as("count_work_items")
@@ -339,7 +335,9 @@ def count_work_items_as(self, alias: str):
 
     def count_work_items_with(self, alias: str, child_request):
         child_request.query.count_field("id", alias)
-        self.query.relation_aggregate("work_item_list", alias, child_request.query, True)
+        self.query.relation_aggregates.append(
+            RelationAggregate("work_item_list", alias, child_request.query, True)
+        )
         return self
 
 
@@ -366,7 +364,7 @@ async def execute_for_result(self, context):
         if not self._purpose or not self._purpose.strip() or not self._comment or not self._comment.strip():
             raise Exception("Security audit failure: comment() and purpose() must be called before execute_for_rows()")
         service = context.require_resource("dataService")
-        req = QueryRequest(context.prepare_query(self.query))
+        req = QueryRequest(context.prepare_query(self.query), _comment=self._comment, _purpose=self._purpose)
         return await service.query(context, req)
 
     async def execute_for_rows(self, context):
@@ -388,21 +386,21 @@ async def execute_for_page(self, context, offset: int, limit: int) -> TeaQLPage[
         service = context.require_resource("dataService")
         alias = "__teaql_total"
         if authorized.id_set_pagination is not None:
-            row_result = await service.query(context, QueryRequest(authorized))
+            row_result = await service.query(context, QueryRequest(authorized, _comment=request._comment, _purpose=request._purpose))
             retained_count, accuracy = context.id_set_count()
             if accuracy == "EXACT":
                 total_count = retained_count
             else:
-                count_result = await service.query(context, QueryRequest(authorized.for_exact_count(alias)))
+                count_result = await service.query(context, QueryRequest(authorized.for_exact_count(alias), _comment=request._comment, _purpose=request._purpose))
                 if not count_result.rows or not isinstance(count_result.rows[0].get(alias), (int, float)):
                     raise RuntimeError("dataService did not return an exact page count")
                 total_count = int(count_result.rows[0][alias])
         else:
-            count_result = await service.query(context, QueryRequest(authorized.for_exact_count(alias)))
+            count_result = await service.query(context, QueryRequest(authorized.for_exact_count(alias), _comment=request._comment, _purpose=request._purpose))
             if not count_result.rows or not isinstance(count_result.rows[0].get(alias), (int, float)):
                 raise RuntimeError("dataService did not return an exact page count")
             total_count = int(count_result.rows[0][alias])
-            row_result = await service.query(context, QueryRequest(authorized))
+            row_result = await service.query(context, QueryRequest(authorized, _comment=request._comment, _purpose=request._purpose))
         query_root = EntityRoot()
         data = SmartList(Platform(_entity_root=query_root, **row) for row in row_result.rows)
         return TeaQLPage(data=data, total_count=total_count, offset=offset, limit=limit)
@@ -421,6 +419,6 @@ async def execute_for_stream(self, context, chunk_size: int = 1000):
         if not hasattr(service, "query_stream"):
             raise RuntimeError("dataService does not implement query_stream")
         query_root = EntityRoot()
-        async for chunk in service.query_stream(context, QueryRequest(request.query), chunk_size):
+        async for chunk in service.query_stream(context, QueryRequest(context.prepare_query(request.query), _comment=request._comment, _purpose=request._purpose), chunk_size):
             for row in chunk.rows:
                 yield Platform(_entity_root=query_root, **row)