Merge pull request #446 from skalish/cat-ops

TC: Add functions to run Categorization and Schema Mapping project workflows

CHANGELOG.md

@@ -12,6 +12,7 @@
   - [#443](https://github.com/Datatamer/tamr-client/pull/443) Added function to materialize datasets.
   - [#445](https://github.com/Datatamer/tamr-client/pull/445) Added functions for getting projects and datasets by name via `tc.project.by_name` and `tc.dataset.by_name`
     - Renamed functions `from_resource_id` to `by_resource_id` in `tc.attribute`, `tc.dataset`, `tc.operation`, and `tc.project`
+  - [#446](https://github.com/Datatamer/tamr-client/pull/446) Added functions for categorization workflow operations in `tc.categorization` and schema mapping workflow operations in `tc.schema_mapping`
 
   **NEW FEATURES**
   - [#383](https://github.com/Datatamer/tamr-client/issues/383) Now able to create an Operation from Job resource id

docs/beta/categorization.md

@@ -1,3 +1,4 @@
 # Categorization
 
+  * [Categorization](/beta/categorization/categorization)
   * [Project](/beta/categorization/project)

docs/beta/categorization/categorization.rst

@@ -0,0 +1,7 @@
+Categorization
+==============
+
+.. autofunction:: tamr_client.categorization.update_unified_dataset
+.. autofunction:: tamr_client.categorization.apply_feedback
+.. autofunction:: tamr_client.categorization.update_results
+.. autofunction:: tamr_client.categorization.manual_labels

docs/beta/categorization/project.rst

@@ -3,5 +3,4 @@ Categorization Project
 
 .. autoclass:: tamr_client.CategorizationProject
 
-.. autofunction:: tamr_client.categorization.project.create
-.. autofunction:: tamr_client.categorization.project.manual_labels
+.. autofunction:: tamr_client.categorization.project.create

docs/beta/schema_mapping.md

@@ -1,3 +1,4 @@
 # Schema Mapping
 
+  * [Schema Mapping](/beta/schema_mapping/schema_mapping)
   * [Project](/beta/schema_mapping/project)

docs/beta/schema_mapping/schema_mapping.rst

@@ -0,0 +1,4 @@
+Schema Mapping
+==============
+
+.. autofunction:: tamr_client.schema_mapping.update_unified_dataset

tamr_client/categorization/__init__.py

@@ -3,3 +3,11 @@
 See https://docs.tamr.com/docs/overall-workflow-classification
 """
 from tamr_client.categorization import project
+from tamr_client.categorization._categorization import (
+    _apply_feedback_async,
+    _update_results_async,
+    apply_feedback,
+    manual_labels,
+    update_results,
+    update_unified_dataset,
+)

tamr_client/categorization/_categorization.py

@@ -0,0 +1,81 @@
+"""
+Tamr - Categorization
+See https://docs.tamr.com/docs/overall-workflow-classification
+
+The terminology used here is consistent with Tamr UI terminology
+
+Asynchronous versions of each function can be found with the suffix `_async` and may be of
+interest to power users
+"""
+from tamr_client import operation
+from tamr_client._types import CategorizationProject, Dataset, Operation, Session
+from tamr_client.dataset import _dataset, unified
+
+
+def manual_labels(session: Session, project: CategorizationProject) -> Dataset:
+    """Get manual labels from a Categorization project.
+
+    Args:
+        project: Tamr project containing labels
+
+    Returns:
+        Dataset containing manual labels
+
+    Raises:
+        dataset.NotFound: If no dataset could be found at the specified URL
+        dataset.Ambiguous: If multiple targets match dataset name
+    """
+    unified_dataset = unified.from_project(session=session, project=project)
+    labels_dataset_name = unified_dataset.name + "_manual_categorizations"
+    return _dataset.by_name(
+        session=session, instance=project.url.instance, name=labels_dataset_name
+    )
+
+
+def update_unified_dataset(
+    session: Session, project: CategorizationProject
+) -> Operation:
+    """Apply changes to the unified dataset and wait for the operation to complete
+
+    Args:
+        project: Tamr Categorization project
+    """
+    unified_dataset = unified.from_project(session, project)
+    op = unified._apply_changes_async(session, unified_dataset)
+    return operation.wait(session, op)
+
+
+def apply_feedback(session: Session, project: CategorizationProject) -> Operation:
+    """Train the categorization model according to verified labels and wait for the
+    operation to complete
+
+    Args:
+        project: Tamr Categorization project
+    """
+    op = _apply_feedback_async(session, project)
+    return operation.wait(session, op)
+
+
+def update_results(session: Session, project: CategorizationProject) -> Operation:
+    """Generate classifications based on the latest categorization model and wait for the
+    operation to complete
+
+    Args:
+        project: Tamr Categorization project
+    """
+    op = _update_results_async(session, project)
+    return operation.wait(session, op)
+
+
+def _apply_feedback_async(
+    session: Session, project: CategorizationProject
+) -> Operation:
+    r = session.post(str(project.url) + "/categorizations/model:refresh")
+    return operation._from_response(project.url.instance, r)
+
+
+def _update_results_async(
+    session: Session, project: CategorizationProject
+) -> Operation:
+    r = session.post(str(project.url) + "/categorizations:refresh")
+    return operation._from_response(project.url.instance, r)

tamr_client/categorization/project.py

@@ -3,14 +3,12 @@
 from tamr_client import project
 from tamr_client._types import (
     CategorizationProject,
-    Dataset,
     Instance,
     JsonDict,
     Project,
     Session,
     URL,
 )
-from tamr_client.dataset import _dataset, unified
 
 
 def _from_json(url: URL, data: JsonDict) -> CategorizationProject:
@@ -58,23 +56,3 @@ def create(
         external_id=external_id,
         unified_dataset_name=unified_dataset_name,
     )
-
-
-def manual_labels(session: Session, project: CategorizationProject) -> Dataset:
-    """Get manual labels from a Categorization project.
-
-    Args:
-        project: Tamr project containing labels
-
-    Returns:
-        Dataset containing manual labels
-
-    Raises:
-        dataset.NotFound: If no dataset could be found at the specified URL
-        dataset.Ambiguous: If multiple targets match dataset name
-    """
-    unified_dataset = unified.from_project(session=session, project=project)
-    labels_dataset_name = unified_dataset.name + "_manual_categorizations"
-    return _dataset.by_name(
-        session=session, instance=project.url.instance, name=labels_dataset_name
-    )

tamr_client/mastering/__init__.py

@@ -11,7 +11,6 @@
     _update_cluster_results_async,
     _update_high_impact_pairs_async,
     _update_pair_results_async,
-    _update_unified_dataset_async,
     apply_feedback,
     estimate_pairs,
     generate_pairs,

tamr_client/mastering/_mastering.py

@@ -18,7 +18,8 @@ def update_unified_dataset(session: Session, project: MasteringProject) -> Opera
     Args:
         project: Tamr Mastering project
     """
-    op = _update_unified_dataset_async(session, project)
+    unified_dataset = unified.from_project(session, project)
+    op = unified._apply_changes_async(session, unified_dataset)
     return operation.wait(session, op)
 
 
@@ -97,13 +98,6 @@ def publish_clusters(session: Session, project: MasteringProject) -> Operation:
     return operation.wait(session, op)
 
 
-def _update_unified_dataset_async(
-    session: Session, project: MasteringProject
-) -> Operation:
-    unified_dataset = unified.from_project(session, project)
-    return unified._apply_changes_async(session, unified_dataset)
-
-
 def _estimate_pairs_async(session: Session, project: MasteringProject) -> Operation:
     r = session.post(str(project.url) + "/estimatedPairCounts:refresh")
     return operation._from_response(project.url.instance, r)

tamr_client/schema_mapping/__init__.py

@@ -3,3 +3,4 @@
 See https://docs.tamr.com/new/docs/overall-workflow-schema
 """
 from tamr_client.schema_mapping import project
+from tamr_client.schema_mapping._schema_mapping import update_unified_dataset

tamr_client/schema_mapping/_schema_mapping.py

@@ -0,0 +1,25 @@
+"""
+Tamr - Schema Mapping
+See https://docs.tamr.com/new/docs/overall-workflow-schema
+
+The terminology used here is consistent with Tamr UI terminology
+
+Asynchronous versions of each function can be found with the suffix `_async` and may be of
+interest to power users
+"""
+from tamr_client import operation
+from tamr_client._types import Operation, SchemaMappingProject, Session
+from tamr_client.dataset import unified
+
+
+def update_unified_dataset(
+    session: Session, project: SchemaMappingProject
+) -> Operation:
+    """Apply changes to the unified dataset and wait for the operation to complete
+
+    Args:
+        project: Tamr Schema Mapping project
+    """
+    unified_dataset = unified.from_project(session, project)
+    op = unified._apply_changes_async(session, unified_dataset)
+    return operation.wait(session, op)

tests/tamr_client/categorization/test_categorization.py

@@ -0,0 +1,42 @@
+import tamr_client as tc
+from tests.tamr_client import fake
+
+
+@fake.json
+def test_manual_labels():
+    s = fake.session()
+    project = fake.categorization_project()
+
+    tc.categorization.manual_labels(session=s, project=project)
+
+
+@fake.json
+def test_apply_feedback_async():
+    s = fake.session()
+    project = fake.categorization_project()
+
+    op = tc.categorization._apply_feedback_async(s, project)
+    assert op.type == "SPARK"
+    assert op.description == "Materialize views to Elastic"
+    assert op.status == {
+        "state": "PENDING",
+        "startTime": "",
+        "endTime": "",
+        "message": "Job has not yet been submitted to Spark",
+    }
+
+
+@fake.json
+def test_update_results_async():
+    s = fake.session()
+    project = fake.categorization_project()
+
+    op = tc.categorization._update_results_async(s, project)
+    assert op.type == "SPARK"
+    assert op.description == "Materialize views to Elastic"
+    assert op.status == {
+        "state": "PENDING",
+        "startTime": "",
+        "endTime": "",
+        "message": "Job has not yet been submitted to Spark",
+    }

tests/tamr_client/fake_json/categorization/test_categorization/test_apply_feedback_async.json

@@ -0,0 +1,33 @@
+[
+    {
+        "request": {
+            "method": "POST",
+            "path": "projects/2/categorizations/model:refresh"
+        },
+        "response": {
+            "status": 200,
+            "json": {
+                "id": "1",
+                "type": "SPARK",
+                "description": "Materialize views to Elastic",
+                "status": {
+                    "state": "PENDING",
+                    "startTime": "",
+                    "endTime": "",
+                    "message": "Job has not yet been submitted to Spark"
+                },
+                "created": {
+                    "username": "admin",
+                    "time": "2020-06-12T18:21:42.288Z",
+                    "version": "operation 1 created version"
+                },
+                "lastModified": {
+                    "username": "admin",
+                    "time": "2020-06-12T18:21:42.288Z",
+                    "version": "operation 1 modified version"
+                },
+                "relativeId": "operations/1"
+            }
+        }
+    }
+]

tests/tamr_client/fake_json/categorization/test_categorization/test_update_results_async.json

@@ -0,0 +1,33 @@
+[
+    {
+        "request": {
+            "method": "POST",
+            "path": "projects/2/categorizations:refresh"
+        },
+        "response": {
+            "status": 200,
+            "json": {
+                "id": "1",
+                "type": "SPARK",
+                "description": "Materialize views to Elastic",
+                "status": {
+                    "state": "PENDING",
+                    "startTime": "",
+                    "endTime": "",
+                    "message": "Job has not yet been submitted to Spark"
+                },
+                "created": {
+                    "username": "admin",
+                    "time": "2020-06-12T18:21:42.288Z",
+                    "version": "operation 1 created version"
+                },
+                "lastModified": {
+                    "username": "admin",
+                    "time": "2020-06-12T18:21:42.288Z",
+                    "version": "operation 1 modified version"
+                },
+                "relativeId": "operations/1"
+            }
+        }
+    }
+]