SubscriptionResource

API Access class for subscriptions

Source code in src/specklepy/api/resources/current/subscription_resource.py

def __init__(self, account, basepath, client) -> None:
    super().__init__(
        account=account,
        basepath=basepath,
        client=client,
        name=NAME,
    )

user_projects_updated async

user_projects_updated(
    callback: Callable[[UserProjectsUpdatedMessage], None],
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

async def user_projects_updated(
    self, callback: Callable[[UserProjectsUpdatedMessage], None]
) -> None:
    request = gql(
        """
        subscription UserProjectsUpdated {
          data:userProjectsUpdated {
            id
            project {
              id
              name
              description
              visibility
              allowPublicComments
              role
              createdAt
              updatedAt
              sourceApps
              workspaceId
            }
            type
          }
        }
        """
    )

    await self.subscribe_2(
        DataResponse[UserProjectsUpdatedMessage],
        request,
        callback=lambda d: callback(d.data),
    )

project_models_updated async

project_models_updated(
    callback: Callable[[ProjectModelsUpdatedMessage], None],
    id: str,
    model_ids: Sequence[str] | None = None,
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

async def project_models_updated(
    self,
    callback: Callable[[ProjectModelsUpdatedMessage], None],
    id: str,
    model_ids: Sequence[str] | None = None,
) -> None:
    request = gql(
        """
        subscription ProjectModelsUpdated($id: String!, $modelIds: [String!]) {
          data:projectModelsUpdated(id: $id, modelIds: $modelIds) {
            id
            model {
              id
              name
              previewUrl
              updatedAt
              description
              displayName
              createdAt
              author {
                avatar
                bio
                company
                id
                name
                role
                verified
              }
            }
            type
          }
        }
        """
    )

    request.variable_values = {"id": id, "modelIds": model_ids}

    await self.subscribe_2(
        DataResponse[ProjectModelsUpdatedMessage],
        request,
        callback=lambda d: callback(d.data),
    )

project_updated async

project_updated(
    callback: Callable[[ProjectUpdatedMessage], None],
    id: str,
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

async def project_updated(
    self,
    callback: Callable[[ProjectUpdatedMessage], None],
    id: str,
) -> None:
    request = gql(
        """
        subscription ProjectUpdated($id: String!) {
          data:projectUpdated(id: $id) {
            id
            project {
              id
              name
              description
              visibility
              allowPublicComments
              role
              createdAt
              updatedAt
              sourceApps
              workspaceId
            }
            type
          }
        }
        """
    )

    request.variable_values = {"id": id}

    await self.subscribe_2(
        DataResponse[ProjectUpdatedMessage],
        request,
        callback=lambda d: callback(d.data),
    )

project_versions_updated async

project_versions_updated(
    callback: Callable[
        [ProjectVersionsUpdatedMessage], None
    ],
    id: str,
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

async def project_versions_updated(
    self,
    callback: Callable[[ProjectVersionsUpdatedMessage], None],
    id: str,
) -> None:
    request = gql(
        """
        subscription ProjectVersionsUpdated($id: String!) {
          data:projectVersionsUpdated(id: $id) {
            id
            modelId
            type
            version {
              id
              referencedObject
              message
              sourceApplication
              createdAt
              previewUrl
              authorUser {
                id
                name
                bio
                company
                verified
                role
                avatar
              }
            }
          }
        }
        """
    )

    request.variable_values = {"id": id}

    await self.subscribe_2(
        DataResponse[ProjectVersionsUpdatedMessage],
        request,
        callback=lambda d: callback(d.data),
    )

project_model_ingestion_updated async

project_model_ingestion_updated(
    callback: Callable[
        [ProjectModelIngestionUpdatedMessage], None
    ],
    input: ProjectModelIngestionSubscriptionInput,
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

async def project_model_ingestion_updated(
    self,
    callback: Callable[[ProjectModelIngestionUpdatedMessage], None],
    input: ProjectModelIngestionSubscriptionInput,
) -> None:
    request = gql(
        """
        subscription IngestionUpdated($input: ProjectModelIngestionSubscriptionInput!) {
          data: projectModelIngestionUpdated(input: $input) {
            modelIngestion {
              id
              createdAt
              updatedAt
              modelId
              projectId
              userId
              cancellationRequested
              statusData {
                ... on HasModelIngestionStatus {
                  status
                }
                ... on HasProgressMessage {
                  progressMessage
                }
                ... on ModelIngestionSuccessStatus
                {
                    versionId
                }
              }
            }
            type
          }
        }
        """  # noqa: E501
    )

    request.variable_values = {
        "input": input.model_dump(
            warnings="error", by_alias=True, exclude_none=True
        ),
    }

    await self.subscribe_2(
        DataResponse[ProjectModelIngestionUpdatedMessage],
        request,
        callback=lambda d: callback(d.data),
    )

project_model_ingestion_cancellation_requested async

project_model_ingestion_cancellation_requested(
    callback: Callable[
        [ProjectModelIngestionUpdatedMessage], None
    ],
    project_id: str,
    ingestion_id: str,
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

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,
        ),
    )

subscribe_2 async

subscribe_2(
    response_type: Type[TEventArgs],
    request: GraphQLRequest,
    callback: Callable[[TEventArgs], None],
) -> None

Source code in src/specklepy/api/resources/current/subscription_resource.py

@check_wsclient
async def subscribe_2(
    self,
    response_type: Type[TEventArgs],
    request: GraphQLRequest,
    callback: Callable[[TEventArgs], None],
) -> None:
    async with self.client as session:
        self.session = session
        gen = session.subscribe(request)
        async for res in gen:
            event_arg = response_type.model_validate(res)
            callback(event_arg)