gergo/queryIngestionFix (#479)

* feat: use mise for docs build

* fix: getting the ingestion query needs to use model ingestion id

.github/workflows/pr.yml

@@ -59,6 +59,8 @@ jobs:
   test-public: # Run integration tests against the public server image
     name: Test (public)
     runs-on: ubuntu-latest
+    env:
+      IS_PUBLIC: "true"
     strategy:
       matrix:
         python-version:

codebook.toml

@@ -0,0 +1 @@
+words = ["specklepy"]

mise.toml

@@ -8,7 +8,7 @@ python.uv_venv_auto = true
 
 
 [tasks.install]
-run= "uv sync --all-groups"
+run= "uv sync --all-extras --all-groups"
 
 
 [tasks.install_docs]
@@ -18,3 +18,10 @@ run= "uv sync --group docs"
 description = "Build static docs "
 run = "uv run mkdocs build"
 depends = ['install_docs']
+
+[tasks.test]
+run = "uv run pytest"
+
+
+[env]
+IS_PUBLIC = "false"

pyproject.toml

@@ -22,6 +22,7 @@ dependencies = [
 speckleifc = ["ifcopenshell>=0.8.3.post2"]
 
 [dependency-groups]
+
 dev = [
     "commitizen>=4.1.0",
     "devtools>=0.12.2",
@@ -32,6 +33,7 @@ dev = [
     "pytest-asyncio>=0.25.2",
     "pytest-cov>=6.0.0",
     "pytest-ordering>=0.6",
+    "pytest_httpserver >=1.1.3",
     "ruff==0.9.2",
     "types-deprecated>=1.2.15.20241117",
     "types-requests>=2.32.0.20241016",

src/speckle_automate/automation_context.py

@@ -19,8 +19,12 @@ from speckle_automate.schema import (
 from specklepy.api import operations
 from specklepy.api.client import SpeckleClient
 from specklepy.core.api.inputs.model_inputs import CreateModelInput
-from specklepy.core.api.inputs.version_inputs import CreateVersionInput
+from specklepy.core.api.inputs.version_inputs import (
+    CreateVersionInput,
+    MarkReceivedVersionInput,
+)
 from specklepy.core.api.models.current import Model, Version
+from specklepy.logging import metrics
 from specklepy.logging.exceptions import SpeckleException
 from specklepy.objects.base import Base
 from specklepy.transports.memory import MemoryTransport
@@ -66,6 +70,7 @@ class AutomationContext:
             if isinstance(automation_run_data, AutomationRunData)
             else AutomationRunData.model_validate_json(automation_run_data)
         )
+        metrics.set_host_app("automate")
         speckle_client = SpeckleClient(
             automation_run_data.speckle_server_url,
             automation_run_data.speckle_server_url.startswith("https"),
@@ -100,6 +105,7 @@ class AutomationContext:
         """Receive the Speckle project version that triggered this automation run."""
         # TODO: this is a quick hack to keep implementation consistency.
         # Move to proper receive many versions
+        project_id = self.automation_run_data.project_id
         version_id = self.automation_run_data.triggers[0].payload.version_id
         try:
             version = self.speckle_client.version.get(
@@ -109,7 +115,7 @@ class AutomationContext:
             raise ValueError(
                 f"""Could not receive specified version.
                 Is your environment configured correctly?
-                project_id: {self.automation_run_data.project_id}
+                project_id: {project_id}
                 model_id: {self.automation_run_data.triggers[0].payload.model_id}
                 version_id: {self.automation_run_data.triggers[0].payload.version_id}
                 """
@@ -124,6 +130,13 @@ class AutomationContext:
         base = operations.receive(
             version.referenced_object, self._server_transport, self._memory_transport
         )
+        self.speckle_client.version.received(
+            MarkReceivedVersionInput(
+                version_id=version_id,
+                project_id=project_id,
+                source_application="automate_function",
+            )
+        )
         # self._closure_tree = base["__closure"]
         print(
             f"It took {self.elapsed():.2f} seconds to receive",

src/speckleifc/__main__.py

@@ -18,7 +18,7 @@ def cmd_line_import() -> None:
     parser.add_argument("output_path")
     parser.add_argument("project_id")
     parser.add_argument("version_message")
-    parser.add_argument("model_id")
+    parser.add_argument("model_ingestion_id")
     # parser.add_argument("model_name")
     # parser.add_argument("region_name")
 
@@ -32,6 +32,8 @@ def cmd_line_import() -> None:
         "ifc",
     )
 
+    client: SpeckleClient | None = None
+
     try:
         client = SpeckleClient(SERVER_URL, use_ssl=not SERVER_URL.startswith("http://"))
         client.authenticate_with_token(TOKEN)
@@ -41,13 +43,14 @@ def cmd_line_import() -> None:
             args.file_path,
             project,
             args.version_message,
-            args.model_id,
+            args.model_ingestion_id,
             client,
         )
         with open(args.output_path, "w") as f:
             json.dump({"success": True, "commitId": version.id}, f)
     except Exception as e:
-        error_msg = f"IFC Importer failed with exception:\n{traceback.format_exc()}"
+        stack_trace = traceback.format_exc()
+        error_msg = f"IFC Importer failed with exception:\n{stack_trace}"
         print(error_msg)
 
         # Write error result

src/speckleifc/ifc_geometry_processing.py

@@ -51,4 +51,10 @@ def open_ifc(file_path: str) -> file:
 
 
 def create_geometry_iterator(ifc_file: file | sqlite) -> iterator:
-    return iterator(_create_iterator_settings(), ifc_file, multiprocessing.cpu_count())
+    GEOMETRY_LIBRARY = "hybrid-opencascade-cgal"  # First OCC then fallback to CGAL
+    return iterator(
+        _create_iterator_settings(),
+        ifc_file,
+        multiprocessing.cpu_count(),
+        geometry_library=GEOMETRY_LIBRARY,  # type: ignore
+    )

src/speckleifc/main.py

@@ -1,9 +1,19 @@
+import contextlib
+import importlib.metadata
 import time
+import traceback
+from pathlib import Path
 
 from speckleifc.ifc_geometry_processing import open_ifc
 from speckleifc.importer import ImportJob
 from specklepy.core.api.client import SpeckleClient
-from specklepy.core.api.inputs.version_inputs import CreateVersionInput
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionFailedInput,
+    ModelIngestionStartProcessingInput,
+    ModelIngestionSuccessInput,
+    ModelIngestionUpdateInput,
+    SourceDataInput,
+)
 from specklepy.core.api.models.current import Project, Version
 from specklepy.core.api.operations import send
 from specklepy.logging import metrics
@@ -13,49 +23,105 @@ from specklepy.transports.server import ServerTransport
 def open_and_convert_file(
     file_path: str,
     project: Project,
-    version_message: str | None,
-    model_id: str,
+    model_ingestion_id: str,
     client: SpeckleClient,
 ) -> Version:
-    start = time.time()
-    very_start = start
+    try:
+        start = time.time()
+        very_start = start
+        path = Path(file_path)
 
-    account = client.account
-    server_url = account.serverInfo.url
-    assert server_url
-    remote_transport = ServerTransport(project.id, account=account)
+        specklepy_version = importlib.metadata.version("specklepy")
+        client.model_ingestion.start_processing(
+            ModelIngestionStartProcessingInput(
+                project_id=project.id,
+                ingestion_id=model_ingestion_id,
+                progress_message="Importing IFC file",
+                source_data=SourceDataInput(
+                    file_name=path.name,
+                    file_size_bytes=path.stat().st_size,
+                    source_application_slug=metrics.HOST_APP,
+                    source_application_version=specklepy_version,
+                ),
+            )
+        )
 
-    ifc_file = open_ifc(file_path)  # pyright: ignore[reportUnknownVariableType]
+        account = client.account
+        server_url = account.serverInfo.url
+        assert server_url
+        remote_transport = ServerTransport(project.id, account=account)
 
-    import_job = ImportJob(ifc_file)  # pyright: ignore[reportUnknownArgumentType]
-    data = import_job.convert()
+        ifc_file = open_ifc(file_path)  # pyright: ignore[reportUnknownVariableType]
 
-    print(f"File conversion complete after {(time.time() - start) * 1000}ms")
+        client.model_ingestion.update_progress(
+            ModelIngestionUpdateInput(
+                project_id=project.id,
+                ingestion_id=model_ingestion_id,
+                progress_message="Converting file",
+                progress=None,
+            )
+        )
+        import_job = ImportJob(ifc_file)  # pyright: ignore[reportUnknownArgumentType]
+        data = import_job.convert()
 
-    start = time.time()
+        print(f"File conversion complete after {(time.time() - start) * 1000}ms")
 
-    root_id = send(data, transports=[remote_transport], use_default_cache=False)
-    print(f"Sending to speckle complete after: {(time.time() - start) * 1000}ms")
+        start = time.time()
 
-    start = time.time()
+        client.model_ingestion.update_progress(
+            ModelIngestionUpdateInput(
+                project_id=project.id,
+                ingestion_id=model_ingestion_id,
+                progress_message="Uploading objects",
+                progress=None,
+            )
+        )
+        root_id = send(data, transports=[remote_transport], use_default_cache=False)
+        print(f"Sending to speckle complete after: {(time.time() - start) * 1000}ms")
 
-    create_version = CreateVersionInput(
-        object_id=root_id,
-        model_id=model_id,
-        project_id=project.id,
-        message=version_message,
-        source_application="ifc",
-    )
-    version = client.version.create(create_version)
-    end = time.time()
-    print(f"Version committed after: {(end - start) * 1000}ms")
+        start = time.time()
 
-    print(f"Total time (to commit): {(end - very_start) * 1000}ms")
-    del ifc_file
+        version_id = client.model_ingestion.complete(
+            ModelIngestionSuccessInput(
+                project_id=project.id,
+                ingestion_id=model_ingestion_id,
+                root_object_id=root_id,
+                # version_message=version_message,
+            )
+        )
 
-    custom_properties = {"ui": "dui3", "actionSource": "import"}
-    if project.workspace_id:
-        custom_properties["workspace_id"] = project.workspace_id
-    metrics.track(metrics.SEND, account, custom_properties, send_sync=True)
+        # needed to query version until ingestion api expands to serve it
+        version = client.version.get(version_id, project.id)
 
-    return version
+        end = time.time()
+        print(f"Version committed after: {(end - start) * 1000}ms")
+
+        print(f"Total time (to commit): {(end - very_start) * 1000}ms")
+        del ifc_file
+
+        custom_properties = {"ui": "dui3", "actionSource": "import"}
+        if project.workspace_id:
+            custom_properties["workspace_id"] = project.workspace_id
+
+        metrics.track(
+            metrics.SEND,
+            account,
+            custom_properties,
+            send_sync=True,
+            track_email=True,
+        )
+
+        return version
+    except Exception as e:
+        stack_trace = traceback.format_exc()
+        with contextlib.suppress(Exception):
+            # make sure to not report process kills when we're cancelling
+            client.model_ingestion.fail_with_error(
+                ModelIngestionFailedInput(
+                    project_id=project.id,
+                    ingestion_id=model_ingestion_id,
+                    error_reason=str(e),
+                    error_stacktrace=stack_trace,
+                )
+            )
+        raise e

src/specklepy/api/client.py

@@ -4,6 +4,7 @@ from specklepy.api.credentials import Account
 from specklepy.api.resources import (
     ActiveUserResource,
     FileImportResource,
+    ModelIngestionResource,
     ModelResource,
     OtherUserResource,
     ProjectInviteResource,
@@ -119,6 +120,12 @@ class SpeckleClient(CoreSpeckleClient):
             client=self.httpclient,
             server_version=server_version,
         )
+        self.model_ingestion = ModelIngestionResource(
+            account=self.account,
+            basepath=self.url,
+            client=self.httpclient,
+            server_version=server_version,
+        )
         self.file_import = FileImportResource(
             account=self.account,
             basepath=self.url,

src/specklepy/api/resources/__init__.py

@@ -1,5 +1,8 @@
 from specklepy.api.resources.current.active_user_resource import ActiveUserResource
 from specklepy.api.resources.current.file_import_resource import FileImportResource
+from specklepy.api.resources.current.model_ingestion_resource import (
+    ModelIngestionResource,
+)
 from specklepy.api.resources.current.model_resource import ModelResource
 from specklepy.api.resources.current.other_user_resource import OtherUserResource
 from specklepy.api.resources.current.project_invite_resource import (
@@ -22,4 +25,6 @@ __all__ = [
     "SubscriptionResource",
     "VersionResource",
     "WorkspaceResource",
+    "FileImportResource",
+    "ModelIngestionResource",
 ]

src/specklepy/api/resources/current/file_import_resource.py

@@ -15,7 +15,7 @@ from specklepy.logging import metrics
 
 
 class FileImportResource(CoreResource):
-    """API Access class for projects"""
+    """API Access class for file imports"""
 
     def __init__(self, account, basepath, client, server_version) -> None:
         super().__init__(

src/specklepy/api/resources/current/model_ingestion_resource.py

@@ -0,0 +1,57 @@
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionCancelledInput,
+    ModelIngestionCreateInput,
+    ModelIngestionFailedInput,
+    ModelIngestionRequeueInput,
+    ModelIngestionStartProcessingInput,
+    ModelIngestionSuccessInput,
+    ModelIngestionUpdateInput,
+)
+from specklepy.core.api.models.current import (
+    ModelIngestion,
+)
+from specklepy.core.api.resources import (
+    ModelIngestionResource as CoreResource,
+)
+from specklepy.logging import metrics
+
+
+class ModelIngestionResource(CoreResource):
+    """API Access class for model ingestion"""
+
+    def __init__(self, account, basepath, client, server_version) -> None:
+        super().__init__(account, basepath, client, server_version)
+
+    def create(self, input: ModelIngestionCreateInput) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Create"})
+        return super().create(input)
+
+    def get_ingestion(self, project_id: str, model_ingestion_id: str) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Get"})
+        return super().get_ingestion(project_id, model_ingestion_id)
+
+    def update_progress(self, input: ModelIngestionUpdateInput) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Update"})
+        return super().update_progress(input)
+
+    def start_processing(
+        self, input: ModelIngestionStartProcessingInput
+    ) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Start Processing"})
+        return super().start_processing(input)
+
+    def requeue(self, input: ModelIngestionRequeueInput) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Update"})
+        return super().requeue(input)
+
+    def complete(self, input: ModelIngestionSuccessInput) -> str:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion End"})
+        return super().complete(input)
+
+    def fail_with_error(self, input: ModelIngestionFailedInput) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Error"})
+        return super().fail_with_error(input)
+
+    def fail_with_cancel(self, input: ModelIngestionCancelledInput) -> ModelIngestion:
+        metrics.track(metrics.SDK, self.account, {"name": "Ingestion Cancel"})
+        return super().fail_with_cancel(input)

src/specklepy/core/api/client.py

@@ -12,6 +12,7 @@ from specklepy.core.api.credentials import Account
 from specklepy.core.api.resources import (
     ActiveUserResource,
     FileImportResource,
+    ModelIngestionResource,
     ModelResource,
     OtherUserResource,
     ProjectInviteResource,
@@ -250,6 +251,12 @@ class SpeckleClient:
             client=self.httpclient,
             server_version=server_version,
         )
+        self.model_ingestion = ModelIngestionResource(
+            account=self.account,
+            basepath=self.url,
+            client=self.httpclient,
+            server_version=server_version,
+        )
         self.subscription = SubscriptionResource(
             account=self.account,
             basepath=self.ws_url,

src/specklepy/core/api/enums.py

@@ -30,3 +30,18 @@ class ProjectVersionsUpdatedMessageType(str, Enum):
     CREATED = "CREATED"
     DELETED = "DELETED"
     UPDATED = "UPDATED"
+
+
+class ProjectModelIngestionUpdatedMessageType(str, Enum):
+    CANCELLATION_REQUESTED = "cancellationRequested"
+    CREATED = "created"
+    DELETED = "deleted"
+    UPDATED = "updated"
+
+
+class ModelIngestionStatus(str, Enum):
+    CANCELLED = "cancelled"
+    FAILED = "failed"
+    PROCESSING = "processing"
+    QUEUED = "queued"
+    SUCCESS = "success"

src/specklepy/core/api/inputs/model_ingestion_inputs.py

@@ -0,0 +1,76 @@
+from specklepy.core.api.enums import ProjectModelIngestionUpdatedMessageType
+from specklepy.core.api.models.graphql_base_model import GraphQLBaseModel
+
+
+class SourceDataInput(GraphQLBaseModel):
+    source_application_slug: str
+    source_application_version: str
+    file_name: str | None
+    file_size_bytes: int | None
+
+
+class ModelIngestionCreateInput(GraphQLBaseModel):
+    model_id: str
+    project_id: str
+    progress_message: str
+    source_data: SourceDataInput
+
+
+class ModelIngestionStartProcessingInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    progress_message: str
+    source_data: SourceDataInput
+
+
+class ModelIngestionRequeueInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    progress_message: str
+
+
+class ModelIngestionUpdateInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    progress: float | None
+    progress_message: str
+
+
+class ModelIngestionSuccessInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    root_object_id: str
+
+
+class ModelIngestionFailedInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    error_reason: str
+    error_stacktrace: str | None
+
+
+class ModelIngestionCancelledInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    cancellation_message: str
+
+
+class ModelIngestionRequestCancellationInput(GraphQLBaseModel):
+    ingestion_id: str
+    project_id: str
+    cancellation_message: str
+
+
+class ModelIngestionReference(GraphQLBaseModel):
+    """
+    `@oneOf` i.e. server expects **either** `ingestion_id` or `model_id`, but not both.
+    """
+
+    ingestion_id: str | None
+    model_id: str | None
+
+
+class ProjectModelIngestionSubscriptionInput(GraphQLBaseModel):
+    project_id: str
+    ingestion_reference: ModelIngestionReference
+    message_type: ProjectModelIngestionUpdatedMessageType | None = None

src/specklepy/core/api/models/current.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Generic, List, TypeVar
 
-from specklepy.core.api.enums import ProjectVisibility
+from specklepy.core.api.enums import ModelIngestionStatus, ProjectVisibility
 from specklepy.core.api.models.graphql_base_model import GraphQLBaseModel
 from specklepy.logging.exceptions import WorkspacePermissionException
 
@@ -244,3 +244,17 @@ class FileImport(GraphQLBaseModel):
 class FileUploadUrl(GraphQLBaseModel):
     url: str
     file_id: str
+
+
+class ModelIngestionStatusData(GraphQLBaseModel):
+    status: ModelIngestionStatus
+    progress_message: str | None = None
+
+
+class ModelIngestion(GraphQLBaseModel):
+    id: str
+    created_at: datetime
+    updated_at: datetime
+    cancellation_requested: bool
+    model_id: str
+    status_data: ModelIngestionStatusData

src/specklepy/core/api/models/subscription_messages.py

@@ -1,12 +1,13 @@
 from typing import Optional
 
 from specklepy.core.api.enums import (
+    ProjectModelIngestionUpdatedMessageType,
     ProjectModelsUpdatedMessageType,
     ProjectUpdatedMessageType,
     ProjectVersionsUpdatedMessageType,
     UserProjectsUpdatedMessageType,
 )
-from specklepy.core.api.models.current import Model, Project, Version
+from specklepy.core.api.models.current import Model, ModelIngestion, Project, Version
 from specklepy.core.api.models.graphql_base_model import GraphQLBaseModel
 
 
@@ -33,3 +34,8 @@ class ProjectVersionsUpdatedMessage(GraphQLBaseModel):
     type: ProjectVersionsUpdatedMessageType
     model_id: str
     version: Optional[Version]
+
+
+class ProjectModelIngestionUpdatedMessage(GraphQLBaseModel):
+    model_ingestion: ModelIngestion
+    type: ProjectModelIngestionUpdatedMessageType

src/specklepy/core/api/resources/__init__.py

@@ -1,5 +1,8 @@
 from specklepy.core.api.resources.current.active_user_resource import ActiveUserResource
 from specklepy.core.api.resources.current.file_import_resource import FileImportResource
+from specklepy.core.api.resources.current.model_ingestion_resource import (
+    ModelIngestionResource,
+)
 from specklepy.core.api.resources.current.model_resource import ModelResource
 from specklepy.core.api.resources.current.other_user_resource import OtherUserResource
 from specklepy.core.api.resources.current.project_invite_resource import (
@@ -24,4 +27,6 @@ __all__ = [
     "SubscriptionResource",
     "VersionResource",
     "WorkspaceResource",
+    "FileImportResource",
+    "ModelIngestionResource",
 ]

src/specklepy/core/api/resources/current/file_import_resource.py

@@ -16,13 +16,15 @@ from specklepy.core.api.resource import ResourceBase
 from specklepy.core.api.responses import DataResponse
 from specklepy.logging.exceptions import SpeckleException
 
+NAME = "file_import"
+
 
 class UploadFileResponse(GraphQLBaseModel):
     etag: str
 
 
 class FileImportResource(ResourceBase):
-    """API Access class for project invites"""
+    """API Access class for file imports"""
 
     def __init__(
         self,
@@ -36,7 +38,7 @@ class FileImportResource(ResourceBase):
             basepath=basepath,
             client=client,
             server_version=server_version,
-            name="file-import",
+            name=NAME,
         )
 
     def finish_file_import_job(self, input: FinishFileImportInput) -> bool:

src/specklepy/core/api/resources/current/model_ingestion_resource.py

@@ -0,0 +1,397 @@
+from typing import Any, Tuple
+
+from gql import Client, gql
+
+from specklepy.api.credentials import Account
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionCancelledInput,
+    ModelIngestionCreateInput,
+    ModelIngestionFailedInput,
+    ModelIngestionRequestCancellationInput,
+    ModelIngestionRequeueInput,
+    ModelIngestionStartProcessingInput,
+    ModelIngestionSuccessInput,
+    ModelIngestionUpdateInput,
+)
+from specklepy.core.api.models.current import (
+    ModelIngestion,
+)
+from specklepy.core.api.resource import ResourceBase
+from specklepy.core.api.responses import DataResponse
+
+NAME = "ingestion"
+
+
+class ModelIngestionResource(ResourceBase):
+    """API Access class for model ingestion"""
+
+    def __init__(
+        self,
+        account: Account,
+        basepath: str,
+        client: Client,
+        server_version: Tuple[Any, ...] | None,
+    ) -> None:
+        super().__init__(
+            account=account,
+            basepath=basepath,
+            client=client,
+            name=NAME,
+            server_version=server_version,
+        )
+
+    def get_ingestion(self, project_id: str, model_ingestion_id: str) -> ModelIngestion:
+        QUERY = gql(
+            """
+            query Query($projectId: String!, $modelIngestionId: ID!) {
+              data:project(id: $projectId) {
+                data:ingestion(id: $modelIngestionId) {
+                  id
+                  createdAt
+                  updatedAt
+                  modelId
+                  cancellationRequested
+                  statusData {
+                    ... on HasModelIngestionStatus {
+                      status
+                    }
+                    ... on HasProgressMessage {
+                      progressMessage
+                    }
+                  }
+                }
+              }
+            }
+            """  # noqa: E501
+        )
+
+        variables = {
+            "projectId": project_id,
+            "modelIngestionId": model_ingestion_id,
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[ModelIngestion]],
+            QUERY,
+            variables,
+        ).data.data
+
+    def create(self, input: ModelIngestionCreateInput) -> ModelIngestion:
+        QUERY = gql(
+            """
+            mutation IngestionCreate($input: ModelIngestionCreateInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: create(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+           """
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]], QUERY, variables
+        ).data.data.data
+
+    def start_processing(
+        self, input: ModelIngestionStartProcessingInput
+    ) -> ModelIngestion:
+        QUERY = gql(
+            """
+            mutation IngestionStartProcessing($input: ModelIngestionStartProcessingInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: startProcessing(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+           """  # noqa: E501
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]], QUERY, variables
+        ).data.data.data
+
+    def requeue(self, input: ModelIngestionRequeueInput) -> ModelIngestion:
+        QUERY = gql(
+            """
+            mutation IngestionRequeue($input: ModelIngestionRequeueInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: requeue(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+           """  # noqa: E501
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]], QUERY, variables
+        ).data.data.data
+
+    def update_progress(self, input: ModelIngestionUpdateInput) -> ModelIngestion:
+        QUERY = gql(
+            """
+            mutation IngestionUpdateProgress(
+              $input: ModelIngestionUpdateInput!
+            ) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: updateProgress(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+           """
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]], QUERY, variables
+        ).data.data.data
+
+    def complete(self, input: ModelIngestionSuccessInput) -> str:
+        """
+        Request that the server completes the ingestion by creating a version
+        If successful, the job will be in a terminal "successful" state.
+
+        For failed Ingestions, use `fail_with_error` instead
+        For user cancellation, use `fail_with_cancelled` instead
+
+        Arguments:
+            input {ModelIngestionSuccessInput} -- input variable
+
+        Returns:
+            str -- the id of the version that was just created to complete the ingestion
+        """
+        QUERY = gql(
+            """
+            mutation IngestionComplete($input: ModelIngestionSuccessInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: completeWithVersion(input: $input) {
+                    data:statusData {
+                      ... on ModelIngestionSuccessStatus {
+                        data:versionId
+                      }
+                    }
+                  }
+                }
+              }
+            }
+           """
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[DataResponse[DataResponse[str]]]]],
+            QUERY,
+            variables,
+        ).data.data.data.data.data
+
+    def fail_with_error(self, input: ModelIngestionFailedInput) -> ModelIngestion:
+        """
+        Fail the job with an error.
+        For user requested cancellation, use `fail_with_cancelled` instead
+        """
+        QUERY = gql(
+            """
+            mutation IngestionFailWithError($input: ModelIngestionFailedInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: failWithError(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+            """
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]],
+            QUERY,
+            variables,
+        ).data.data.data
+
+    def fail_with_cancel(self, input: ModelIngestionCancelledInput) -> ModelIngestion:
+        """
+        Fail the ingestion with a `cancelled` status.
+        This should only be done if the user has explicitly requested cancellation
+        Other forms of cancellation use `fail_with_error`
+        The ingestion should then enter a terminal "canceled" state
+        """
+        QUERY = gql(
+            """
+            mutation IngestionFailWithCancel($input: ModelIngestionCancelledInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: failWithCancel(input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+            """
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]],
+            QUERY,
+            variables,
+        ).data.data.data
+
+    def request_cancellation(
+        self, input: ModelIngestionRequestCancellationInput
+    ) -> ModelIngestion:
+        """
+        Request that the ingestion is canceled.
+
+        Note: simply calling this mutation does not immediately cancel,
+        it doesn't even guarantee it will be canceled at all.
+        It's up to the client to observe this cancellation request
+        via `subscription.project_model_ingestion_cancellation_requested`
+        and report it as cancelled (via `ingestion.fail_with_cancel`
+
+        See "cooperative cancellation pattern"
+        """
+        QUERY = gql(
+            """
+            mutation IngestionRequestCancellation($input: ModelIngestionRequestCancellationInput!) {
+              data: projectMutations {
+                data: modelIngestionMutations {
+                  data: requestCancellation (input: $input) {
+                    id
+                    createdAt
+                    updatedAt
+                    modelId
+                    cancellationRequested
+                    statusData {
+                      ... on HasModelIngestionStatus {
+                        status
+                      }
+                      ... on HasProgressMessage {
+                        progressMessage
+                      }
+                    }
+                  }
+                }
+              }
+            }
+            """  # noqa: E501
+        )
+
+        variables = {
+            "input": input.model_dump(warnings="error", by_alias=True),
+        }
+
+        return self.make_request_and_parse_response(
+            DataResponse[DataResponse[DataResponse[ModelIngestion]]],
+            QUERY,
+            variables,
+        ).data.data.data

src/specklepy/core/api/resources/current/subscription_resource.py

@@ -6,17 +6,25 @@ from graphql import DocumentNode
 from pydantic import BaseModel
 from typing_extensions import TypeVar
 
+from specklepy.core.api.enums import ProjectModelIngestionUpdatedMessageType
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionReference,
+    ProjectModelIngestionSubscriptionInput,
+)
 from specklepy.core.api.models import (
     ProjectModelsUpdatedMessage,
     ProjectUpdatedMessage,
     ProjectVersionsUpdatedMessage,
     UserProjectsUpdatedMessage,
 )
+from specklepy.core.api.models.subscription_messages import (
+    ProjectModelIngestionUpdatedMessage,
+)
 from specklepy.core.api.resource import ResourceBase
 from specklepy.core.api.responses import DataResponse
 from specklepy.logging.exceptions import SpeckleException
 
-NAME = "subscribe"
+NAME = "subscription"
 
 TEventArgs = TypeVar("TEventArgs", bound=BaseModel)
 
@@ -202,6 +210,66 @@ class SubscriptionResource(ResourceBase):
             callback=lambda d: callback(d.data),
         )
 
+    async def project_model_ingestion_updated(
+        self,
+        callback: Callable[[ProjectModelIngestionUpdatedMessage], None],
+        input: ProjectModelIngestionSubscriptionInput,
+    ) -> None:
+        QUERY = gql(
+            """
+            subscription IngestionUpdated($input: ProjectModelIngestionSubscriptionInput!) {
+              data: projectModelIngestionUpdated(input: $input) {
+                modelIngestion {
+                  id
+                  createdAt
+                  updatedAt
+                  modelId
+                  cancellationRequested
+                  statusData {
+                    ... on HasModelIngestionStatus {
+                      status
+                    }
+                    ... on HasProgressMessage {
+                      progressMessage
+                    }
+                  }
+                }
+                type
+              }
+            }
+            """  # noqa: E501
+        )
+
+        variables = {
+            "input": input.model_dump(
+                warnings="error", by_alias=True, exclude_none=True
+            ),
+        }
+
+        await self.subscribe_2(
+            DataResponse[ProjectModelIngestionUpdatedMessage],
+            QUERY,
+            variables,
+            callback=lambda d: callback(d.data),
+        )
+
+    async def project_model_ingestion_cancellation_requested(
+        self,
+        callback: Callable[[ProjectModelIngestionUpdatedMessage], None],
+        project_id: str,
+        ingestion_id: str,
+    ) -> None:
+        await self.project_model_ingestion_updated(
+            callback,
+            ProjectModelIngestionSubscriptionInput(
+                project_id=project_id,
+                ingestion_reference=ModelIngestionReference(
+                    ingestion_id=ingestion_id, model_id=None
+                ),
+                message_type=ProjectModelIngestionUpdatedMessageType.CANCELLATION_REQUESTED,
+            ),
+        )
+
     @check_wsclient
     async def subscribe_2(
         self,

src/specklepy/core/api/resources/current/version_resource.py

@@ -14,7 +14,7 @@ from specklepy.core.api.models import ResourceCollection, Version
 from specklepy.core.api.resource import ResourceBase
 from specklepy.core.api.responses import DataResponse
 
-NAME = "model"
+NAME = "version"
 
 
 class VersionResource(ResourceBase):

src/specklepy/core/api/resources/current/workspace_resource.py

@@ -16,7 +16,7 @@ NAME = "workspace"
 
 
 class WorkspaceResource(ResourceBase):
-    """API Access class for models"""
+    """API Access class for workspaces"""
 
     def __init__(self, account, basepath, client, server_version) -> None:
         super().__init__(

src/specklepy/logging/metrics.py

@@ -1,19 +1,20 @@
 import contextlib
 import getpass
 import hashlib
+import importlib.metadata
 import logging
 import platform
 import queue
 import sys
 import threading
-from typing import Any
+from typing import Any, Literal
 
 import requests
 
 from specklepy.core.api.credentials import Account
 
 """
-Anonymous telemetry to help us understand how to make a better Speckle.
+Lightweight usage telemetry to help us understand how to make a better Speckle.
 This really helps us to deliver a better open source project and product!
 """
 TRACK = True
@@ -22,13 +23,14 @@ HOST_APP_VERSION = f"python {'.'.join(map(str, sys.version_info[:2]))}"
 PLATFORMS = {"win32": "Windows", "cygwin": "Windows", "darwin": "Mac OS X"}
 
 LOG = logging.getLogger(__name__)
-METRICS_TRACKER = None
+METRICS_TRACKER: "MetricsTracker | None" = None
 
 # actions
 SDK = "SDK Action"
 CONNECTOR = "Connector Action"
 RECEIVE = "Receive"
 SEND = "Send"
+ACTIONS = Literal["SDK Action", "Connector Action", "Receive", "Send"]
 
 
 def disable():
@@ -48,15 +50,32 @@ def set_host_app(host_app: str, host_app_version: str | None = None):
 
 
 def track(
-    action: str,
+    action: ACTIONS,
     account: Account | None = None,
     custom_props: dict | None = None,
     send_sync: bool = False,
+    track_email: bool = False,
 ):
+    """
+    :param action:
+    :type action: ACTIONS
+    :param account:
+    :type account: Account | None
+    :param custom_props:
+    :type custom_props: dict | None
+    :param send_sync: When `True`, the track event is executed synchronously,
+           and any exceptions will be raised.
+           When `False`, the track it is deferred to a queue, and any exceptions will be
+           swallowed and reported as warnings.
+    :type send_sync: bool
+    :param track_email: When `True`, the users plain text email address will be included
+    :type track_email: bool
+    """
     if not TRACK:
         return
 
     tracker = initialise_tracker(account)
+
     event_params: dict[str, Any] = {
         "event": action,
         "properties": {
@@ -72,6 +91,18 @@ def track(
     if custom_props:
         event_params["properties"].update(custom_props)
 
+    if track_email:
+        event_params["properties"]["email"] = tracker.last_email
+
+    try:
+        specklepy_version = importlib.metadata.version("specklepy")
+        event_params["properties"]["core_version"] = specklepy_version
+    except importlib.metadata.PackageNotFoundError:
+        if send_sync:
+            raise
+        else:
+            LOG.warning("Failed to read specklepy's version number", exc_info=True)
+
     if send_sync:
         tracker.send_event(event_params)
     else:
@@ -84,7 +115,7 @@ def initialise_tracker(account: Account | None = None) -> "MetricsTracker":
         METRICS_TRACKER = MetricsTracker()
 
     if account:
-        METRICS_TRACKER.set_last_user(account.userInfo.email)
+        METRICS_TRACKER.set_last_user_email(account.userInfo.email)
         METRICS_TRACKER.set_last_server(account.serverInfo.url)
 
     return METRICS_TRACKER
@@ -103,6 +134,7 @@ class MetricsTracker(metaclass=Singleton):
     analytics_url: str = "https://analytics.speckle.systems/track?ip=1"
     analytics_token: str = "acd87c5a50b56df91a795e999812a3a4"
     last_user: str = ""
+    last_email: str = ""
     last_server: str | None = None
     platform: str
 
@@ -121,17 +153,19 @@ class MetricsTracker(metaclass=Singleton):
             if node and user:
                 self.last_user = f"@{self.hash(f'{node}-{user}')}"
 
-    def set_last_user(self, email: str | None) -> None:
+    def set_last_user_email(self, email: str | None) -> None:
         if not email:
             return
         self.last_user = f"@{self.hash(email)}"
+        self.last_email = email
 
     def set_last_server(self, server: str | None) -> None:
         if not server:
             return
         self.last_server = self.hash(server)
 
-    def hash(self, value: str) -> str:
+    @staticmethod
+    def hash(value: str) -> str:
         inputList = value.lower().split("://")
         input = inputList[len(inputList) - 1].split("/")[0].split("?")[0]
         return hashlib.md5(input.encode("utf-8")).hexdigest().upper()

tests/integration/client/current/test_model_ingestion_resource.py

@@ -0,0 +1,277 @@
+from datetime import datetime
+
+import pytest
+
+from specklepy.api import operations
+from specklepy.api.client import SpeckleClient
+from specklepy.core.api.enums import ModelIngestionStatus
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionCancelledInput,
+    ModelIngestionCreateInput,
+    ModelIngestionFailedInput,
+    ModelIngestionRequeueInput,
+    ModelIngestionStartProcessingInput,
+    ModelIngestionSuccessInput,
+    ModelIngestionUpdateInput,
+    SourceDataInput,
+)
+from specklepy.core.api.inputs.model_inputs import CreateModelInput
+from specklepy.core.api.inputs.project_inputs import ProjectCreateInput
+from specklepy.core.api.models.current import (
+    Model,
+    ModelIngestion,
+    ModelIngestionStatusData,
+    Project,
+    ProjectVisibility,
+    Version,
+)
+from specklepy.logging.exceptions import GraphQLException
+from specklepy.objects.base import Base
+from specklepy.transports.server.server import ServerTransport
+from tests.integration.conftest import is_public
+
+
+@pytest.mark.run()
+@pytest.mark.skipif(is_public(), reason="The public API does not support these tests")
+class TestIngestionResource:
+    @pytest.fixture
+    def project(self, client: SpeckleClient):
+        return client.project.create(
+            ProjectCreateInput(
+                name="test", description=None, visibility=ProjectVisibility.PUBLIC
+            )
+        )
+
+    @pytest.fixture
+    def model(self, client: SpeckleClient, project: Project):
+        return client.model.create(
+            CreateModelInput(name="test", description=None, project_id=project.id)
+        )
+
+    @pytest.fixture()
+    def ingestion(
+        self, client: SpeckleClient, model: Model, project: Project
+    ) -> ModelIngestion:
+        input = ModelIngestionCreateInput(
+            model_id=model.id,
+            project_id=project.id,
+            progress_message="Starting processing",
+            source_data=SourceDataInput(
+                source_application_slug="pytest",
+                source_application_version="0.0.0",
+                file_name=None,
+                file_size_bytes=None,
+            ),
+        )
+
+        ingestion = client.model_ingestion.create(input)
+        assert isinstance(ingestion, ModelIngestion)
+        assert isinstance(ingestion.id, str)
+        assert isinstance(ingestion.created_at, datetime)
+        assert isinstance(ingestion.updated_at, datetime)
+        assert isinstance(ingestion.cancellation_requested, bool)
+        assert isinstance(ingestion.model_id, str)
+        assert isinstance(ingestion.status_data, ModelIngestionStatusData)
+        assert isinstance(ingestion.status_data.progress_message, str | None)
+        assert ingestion.status_data.status == ModelIngestionStatus.PROCESSING
+        assert not ingestion.cancellation_requested
+        assert ingestion.model_id == model.id
+
+        return ingestion
+
+    def test_get_ingestion(
+        self, client: SpeckleClient, project: Project, ingestion: ModelIngestion
+    ):
+        queried_ingestion = client.model_ingestion.get_ingestion(
+            project.id, ingestion.id
+        )
+        assert queried_ingestion.id == ingestion.id
+        assert queried_ingestion.status_data.status == ingestion.status_data.status
+
+    def test_update_progress(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        def update(progress: float | None, message: str):
+            input = ModelIngestionUpdateInput(
+                ingestion_id=ingestion.id,
+                project_id=project.id,
+                progress=progress,
+                progress_message=message,
+            )
+            res = client.model_ingestion.update_progress(input)
+
+            assert isinstance(res, ModelIngestion)
+            assert res.status_data.progress_message == message
+            assert not res.cancellation_requested
+            assert res.status_data.status == ModelIngestionStatus.PROCESSING
+
+        update(None, "None")
+        update(0.1, "0.1")
+        update(0.5, "Whoa-oh! We're half way there!")
+        update(1, "Finished")
+        update(0.2, "Back to processing again")
+
+    def test_start_processing_and_requeue(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        # just setting the baseline state
+        assert ingestion.status_data.status == ModelIngestionStatus.PROCESSING
+
+        def requeue(message: str):
+            input = ModelIngestionRequeueInput(
+                project_id=project.id,
+                ingestion_id=ingestion.id,
+                progress_message=message,
+            )
+
+            res = client.model_ingestion.requeue(input)
+
+            assert isinstance(res, ModelIngestion)
+            assert res.status_data.progress_message == message
+            assert not res.cancellation_requested
+            assert res.status_data.status == ModelIngestionStatus.QUEUED
+
+        def start_processing(message: str):
+            input = ModelIngestionStartProcessingInput(
+                ingestion_id=ingestion.id,
+                project_id=project.id,
+                progress_message=message,
+                source_data=SourceDataInput(
+                    source_application_slug="test",
+                    source_application_version="test",
+                    file_name="test",
+                    file_size_bytes=1,
+                ),
+            )
+            res = client.model_ingestion.start_processing(input)
+
+            assert isinstance(res, ModelIngestion)
+            assert res.status_data.progress_message == message
+            assert not res.cancellation_requested
+            assert res.status_data.status == ModelIngestionStatus.PROCESSING
+
+        requeue("put it back in there")
+        start_processing("go for it")
+        requeue("and again")
+        start_processing("run forest run")
+
+        def update(progress: float | None, message: str):
+            input = ModelIngestionUpdateInput(
+                ingestion_id=ingestion.id,
+                project_id=project.id,
+                progress=progress,
+                progress_message=message,
+            )
+            res = client.model_ingestion.update_progress(input)
+
+            assert isinstance(res, ModelIngestion)
+            assert res.status_data.progress_message == message
+            assert not res.cancellation_requested
+            assert res.status_data.status == ModelIngestionStatus.PROCESSING
+
+        update(None, "None")
+        update(0.1, "0.1")
+        update(0.5, "Whoa-oh! We're half way there!")
+        update(1, "Finished")
+        update(0.2, "Back to processing again")
+
+    def test_error(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        input = ModelIngestionFailedInput(
+            ingestion_id=ingestion.id,
+            project_id=project.id,
+            error_reason="Failed to integration test an error",
+            error_stacktrace="over here in test_error",
+        )
+
+        res = client.model_ingestion.fail_with_error(input)
+
+        assert isinstance(res, ModelIngestion)
+        assert res.status_data.progress_message is None
+        assert not res.cancellation_requested
+        assert res.status_data.status == ModelIngestionStatus.FAILED
+
+        # trying to fail for a second time should throw
+        # with pytest.raises(GraphQLException):
+        #     _ = client.ingestion.fail_with_error(input)
+
+    def test_complete(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        remote = ServerTransport(project.id, client)
+        object_id = operations.send(
+            Base(applicationId="ASDFGHJKL"), [remote], use_default_cache=False
+        )
+        input = ModelIngestionSuccessInput(
+            ingestion_id=ingestion.id,
+            root_object_id=object_id,
+            project_id=project.id,
+            # version_message=None,
+        )
+
+        res = client.model_ingestion.complete(input)
+
+        assert isinstance(res, str)
+        version = client.version.get(res, project.id)
+        assert isinstance(version, Version)
+
+        # trying to complete for a second time should throw
+        # with pytest.raises(GraphQLException):
+        #     _ = client.ingestion.complete(input)
+
+    def test_cancel(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        input = ModelIngestionCancelledInput(
+            ingestion_id=ingestion.id,
+            project_id=project.id,
+            cancellation_message="This was cancelled for testing purposes",
+        )
+        res = client.model_ingestion.fail_with_cancel(input)
+        assert isinstance(res, ModelIngestion)
+        assert res.status_data.progress_message is None
+        assert not res.cancellation_requested
+        assert res.status_data.status == ModelIngestionStatus.CANCELLED
+
+        # Cancel again, should be idempotent
+        res = client.model_ingestion.fail_with_cancel(input)
+        assert res.status_data.progress_message is None
+        assert not res.cancellation_requested
+        assert res.status_data.status == ModelIngestionStatus.CANCELLED
+
+    def test_error_non_existent_ingestion(
+        self, client: SpeckleClient, project: Project
+    ):
+        input = ModelIngestionFailedInput(
+            ingestion_id="Non-existent-ingestion",
+            project_id=project.id,
+            error_reason="Failed to integration test an error",
+            error_stacktrace="over here in test_error",
+        )
+        with pytest.raises(GraphQLException):
+            _ = client.model_ingestion.fail_with_error(input)
+
+    def test_complete_failed_non_existent_ingestion(
+        self, client: SpeckleClient, project: Project
+    ):
+        input = ModelIngestionFailedInput(
+            ingestion_id="Non-existent-ingestion",
+            project_id=project.id,
+            error_reason="Failed to integration test an error",
+            error_stacktrace="over here in test_error",
+        )
+        with pytest.raises(GraphQLException):
+            _ = client.model_ingestion.fail_with_error(input)
+
+    def test_complete_non_existent_root_object(
+        self, client: SpeckleClient, ingestion: ModelIngestion, project: Project
+    ):
+        input = ModelIngestionSuccessInput(
+            ingestion_id=ingestion.id,
+            root_object_id="asdfasdfasdfasfd",
+            project_id=project.id,
+            # version_message=None,
+        )
+        with pytest.raises(GraphQLException):
+            _ = client.model_ingestion.complete(input)

tests/integration/client/current/test_subscription_resource.py

@@ -1,15 +1,26 @@
 import asyncio
-from typing import Dict, Optional
+from sys import platform
+from typing import Dict
 
 import pytest
 
 from specklepy.api.client import SpeckleClient
 from specklepy.core.api.enums import (
+    ModelIngestionStatus,
+    ProjectModelIngestionUpdatedMessageType,
     ProjectModelsUpdatedMessageType,
     ProjectUpdatedMessageType,
     ProjectVersionsUpdatedMessageType,
     UserProjectsUpdatedMessageType,
 )
+from specklepy.core.api.inputs.model_ingestion_inputs import (
+    ModelIngestionCreateInput,
+    ModelIngestionReference,
+    ModelIngestionRequestCancellationInput,
+    ModelIngestionUpdateInput,
+    ProjectModelIngestionSubscriptionInput,
+    SourceDataInput,
+)
 from specklepy.core.api.inputs.model_inputs import CreateModelInput
 from specklepy.core.api.inputs.project_inputs import (
     ProjectCreateInput,
@@ -24,9 +35,16 @@ from specklepy.core.api.models import (
     UserProjectsUpdatedMessage,
     Version,
 )
-from tests.integration.conftest import create_client, create_version
+from specklepy.core.api.models.current import ModelIngestion
+from specklepy.core.api.models.subscription_messages import (
+    ProjectModelIngestionUpdatedMessage,
+)
+from tests.integration.conftest import create_client, create_version, is_public
 
-WAIT_PERIOD = 0.4  # time in seconds
+# WSL is slow AF, so for local runs, we're being extra generous
+# For CI runs on linux,m a much smaller wait time is acceptable
+SETUP_TIME_SECONDS = 1 if platform == "linux" else 4
+MAX_WAIT_TIME_SECONDS = 0.75 if platform == "linux" else 5
 
 
 @pytest.mark.run()
@@ -55,48 +73,68 @@ class TestSubscriptionResource:
         )
         return model1
 
+    @pytest.fixture
+    def test_model_ingestion(
+        self,
+        subscription_client: SpeckleClient,
+        test_project: Project,
+        test_model: Model,
+    ) -> ModelIngestion:
+        project = subscription_client.model_ingestion.create(
+            ModelIngestionCreateInput(
+                project_id=test_project.id,
+                model_id=test_model.id,
+                progress_message="",
+                source_data=SourceDataInput(
+                    source_application_slug="pytest",
+                    source_application_version="0.0.0",
+                    file_name=None,
+                    file_size_bytes=None,
+                ),
+            )
+        )
+        return project
+
     @pytest.mark.asyncio
     async def test_user_projects_updated(
         self,
         subscription_client: SpeckleClient,
     ) -> None:
-        message: Optional[UserProjectsUpdatedMessage] = None
-
-        task = None
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[UserProjectsUpdatedMessage] = loop.create_future()
 
         def callback(d: UserProjectsUpdatedMessage):
-            nonlocal message
-            message = d
+            nonlocal future
+            future.set_result(d)
 
         task = asyncio.create_task(
             subscription_client.subscription.user_projects_updated(callback)
         )
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time to subscription to be setup
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
 
         input = ProjectCreateInput(name=None, description=None, visibility=None)
         created = subscription_client.project.create(input)
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time for subscription to be triggered
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
 
         assert isinstance(message, UserProjectsUpdatedMessage)
         assert message.id == created.id
         assert message.type == UserProjectsUpdatedMessageType.ADDED
         assert isinstance(message.project, Project)
-        task.cancel()
-        await task
+        if not task.cancel():
+            await task
 
     @pytest.mark.asyncio
     async def test_project_models_updated(
         self, subscription_client: SpeckleClient, test_project: Project
     ) -> None:
-        message: Optional[ProjectModelsUpdatedMessage] = None
-
-        task = None
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectModelsUpdatedMessage] = loop.create_future()
 
         def callback(d: ProjectModelsUpdatedMessage):
-            nonlocal message
-            message = d
+            nonlocal future
+            future.set_result(d)
 
         task = asyncio.create_task(
             subscription_client.subscription.project_models_updated(
@@ -104,51 +142,53 @@ class TestSubscriptionResource:
             )
         )
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time to subscription to be setup
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
 
         input = CreateModelInput(
             name="my model", description="myDescription", project_id=test_project.id
         )
         created = subscription_client.model.create(input)
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time for subscription to be triggered
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
 
         assert isinstance(message, ProjectModelsUpdatedMessage)
         assert message.id == created.id
         assert message.type == ProjectModelsUpdatedMessageType.CREATED
         assert isinstance(message.model, Model)
-        task.cancel()
-        await task
+
+        if not task.cancel():
+            await task
 
     @pytest.mark.asyncio
     async def test_project_updated(
         self, subscription_client: SpeckleClient, test_project: Project
     ) -> None:
-        message: Optional[ProjectUpdatedMessage] = None
-
-        task = None
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectUpdatedMessage] = loop.create_future()
 
         def callback(d: ProjectUpdatedMessage):
-            nonlocal message
-            message = d
+            nonlocal future
+            future.set_result(d)
 
         task = asyncio.create_task(
             subscription_client.subscription.project_updated(callback, test_project.id)
         )
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time to subscription to be setup
+        await asyncio.sleep(
+            SETUP_TIME_SECONDS
+        )  # Give time for subscription to be triggered
 
         input = ProjectUpdateInput(id=test_project.id, name="This is my new name")
         created = subscription_client.project.update(input)
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time for subscription to be triggered
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
 
         assert isinstance(message, ProjectUpdatedMessage)
         assert message.id == created.id
         assert message.type == ProjectUpdatedMessageType.UPDATED
         assert isinstance(message.project, Project)
-        task.cancel()
-        await task
+        if not task.cancel():
+            await task
 
     @pytest.mark.asyncio
     async def test_project_versions_updated(
@@ -157,13 +197,12 @@ class TestSubscriptionResource:
         test_project: Project,
         test_model: Model,
     ) -> None:
-        message: Optional[ProjectVersionsUpdatedMessage] = None
-
-        task = None
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectVersionsUpdatedMessage] = loop.create_future()
 
         def callback(d: ProjectVersionsUpdatedMessage):
-            nonlocal message
-            message = d
+            nonlocal future
+            future.set_result(d)
 
         task = asyncio.create_task(
             subscription_client.subscription.project_versions_updated(
@@ -171,15 +210,181 @@ class TestSubscriptionResource:
             )
         )
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time to subscription to be setup
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
 
         created = create_version(subscription_client, test_project.id, test_model.id)
 
-        await asyncio.sleep(WAIT_PERIOD)  # Give time for subscription to be triggered
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
 
         assert isinstance(message, ProjectVersionsUpdatedMessage)
         assert message.id == created.id
         assert message.type == ProjectVersionsUpdatedMessageType.CREATED
         assert isinstance(message.version, Version)
-        task.cancel()
-        await task
+        if not task.cancel():
+            await task
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(
+        is_public(), reason="The public API does not support these tests"
+    )
+    async def test_project_model_ingestion_cancellation(
+        self,
+        subscription_client: SpeckleClient,
+        test_project: Project,
+        test_model_ingestion: ModelIngestion,
+    ) -> None:
+        assert not test_model_ingestion.cancellation_requested
+
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectModelIngestionUpdatedMessage] = (
+            loop.create_future()
+        )
+
+        def callback(d: ProjectModelIngestionUpdatedMessage):
+            nonlocal future
+            future.set_result(d)
+
+        task = asyncio.create_task(
+            subscription_client.subscription.project_model_ingestion_cancellation_requested(
+                callback, test_project.id, ingestion_id=test_model_ingestion.id
+            )
+        )
+
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
+
+        cancellation_request = ModelIngestionRequestCancellationInput(
+            ingestion_id=test_model_ingestion.id,
+            project_id=test_project.id,
+            cancellation_message="Please cancel",
+        )
+        created = subscription_client.model_ingestion.request_cancellation(
+            cancellation_request
+        )
+        assert created.id == test_model_ingestion.id
+        assert created.cancellation_requested
+        assert created.status_data.status == ModelIngestionStatus.PROCESSING
+
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
+
+        assert isinstance(message, ProjectModelIngestionUpdatedMessage)
+        assert message.model_ingestion.id == created.id
+        assert message.model_ingestion.cancellation_requested
+        assert (
+            message.type
+            == ProjectModelIngestionUpdatedMessageType.CANCELLATION_REQUESTED
+        )
+        assert created.status_data.status == ModelIngestionStatus.PROCESSING
+        if not task.cancel():
+            await task
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(
+        is_public(), reason="The public API does not support these tests"
+    )
+    async def test_project_model_ingestion_cancellation_isnt_triggered_by_updates(
+        self,
+        subscription_client: SpeckleClient,
+        test_project: Project,
+        test_model_ingestion: ModelIngestion,
+    ) -> None:
+        assert not test_model_ingestion.cancellation_requested
+
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectModelIngestionUpdatedMessage] = (
+            loop.create_future()
+        )
+
+        def callback(d: ProjectModelIngestionUpdatedMessage):
+            nonlocal future
+            future.set_result(d)
+
+        task = asyncio.create_task(
+            subscription_client.subscription.project_model_ingestion_cancellation_requested(
+                callback, test_project.id, ingestion_id=test_model_ingestion.id
+            )
+        )
+
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
+
+        cancellation_request = ModelIngestionUpdateInput(
+            ingestion_id=test_model_ingestion.id,
+            project_id=test_project.id,
+            progress=None,
+            progress_message="this is just an ordinary update",
+        )
+        created = subscription_client.model_ingestion.update_progress(
+            cancellation_request
+        )
+        assert created.id == test_model_ingestion.id
+        assert not created.cancellation_requested
+        assert created.status_data.status == ModelIngestionStatus.PROCESSING
+
+        await asyncio.sleep(MAX_WAIT_TIME_SECONDS)
+
+        assert (
+            not future.done()
+        )  # make sure the sub did not call back and resolve the future
+
+        if not task.cancel():
+            await task
+
+    @pytest.mark.asyncio
+    @pytest.mark.skipif(
+        is_public(), reason="The public API does not support these tests"
+    )
+    async def test_project_model_ingestion_updates(
+        self,
+        subscription_client: SpeckleClient,
+        test_project: Project,
+        test_model_ingestion: ModelIngestion,
+    ) -> None:
+        assert not test_model_ingestion.cancellation_requested
+
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[ProjectModelIngestionUpdatedMessage] = (
+            loop.create_future()
+        )
+
+        def callback(d: ProjectModelIngestionUpdatedMessage):
+            nonlocal future
+            future.set_result(d)
+
+        task = asyncio.create_task(
+            subscription_client.subscription.project_model_ingestion_updated(
+                callback,
+                input=ProjectModelIngestionSubscriptionInput(
+                    project_id=test_project.id,
+                    ingestion_reference=ModelIngestionReference(
+                        ingestion_id=test_model_ingestion.id, model_id=None
+                    ),
+                ),
+                # ingestion_id=test_model_ingestion.id,
+            )
+        )
+
+        await asyncio.sleep(SETUP_TIME_SECONDS)  # Give time to subscription to be setup
+
+        progress_message = "this is just an ordinary update"
+        cancellation_request = ModelIngestionUpdateInput(
+            ingestion_id=test_model_ingestion.id,
+            project_id=test_project.id,
+            progress=None,
+            progress_message=progress_message,
+        )
+        created = subscription_client.model_ingestion.update_progress(
+            cancellation_request
+        )
+        assert created.id == test_model_ingestion.id
+        assert not created.cancellation_requested
+        assert created.status_data.status == ModelIngestionStatus.PROCESSING
+
+        message = await asyncio.wait_for(future, timeout=MAX_WAIT_TIME_SECONDS)
+
+        assert isinstance(message, ProjectModelIngestionUpdatedMessage)
+        assert message.model_ingestion.id == created.id
+        assert not message.model_ingestion.cancellation_requested
+        assert message.type == ProjectModelIngestionUpdatedMessageType.UPDATED
+        assert message.model_ingestion.status_data.progress_message == progress_message
+        assert created.status_data.status == ModelIngestionStatus.PROCESSING
+        if not task.cancel():
+            await task

tests/integration/conftest.py

@@ -1,3 +1,4 @@
+import os
 import random
 import uuid
 from typing import Dict
@@ -29,6 +30,10 @@ def host() -> str:
     return "localhost:3000"
 
 
+def is_public() -> bool:
+    return os.getenv("IS_PUBLIC", "false").lower() == "true"
+
+
 def seed_user(host: str) -> Dict[str, str]:
     seed = uuid.uuid4().hex
     user_dict = {

tests/integration/test_metrics.py

@@ -0,0 +1,92 @@
+from typing import Any, Callable
+
+import pytest
+from pytest_httpserver import HTTPServer
+from requests import HTTPError
+from werkzeug import Request, Response
+
+from specklepy.core.api.client import SpeckleClient
+from specklepy.logging import metrics
+
+PATH = "/"
+
+
+def assert_common_properties(payload: Any) -> None:
+    assert payload["event"] == "SDK Action"
+    assert payload["properties"]["token"] == "acd87c5a50b56df91a795e999812a3a4"
+    assert payload["properties"]["type"] == "action"
+    assert payload["properties"]["server_id"]
+    assert payload["properties"]["distinct_id"]
+    assert payload["properties"]["hostApp"] == "python"
+    assert payload["properties"]["hostAppVersion"]
+    assert payload["properties"]["core_version"]
+
+
+def handler(extra_check: Callable[[Any], bool]) -> Callable[[Request], Response]:
+    def inner(request: Request) -> Response:
+        json = request.get_json()
+        payload = json[0]
+        assert_common_properties(payload)
+        assert extra_check(payload)
+        return Response("", 200)
+
+    return inner
+
+
+def test_metrics_track(httpserver: HTTPServer, client: SpeckleClient):
+    with ScopedMetricsSetup(httpserver.url_for(PATH)) as _:
+        # Test No email
+        httpserver.expect_oneshot_request(PATH, "post").respond_with_handler(
+            handler(lambda payload: "email" not in payload["properties"])
+        )
+        metrics.track("SDK Action", client.account, None, True, False)
+
+        # Test With email
+        httpserver.expect_oneshot_request(PATH, "post").respond_with_handler(
+            handler(
+                lambda payload: payload["properties"]["email"]
+                == client.account.userInfo.email
+            )
+        )
+        metrics.track("SDK Action", client.account, None, True, True)
+
+        # Test With custom value
+        httpserver.expect_oneshot_request(PATH, "post").respond_with_handler(
+            handler(
+                lambda payload: payload["properties"]["myCustomProp"] == "myCustomValue"
+            )
+        )
+        metrics.track(
+            "SDK Action", client.account, {"myCustomProp": "myCustomValue"}, True, True
+        )
+
+
+def test_metrics_errors(httpserver: HTTPServer):
+    with ScopedMetricsSetup(httpserver.url_for(PATH)) as _:
+        httpserver.expect_oneshot_request(PATH, "post").respond_with_data("", 400)
+
+        # Expect send_sync == true to mean mean it will raise
+        with pytest.raises(HTTPError):
+            metrics.track("SDK Action", send_sync=True)
+
+        # Expect send_sync == false to mean mean it won't
+        metrics.track("SDK Action")
+
+
+class ScopedMetricsSetup:
+    """
+    Scoped setup and tear down for enabling metrics tracking
+    """
+
+    tracker: metrics.MetricsTracker
+
+    def __init__(self, metrics_url: str):
+        self.tracker = metrics.initialise_tracker()
+        self.tracker.analytics_url = metrics_url
+
+    def __enter__(self):
+        metrics.enable()
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        metrics.disable()
+        metrics.METRICS_TRACKER = None

uv.lock

@@ -526,7 +526,7 @@ name = "exceptiongroup"
 version = "1.3.0"
 source = { registry = "https://pypi.org/simple/" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.13'" },
+    { name = "typing-extensions", marker = "python_full_version < '3.11'" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/0b/9f/a65090624ecf468cdca03533906e7c69ed7588582240cfe7cc9e770b50eb/exceptiongroup-1.3.0.tar.gz", hash = "sha256:b241f5885f560bc56a59ee63ca4c6a8bfa46ae4ad651af316d4e81817bb9fd88", size = 29749, upload-time = "2025-05-10T17:42:51.123Z" }
 wheels = [
@@ -1867,6 +1867,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ee/49/1377b49de7d0c1ce41292161ea0f721913fa8722c19fb9c1e3aa0367eecb/pytest_cov-7.0.0-py3-none-any.whl", hash = "sha256:3b8e9558b16cc1479da72058bdecf8073661c7f57f7d3c5f22a1c23507f2d861", size = 22424, upload-time = "2025-09-09T10:57:00.695Z" },
 ]
 
+[[package]]
+name = "pytest-httpserver"
+version = "1.1.3"
+source = { registry = "https://pypi.org/simple/" }
+dependencies = [
+    { name = "werkzeug" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/f1/d8/def15ba33bd696dd72dd4562a5287c0cba4d18a591eeb82e0b08ab385afc/pytest_httpserver-1.1.3.tar.gz", hash = "sha256:af819d6b533f84b4680b9416a5b3f67f1df3701f1da54924afd4d6e4ba5917ec", size = 68870, upload-time = "2025-04-10T08:17:15.6Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/0d/d2/dfc2f25f3905921c2743c300a48d9494d29032f1389fc142e718d6978fb2/pytest_httpserver-1.1.3-py3-none-any.whl", hash = "sha256:5f84757810233e19e2bb5287f3826a71c97a3740abe3a363af9155c0f82fdbb9", size = 21000, upload-time = "2025-04-10T08:17:13.906Z" },
+]
+
 [[package]]
 name = "pytest-ordering"
 version = "0.6"
@@ -2222,6 +2234,7 @@ dev = [
     { name = "pytest" },
     { name = "pytest-asyncio" },
     { name = "pytest-cov" },
+    { name = "pytest-httpserver" },
     { name = "pytest-ordering" },
     { name = "ruff" },
     { name = "types-deprecated" },
@@ -2259,6 +2272,7 @@ dev = [
     { name = "pytest", specifier = ">=8.3.4" },
     { name = "pytest-asyncio", specifier = ">=0.25.2" },
     { name = "pytest-cov", specifier = ">=6.0.0" },
+    { name = "pytest-httpserver", specifier = ">=1.1.3" },
     { name = "pytest-ordering", specifier = ">=0.6" },
     { name = "ruff", specifier = "==0.9.2" },
     { name = "types-deprecated", specifier = ">=1.2.15.20241117" },
@@ -2622,6 +2636,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/47/96/9d5749106ff57629b54360664ae7eb9afd8302fad1680ead385383e33746/websockets-11.0.3-py3-none-any.whl", hash = "sha256:6681ba9e7f8f3b19440921e99efbb40fc89f26cd71bf539e45d8c8a25c976dc6", size = 118056, upload-time = "2023-05-07T14:25:18.508Z" },
 ]
 
+[[package]]
+name = "werkzeug"
+version = "3.1.4"
+source = { registry = "https://pypi.org/simple/" }
+dependencies = [
+    { name = "markupsafe" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/45/ea/b0f8eeb287f8df9066e56e831c7824ac6bab645dd6c7a8f4b2d767944f9b/werkzeug-3.1.4.tar.gz", hash = "sha256:cd3cd98b1b92dc3b7b3995038826c68097dcb16f9baa63abe35f20eafeb9fe5e", size = 864687, upload-time = "2025-11-29T02:15:22.841Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2f/f9/9e082990c2585c744734f85bec79b5dae5df9c974ffee58fe421652c8e91/werkzeug-3.1.4-py3-none-any.whl", hash = "sha256:2ad50fb9ed09cc3af22c54698351027ace879a0b60a3b5edf5730b2f7d876905", size = 224960, upload-time = "2025-11-29T02:15:21.13Z" },
+]
+
 [[package]]
 name = "wrapt"
 version = "2.0.1"