huanld/specklepy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removes all FE1 client functions (#380)

Browse Source 
* Removes all FE1 client functions

* Removed usages of deprecated client functions

* removed trailing deprecated client function

* ruff

* Fixed last failing test
This commit is contained in:
Jedd Morgan
2025-02-18 15:32:14 +00:00
committed by GitHub
parent da6e2d92e0
commit d1b6755997
45 changed files with 112 additions and 4337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/specklepy/api/client.py
+6 -59
View File
@@ -1,7 +1,5 @@
import contextlib
from deprecated import deprecated
from specklepy.api.credentials import Account
from specklepy.api.resources import (
ActiveUserResource,
@@ -12,12 +10,6 @@ from specklepy.api.resources import (
ServerResource,
SubscriptionResource,
VersionResource,
branch,
commit,
object,
stream,
subscriptions,
user,
)
from specklepy.core.api.client import SpeckleClient as CoreSpeckleClient
from specklepy.logging import metrics
@@ -36,6 +28,7 @@ class SpeckleClient(CoreSpeckleClient):
```py
from specklepy.api.client import SpeckleClient
from specklepy.core.api.inputs.project_inputs import ProjectCreateInput
from specklepy.api.credentials import get_default_account
# initialise the client
@@ -47,11 +40,12 @@ class SpeckleClient(CoreSpeckleClient):
account = get_default_account()
client.authenticate_with_account(account)
# create a new stream. this returns the stream id
new_stream_id = client.stream.create(name="a shiny new stream")
# create a new project
input = ProjectCreateInput(name="a shiny new project")
project = self.project.create(input)
# use that stream id to get the stream from the server
new_stream = client.stream.get(id=new_stream_id)
# or, use a project id to get an existing project from the server
new_stream = client.project.get("abcdefghij")
```
"""
@@ -123,53 +117,6 @@ class SpeckleClient(CoreSpeckleClient):
client=self.wsclient,
# todo: why doesn't this take a server version
)
# Deprecated Resources
self.user = user.Resource(
account=self.account,
basepath=self.url,
client=self.httpclient,
server_version=server_version,
)
self.stream = stream.Resource(
account=self.account,
basepath=self.url,
client=self.httpclient,
server_version=server_version,
)
self.commit = commit.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.branch = branch.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.object = object.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.subscribe = subscriptions.Resource(
account=self.account,
basepath=self.ws_url,
client=self.wsclient,
)
@deprecated(
version="2.6.0",
reason=(
"Renamed: please use `authenticate_with_account` or"
" `authenticate_with_token` instead."
),
)
def authenticate(self, token: str) -> None:
"""Authenticate the client using a personal access token
The token is saved in the client object and a synchronous GraphQL
entrypoint is created
Arguments:
token {str} -- an api token
"""
metrics.track(
metrics.SDK, self.account, {"name": "Client Authenticate_deprecated"}
)
return super().authenticate(token)
def authenticate_with_token(self, token: str) -> None:
"""
src/specklepy/api/models/__init__.py
-20
View File
@@ -1,35 +1,15 @@
# following imports seem to be unnecessary, but they need to stay
# to not break the scripts using these functions as non-core
from specklepy.core.api.models import (
Activity,
ActivityCollection,
Branch,
Branches,
Collaborator,
Commit,
Commits,
LimitedUser,
Object,
PendingStreamCollaborator,
ServerInfo,
Stream,
Streams,
User,
)
__all__ = [
"Activity",
"ActivityCollection",
"Branch",
"Branches",
"Collaborator",
"Commit",
"Commits",
"LimitedUser",
"Object",
"PendingStreamCollaborator",
"ServerInfo",
"Stream",
"Streams",
"User",
]
src/specklepy/api/resources/__init__.py
-20
View File
@@ -8,17 +8,6 @@ from specklepy.api.resources.current.project_resource import ProjectResource
from specklepy.api.resources.current.server_resource import ServerResource
from specklepy.api.resources.current.subscription_resource import SubscriptionResource
from specklepy.api.resources.current.version_resource import VersionResource
from specklepy.api.resources.deprecated import (
active_user,
branch,
commit,
object,
other_user,
server,
stream,
subscriptions,
user,
)
__all__ = [
"ActiveUserResource",
@@ -29,13 +18,4 @@ __all__ = [
"ServerResource",
"SubscriptionResource",
"VersionResource",
"active_user",
"branch",
"commit",
"object",
"other_user",
"server",
"stream",
"subscriptions",
"user",
]
src/specklepy/api/resources/current/active_user_resource.py
+4 -96
View File
@@ -1,7 +1,4 @@
from datetime import datetime
from typing import List, Optional, overload
from deprecated import deprecated
from typing import List, Optional
from specklepy.core.api.inputs.user_inputs import UserProjectsFilter, UserUpdateInput
from specklepy.core.api.models import (
@@ -10,10 +7,6 @@ from specklepy.core.api.models import (
ResourceCollection,
User,
)
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources import ActiveUserResource as CoreResource
from specklepy.logging import metrics
@@ -35,40 +28,13 @@ class ActiveUserResource(CoreResource):
metrics.track(metrics.SDK, self.account, {"name": "Active User Get"})
return super().get()
@deprecated("Use UserUpdateInput overload", version=FE1_DEPRECATION_VERSION)
@overload
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
) -> User: ...
@overload
def update(self, *, input: UserUpdateInput) -> User: ...
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
*,
input: Optional[UserUpdateInput] = None,
input: UserUpdateInput,
) -> User:
metrics.track(metrics.SDK, self.account, {"name": "Active User Update"})
if isinstance(input, UserUpdateInput):
return super()._update(input=input)
else:
return super()._update(
input=UserUpdateInput(
name=name,
company=company,
bio=bio,
avatar=avatar,
)
)
return super().update(input=input)
def get_projects(
self,
@@ -85,61 +51,3 @@ class ActiveUserResource(CoreResource):
metrics.SDK, self.account, {"name": "Active User Get Project Invites"}
)
return super().get_project_invites()
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
):
"""
Fetches collection the current authenticated user's activity
as filtered by given parameters
Note: all timestamps arguments should be `datetime` of any tz as they will be
converted to UTC ISO format strings
Args:
limit (int): The maximum number of activity items to return.
action_type (Optional[str]): Filter results to a single action type.
before (Optional[datetime]): Latest cutoff for activity to include.
after (Optional[datetime]): Oldest cutoff for an activity to include.
cursor (Optional[datetime]): Timestamp cursor for pagination.
Returns:
Activity collection, filtered according to the provided parameters.
"""
metrics.track(metrics.SDK, self.account, {"name": "User Active Activity"})
return super().activity(limit, action_type, before, after, cursor)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_all_pending_invites(self) -> List[PendingStreamCollaborator]:
"""Fetches all of the current user's pending stream invitations.
Returns:
List[PendingStreamCollaborator]: A list of pending stream invitations.
"""
metrics.track(
metrics.SDK, self.account, {"name": "User Active Invites All Get"}
)
return super().get_all_pending_invites()
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_pending_invite(
self, stream_id: str, token: Optional[str] = None
) -> Optional[PendingStreamCollaborator]:
"""Fetches a specific pending invite for the current user on a given stream.
Args:
stream_id (str): The ID of the stream to look for invites on.
token (Optional[str]): The token of the invite to look for (optional).
Returns:
Optional[PendingStreamCollaborator]: The invite for the given stream,
or None if not found.
"""
metrics.track(metrics.SDK, self.account, {"name": "User Active Invite Get"})
return super().get_pending_invite(stream_id, token)
src/specklepy/api/resources/current/other_user_resource.py
+1 -64
View File
@@ -1,20 +1,11 @@
from datetime import datetime
from typing import List, Optional, Union
from deprecated import deprecated
from typing import Optional
from specklepy.core.api.models import (
ActivityCollection,
LimitedUser,
UserSearchResultCollection,
)
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources import OtherUserResource as CoreResource
from specklepy.logging import metrics
from specklepy.logging.exceptions import SpeckleException
class OtherUserResource(CoreResource):
@@ -52,57 +43,3 @@ class OtherUserResource(CoreResource):
return super().user_search(
query, limit=limit, cursor=cursor, archived=archived, emailOnly=emailOnly
)
@deprecated(reason="Use user_search instead", version=FE1_DEPRECATION_VERSION)
def search(
self, search_query: str, limit: int = 25
) -> Union[List[LimitedUser], SpeckleException]:
"""
Searches for users by name or email.
The search requires a minimum query length of 3 characters.
Args:
search_query (str): The search string.
limit (int): Maximum number of search results to return.
Returns:
Union[List[LimitedUser], SpeckleException]:
A list of users matching the search
query or an exception if the query is too short.
"""
if len(search_query) < 3:
return SpeckleException(
message="User search query must be at least 3 characters."
)
metrics.track(metrics.SDK, self.account, {"name": "Other User Search"})
return super().search(search_query, limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
user_id: str,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
) -> ActivityCollection:
"""
Retrieves a collection of activities for a specified user,
with optional filters for activity type, time frame, and pagination.
Args:
user_id (str): The ID of the user whose activities are being requested.
limit (int): The maximum number of activity items to return.
action_type (Optional[str]): A specific type of activity to filter.
before (Optional[datetime]): Latest timestamp to include activities before.
after (Optional[datetime]): Earliest timestamp to include activities after.
cursor (Optional[datetime]): Timestamp for pagination cursor.
Returns:
ActivityCollection: A collection of user activities filtered
according to specified criteria.
"""
metrics.track(metrics.SDK, self.account, {"name": "Other User Activity"})
return super().activity(user_id, limit, action_type, before, after, cursor)
src/specklepy/api/resources/deprecated/active_user.py
-9
View File
@@ -1,9 +0,0 @@
from deprecated import deprecated
from specklepy.api.resources import ActiveUserResource
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
@deprecated(reason="Renamed to ActiveUserResource", version=FE1_DEPRECATION_VERSION)
class Resource(ActiveUserResource):
"""Renamed to ActiveUserResource"""
src/specklepy/api/resources/deprecated/branch.py
-108
View File
@@ -1,108 +0,0 @@
from typing import Optional, Union
from deprecated import deprecated
from specklepy.api.models import Branch
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources.deprecated.branch import Resource as CoreResource
from specklepy.logging import metrics
from specklepy.logging.exceptions import SpeckleException
class Resource(CoreResource):
"""API Access class for branches"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
)
self.schema = Branch
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self, stream_id: str, name: str, description: str = "No description provided"
) -> str:
"""Create a new branch on this stream
Arguments:
name {str} -- the name of the new branch
description {str} -- a short description of the branch
Returns:
id {str} -- the newly created branch's id
"""
metrics.track(metrics.SDK, self.account, {"name": "Branch Create"})
return super().create(stream_id, name, description)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(
self, stream_id: str, name: str, commits_limit: int = 10
) -> Union[Branch, None, SpeckleException]:
"""Get a branch by name from a stream
Arguments:
stream_id {str} -- the id of the stream to get the branch from
name {str} -- the name of the branch to get
commits_limit {int} -- maximum number of commits to get
Returns:
Branch -- the fetched branch with its latest commits
"""
metrics.track(metrics.SDK, self.account, {"name": "Branch Get"})
return super().get(stream_id, name, commits_limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_id: str, branches_limit: int = 10, commits_limit: int = 10):
"""Get a list of branches from a given stream
Arguments:
stream_id {str} -- the id of the stream to get the branches from
branches_limit {int} -- maximum number of branches to get
commits_limit {int} -- maximum number of commits to get
Returns:
List[Branch] -- the branches on the stream
"""
metrics.track(metrics.SDK, self.account, {"name": "Branch List"})
return super().list(stream_id, branches_limit, commits_limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(
self,
stream_id: str,
branch_id: str,
name: Optional[str] = None,
description: Optional[str] = None,
):
"""Update a branch
Arguments:
stream_id {str} -- the id of the stream containing the branch to update
branch_id {str} -- the id of the branch to update
name {str} -- optional: the updated branch name
description {str} -- optional: the updated branch description
Returns:
bool -- True if update is successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Branch Update"})
return super().update(stream_id, branch_id, name, description)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, stream_id: str, branch_id: str):
"""Delete a branch
Arguments:
stream_id {str} -- the id of the stream containing the branch to delete
branch_id {str} -- the branch to delete
Returns:
bool -- True if deletion is successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Branch Delete"})
return super().delete(stream_id, branch_id)
src/specklepy/api/resources/deprecated/commit.py
-134
View File
@@ -1,134 +0,0 @@
from typing import List, Optional, Union
from deprecated import deprecated
from specklepy.api.models import Commit
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources.deprecated.commit import Resource as CoreResource
from specklepy.logging import metrics
from specklepy.logging.exceptions import SpeckleException
class Resource(CoreResource):
"""API Access class for commits"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
)
self.schema = Commit
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, stream_id: str, commit_id: str) -> Commit:
"""
Gets a commit given a stream and the commit id
Arguments:
stream_id {str} -- the stream where we can find the commit
commit_id {str} -- the id of the commit you want to get
Returns:
Commit -- the retrieved commit object
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit Get"})
return super().get(stream_id, commit_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_id: str, limit: int = 10) -> List[Commit]:
"""
Get a list of commits on a given stream
Arguments:
stream_id {str} -- the stream where the commits are
limit {int} -- the maximum number of commits to fetch (default = 10)
Returns:
List[Commit] -- a list of the most recent commit objects
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit List"})
return super().list(stream_id, limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self,
stream_id: str,
object_id: str,
branch_name: str = "main",
message: str = "",
source_application: str = "python",
parents: Optional[List[str]] = None,
) -> Union[str, SpeckleException]:
"""
Creates a commit on a branch
Arguments:
stream_id {str} -- the stream you want to commit to
object_id {str} -- the hash of your commit object
branch_name {str}
-- the name of the branch to commit to (defaults to "main")
message {str}
-- optional: a message to give more information about the commit
source_application{str}
-- optional: the application from which the commit was created
(defaults to "python")
parents {List[str]} -- optional: the id of the parent commits
Returns:
str -- the id of the created commit
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit Create"})
return super().create(
stream_id, object_id, branch_name, message, source_application, parents
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(self, stream_id: str, commit_id: str, message: str) -> bool:
"""
Update a commit
Arguments:
stream_id {str}
-- the id of the stream that contains the commit you'd like to update
commit_id {str} -- the id of the commit you'd like to update
message {str} -- the updated commit message
Returns:
bool -- True if the operation succeeded
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit Update"})
return super().update(stream_id, commit_id, message)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, stream_id: str, commit_id: str) -> bool:
"""
Delete a commit
Arguments:
stream_id {str}
-- the id of the stream that contains the commit you'd like to delete
commit_id {str} -- the id of the commit you'd like to delete
Returns:
bool -- True if the operation succeeded
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit Delete"})
return super().delete(stream_id, commit_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def received(
self,
stream_id: str,
commit_id: str,
source_application: str = "python",
message: Optional[str] = None,
) -> bool:
"""
Mark a commit object a received by the source application.
"""
metrics.track(metrics.SDK, self.account, {"name": "Commit Received"})
return super().received(stream_id, commit_id, source_application, message)
src/specklepy/api/resources/deprecated/object.py
-63
View File
@@ -1,63 +0,0 @@
from typing import Dict, List
from deprecated import deprecated
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources.deprecated.object import Resource as CoreResource
from specklepy.logging import metrics
from specklepy.objects.base import Base
class Resource(CoreResource):
"""API Access class for objects"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
)
self.schema = Base
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, stream_id: str, object_id: str) -> Base:
"""
Get a stream object
Arguments:
stream_id {str} -- the id of the stream for the object
object_id {str} -- the hash of the object you want to get
Returns:
Base -- the returned Base object
"""
metrics.track(metrics.SDK, self.account, {"name": "Object Get"})
return super().get(stream_id, object_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(self, stream_id: str, objects: List[Dict]) -> str:
"""
Not advised - generally, you want to use `operations.send()`.
Create a new object on a stream.
To send a base object, you can prepare it by running it through the
`BaseObjectSerializer.traverse_base()` function to get a valid (serialisable)
object to send.
NOTE: this does not create a commit - you can create one with
`SpeckleClient.commit.create`.
Dynamic fields will be located in the 'data' dict of the received `Base` object
Arguments:
stream_id {str} -- the id of the stream you want to send the object to
objects {List[Dict]}
-- a list of base dictionary objects (NOTE: must be json serialisable)
Returns:
str -- the id of the object
"""
metrics.track(metrics.SDK, self.account, {"name": "Object Create"})
return super().create(stream_id, objects)
src/specklepy/api/resources/deprecated/other_user.py
-11
View File
@@ -1,11 +0,0 @@
from deprecated import deprecated
from specklepy.api.resources import OtherUserResource
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
@deprecated(reason="Renamed to OtherUserResource", version=FE1_DEPRECATION_VERSION)
class Resource(OtherUserResource):
"""
Renamed to OtherUserResource
"""
src/specklepy/api/resources/deprecated/server.py
-9
View File
@@ -1,9 +0,0 @@
from deprecated import deprecated
from specklepy.api.resources import ServerResource
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
@deprecated(reason="Renamed to ServerResource", version=FE1_DEPRECATION_VERSION)
class Resource(ServerResource):
"""Renamed to ServerResource"""
src/specklepy/api/resources/deprecated/stream.py
-322
View File
@@ -1,322 +0,0 @@
from datetime import datetime
from typing import List, Optional
from deprecated import deprecated
from specklepy.api.models import PendingStreamCollaborator, Stream
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources.deprecated.stream import Resource as CoreResource
from specklepy.logging import metrics
class Resource(CoreResource):
"""API Access class for streams"""
def __init__(self, account, basepath, client, server_version) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
server_version=server_version,
)
self.schema = Stream
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, id: str, branch_limit: int = 10, commit_limit: int = 10) -> Stream:
"""Get the specified stream from the server
Arguments:
id {str} -- the stream id
branch_limit {int} -- the maximum number of branches to return
commit_limit {int} -- the maximum number of commits to return
Returns:
Stream -- the retrieved stream
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Get"})
return super().get(id, branch_limit, commit_limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_limit: int = 10) -> List[Stream]:
"""Get a list of the user's streams
Arguments:
stream_limit {int} -- The maximum number of streams to return
Returns:
List[Stream] -- A list of Stream objects
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream List"})
return super().list(stream_limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self,
name: str = "Anonymous Python Stream",
description: str = "No description provided",
is_public: bool = True,
) -> str:
"""Create a new stream
Arguments:
name {str} -- the name of the string
description {str} -- a short description of the stream
is_public {bool}
-- whether or not the stream can be viewed by anyone with the id
Returns:
id {str} -- the id of the newly created stream
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Create"})
return super().create(name, description, is_public)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(
self,
id: str,
name: Optional[str] = None,
description: Optional[str] = None,
is_public: Optional[bool] = None,
) -> bool:
"""Update an existing stream
Arguments:
id {str} -- the id of the stream to be updated
name {str} -- the name of the string
description {str} -- a short description of the stream
is_public {bool}
-- whether or not the stream can be viewed by anyone with the id
Returns:
bool -- whether the stream update was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Update"})
return super().update(id, name, description, is_public)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, id: str) -> bool:
"""Delete a stream given its id
Arguments:
id {str} -- the id of the stream to delete
Returns:
bool -- whether the deletion was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Delete"})
return super().delete(id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def search(
self,
search_query: str,
limit: int = 25,
branch_limit: int = 10,
commit_limit: int = 10,
):
"""Search for streams by name, description, or id
Arguments:
search_query {str} -- a string to search for
limit {int} -- the maximum number of results to return
branch_limit {int} -- the maximum number of branches to return
commit_limit {int} -- the maximum number of commits to return
Returns:
List[Stream] -- a list of Streams that match the search query
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Search"})
return super().search(search_query, limit, branch_limit, commit_limit)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def favorite(self, stream_id: str, favorited: bool = True):
"""Favorite or unfavorite the given stream.
Arguments:
stream_id {str} -- the id of the stream to favorite / unfavorite
favorited {bool}
-- whether to favorite (True) or unfavorite (False) the stream
Returns:
Stream -- the stream with its `id`, `name`, and `favoritedDate`
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Favorite"})
return super().favorite(stream_id, favorited)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_all_pending_invites(
self, stream_id: str
) -> List[PendingStreamCollaborator]:
"""Get all of the pending invites on a stream.
You must be a `stream:owner` to query this.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the stream id from which to get the pending invites
Returns:
List[PendingStreamCollaborator]
-- a list of pending invites for the specified stream
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Invite Get"})
return super().get_all_pending_invites(stream_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite(
self,
stream_id: str,
email: Optional[str] = None,
user_id: Optional[str] = None,
role: str = "stream:contributor", # should default be reviewer?
message: Optional[str] = None,
):
"""Invite someone to a stream using either their email or user id
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to invite the user to
email {str} -- the email of the user to invite (use this OR `user_id`)
user_id {str} -- the id of the user to invite (use this OR `email`)
role {str}
-- the role to assign to the user (defaults to `stream:contributor`)
message {str}
-- a message to send along with this invite to the specified user
Returns:
bool -- True if the operation was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Invite Create"})
return super().invite(stream_id, email, user_id, role, message)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_batch(
self,
stream_id: str,
emails: Optional[List[str]] = None,
user_ids: Optional[List[None]] = None,
message: Optional[str] = None,
) -> bool:
"""Invite a batch of users to a specified stream.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to invite the user to
emails {List[str]}
-- the email of the user to invite (use this and/or `user_ids`)
user_id {List[str]}
-- the id of the user to invite (use this and/or `emails`)
message {str}
-- a message to send along with this invite to the specified user
Returns:
bool -- True if the operation was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Invite Batch Create"})
return super().invite_batch(stream_id, emails, user_ids, message)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_cancel(self, stream_id: str, invite_id: str) -> bool:
"""Cancel an existing stream invite
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream invite
invite_id {str} -- the id of the invite to use
Returns:
bool -- true if the operation was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Invite Cancel"})
return super().invite_cancel(stream_id, invite_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_use(self, stream_id: str, token: str, accept: bool = True) -> bool:
"""Accept or decline a stream invite
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str}
-- the id of the stream for which the user has a pending invite
token {str} -- the token of the invite to use
accept {bool} -- whether or not to accept the invite (defaults to True)
Returns:
bool -- true if the operation was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Invite Use"})
return super().invite_use(stream_id, token, accept)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update_permission(self, stream_id: str, user_id: str, role: str):
"""Updates permissions for a user on a given stream
Valid for Speckle Server >=2.6.4
Arguments:
stream_id {str} -- the id of the stream to grant permissions to
user_id {str} -- the id of the user to grant permissions for
role {str} -- the role to grant the user
Returns:
bool -- True if the operation was successful
"""
metrics.track(
metrics.SDK,
self.account,
{"name": "Stream Permission Update", "role": role},
)
return super().update_permission(stream_id, user_id, role)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def revoke_permission(self, stream_id: str, user_id: str):
"""Revoke permissions from a user on a given stream
Arguments:
stream_id {str} -- the id of the stream to revoke permissions from
user_id {str} -- the id of the user to revoke permissions from
Returns:
bool -- True if the operation was successful
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Permission Revoke"})
return super().revoke_permission(stream_id, user_id)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
stream_id: str,
action_type: Optional[str] = None,
limit: int = 20,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
):
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
Note: all timestamps arguments should be `datetime` of any tz
as they will be converted to UTC ISO format strings
stream_id {str} -- the id of the stream to get activity from
action_type {str}
-- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime}
-- latest cutoff for activity (ie: return all activity _before_ this time)
after {datetime}
-- oldest cutoff for activity (ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
metrics.track(metrics.SDK, self.account, {"name": "Stream Activity"})
return super().activity(stream_id, action_type, limit, before, after, cursor)
src/specklepy/api/resources/deprecated/subscriptions.py
-107
View File
@@ -1,107 +0,0 @@
from typing import Callable, Dict, List, Optional, Union
from deprecated import deprecated
from graphql import DocumentNode
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resources.current.subscription_resource import check_wsclient
from specklepy.core.api.resources.deprecated.subscriptions import (
Resource as CoreResource,
)
from specklepy.logging import metrics
class Resource(CoreResource):
"""API Access class for subscriptions"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
@check_wsclient
async def stream_added(self, callback: Optional[Callable] = None):
"""Subscribes to new stream added event for your profile.
Use this to display an up-to-date list of streams.
Arguments:
callback {Callable[Stream]} -- a function that takes the updated stream
as an argument and executes each time a stream is added
Returns:
Stream -- the update stream
"""
metrics.track(metrics.SDK, self.account, {"name": "Subscription Stream Added"})
return super().stream_added(callback)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
@check_wsclient
async def stream_updated(self, id: str, callback: Optional[Callable] = None):
"""
Subscribes to stream updated event.
Use this in clients/components that pertain only to this stream.
Arguments:
id {str} -- the stream id of the stream to subscribe to
callback {Callable[Stream]}
-- a function that takes the updated stream
as an argument and executes each time the stream is updated
Returns:
Stream -- the update stream
"""
metrics.track(
metrics.SDK, self.account, {"name": "Subscription Stream Updated"}
)
return super().stream_updated(id, callback)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
@check_wsclient
async def stream_removed(self, callback: Optional[Callable] = None):
"""Subscribes to stream removed event for your profile.
Use this to display an up-to-date list of streams for your profile.
NOTE: If someone revokes your permissions on a stream,
this subscription will be triggered with an extra value of revokedBy
in the payload.
Arguments:
callback {Callable[Dict]}
-- a function that takes the returned dict as an argument
and executes each time a stream is removed
Returns:
dict -- dict containing 'id' of stream removed and optionally 'revokedBy'
"""
metrics.track(
metrics.SDK, self.account, {"name": "Subscription Stream Removed"}
)
return super().stream_removed(callback)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
@check_wsclient
async def subscribe(
self,
query: DocumentNode,
params: Optional[Dict] = None,
callback: Optional[Callable] = None,
return_type: Optional[Union[str, List]] = None,
schema=None,
parse_response: bool = True,
):
# if self.client.transport.websocket is None:
# TODO: add multiple subs to the same ws connection
async with self.client as session:
async for res in session.subscribe(query, variable_values=params):
res = self._step_into_response(response=res, return_type=return_type)
if parse_response:
res = self._parse_response(response=res, schema=schema)
if callback is not None:
callback(res)
else:
return res
src/specklepy/api/resources/deprecated/user.py
-153
View File
@@ -1,153 +0,0 @@
from datetime import datetime
from typing import List, Optional, Union
from deprecated import deprecated
from specklepy.api.models import PendingStreamCollaborator, User
from specklepy.core.api.resources.deprecated.user import Resource as CoreResource
from specklepy.logging import metrics
from specklepy.logging.exceptions import SpeckleException
DEPRECATION_VERSION = "2.9.0"
DEPRECATION_TEXT = (
"The user resource is deprecated, please use the active_user or other_user"
" resources"
)
class Resource(CoreResource):
"""API Access class for users"""
def __init__(self, account, basepath, client, server_version) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
server_version=server_version,
)
self.schema = User
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get(self, id: Optional[str] = None) -> User:
"""
Gets the profile of a user.
If no id argument is provided, will return the current authenticated
user's profile (as extracted from the authorization header).
Arguments:
id {str} -- the user id
Returns:
User -- the retrieved user
"""
metrics.track(metrics.SDK, self.account, {"name": "User Get_deprecated"})
return super().get(id)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def search(
self, search_query: str, limit: int = 25
) -> Union[List[User], SpeckleException]:
"""
Searches for user by name or email.
The search query must be at least 3 characters long
Arguments:
search_query {str} -- a string to search for
limit {int} -- the maximum number of results to return
Returns:
List[User] -- a list of User objects that match the search query
"""
metrics.track(metrics.SDK, self.account, {"name": "User Search_deprecated"})
return super().search(search_query, limit)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
):
"""Updates your user profile. All arguments are optional.
Arguments:
name {str} -- your name
company {str} -- the company you may or may not work for
bio {str} -- tell us about yourself
avatar {str} -- a nice photo of yourself
Returns:
bool -- True if your profile was updated successfully
"""
# metrics.track(metrics.USER, self.account, {"name": "update"})
metrics.track(metrics.SDK, self.account, {"name": "User Update_deprecated"})
return super().update(name, company, bio, avatar)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def activity(
self,
user_id: Optional[str] = None,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
):
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
If no id argument is provided, will return the current authenticated
user's activity (as extracted from the authorization header).
Note: all timestamps arguments should be `datetime` of any tz as
they will be converted to UTC ISO format strings
user_id {str} -- the id of the user to get the activity from
action_type {str} -- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime}
-- latest cutoff for activity (ie: return all activity _before_ this time)
after {datetime}
-- oldest cutoff for activity (ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
metrics.track(metrics.SDK, self.account, {"name": "User Activity_deprecated"})
return super().activity(user_id, limit, action_type, before, after, cursor)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get_all_pending_invites(self) -> List[PendingStreamCollaborator]:
"""Get all of the active user's pending stream invites
Requires Speckle Server version >= 2.6.4
Returns:
List[PendingStreamCollaborator]
-- a list of pending invites for the current user
"""
# metrics.track(metrics.INVITE, self.account, {"name": "get"})
metrics.track(
metrics.SDK, self.account, {"name": "User GetAllInvites_deprecated"}
)
return super().get_all_pending_invites()
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get_pending_invite(
self, stream_id: str, token: Optional[str] = None
) -> Optional[PendingStreamCollaborator]:
"""Get a particular pending invite for the active user on a given stream.
If no invite_id is provided, any valid invite will be returned.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to look for invites on
token {str} -- the token of the invite to look for (optional)
Returns:
PendingStreamCollaborator
-- the invite for the given stream (or None if it isn't found)
"""
# metrics.track(metrics.INVITE, self.account, {"name": "get"})
metrics.track(metrics.SDK, self.account, {"name": "User GetInvite_deprecated"})
return super().get_pending_invite(stream_id, token)
src/specklepy/api/wrapper.py
+3 -3
View File
@@ -10,7 +10,7 @@ class StreamWrapper(CoreStreamWrapper):
The `StreamWrapper` gives you some handy helpers to deal with urls and
get authenticated clients and transports.
Construct a `StreamWrapper` with a stream, branch, commit, or object URL.
Construct a `StreamWrapper` with a URL of a model, version, or object.
The corresponding ids will be stored
in the wrapper. If you have local accounts on the machine,
you can use the `get_account` and `get_client` methods
@@ -21,8 +21,8 @@ class StreamWrapper(CoreStreamWrapper):
```py
from specklepy.api.wrapper import StreamWrapper
# provide any stream, branch, commit, object, or globals url
wrapper = StreamWrapper("https://app.speckle.systems/streams/3073b96e86/commits/604bea8cc6")
# provide a url for a model, version, or object
wrapper = StreamWrapper("https://app.speckle.systems/projects/3073b96e86/models/0fe47c9dca@604bea8cc6")
# get an authenticated SpeckleClient if you have a local account for the server
client = wrapper.get_client()
src/specklepy/core/api/client.py
+7 -68
View File
@@ -3,14 +3,12 @@ import re
from typing import Dict
from warnings import warn
from deprecated import deprecated
from gql import Client
from gql.transport.exceptions import TransportServerError
from gql.transport.requests import RequestsHTTPTransport
from gql.transport.websockets import WebsocketsTransport
from specklepy.core.api import resources
from specklepy.core.api.credentials import Account, get_account_from_token
from specklepy.core.api.credentials import Account
from specklepy.core.api.resources import (
ActiveUserResource,
ModelResource,
@@ -20,12 +18,6 @@ from specklepy.core.api.resources import (
ServerResource,
SubscriptionResource,
VersionResource,
branch,
commit,
object,
stream,
subscriptions,
user,
)
from specklepy.logging import metrics
from specklepy.logging.exceptions import SpeckleException, SpeckleWarning
@@ -44,6 +36,7 @@ class SpeckleClient:
```py
from specklepy.api.client import SpeckleClient
from specklepy.core.api.inputs.project_inputs import ProjectCreateInput
from specklepy.api.credentials import get_default_account
# initialise the client
@@ -55,11 +48,12 @@ class SpeckleClient:
account = get_default_account()
client.authenticate_with_account(account)
# create a new stream. this returns the stream id
new_stream_id = client.stream.create(name="a shiny new stream")
# create a new project
input = ProjectCreateInput(name="a shiny new project")
project = self.project.create(input)
# use that stream id to get the stream from the server
new_stream = client.stream.get(id=new_stream_id)
# or, use a project id to get an existing project from the server
new_stream = client.project.get("abcdefghij")
```
"""
@@ -123,23 +117,6 @@ class SpeckleClient:
f" {self.account.token is not None} )"
)
@deprecated(
version="2.6.0",
reason=(
"Renamed: please use `authenticate_with_account` or"
" `authenticate_with_token` instead."
),
)
def authenticate(self, token: str) -> None:
"""Authenticate the client using a personal access token
The token is saved in the client object and a synchronous GraphQL
entrypoint is created
Arguments:
token {str} -- an api token
"""
self.authenticate_with_account(get_account_from_token(token))
def authenticate_with_token(self, token: str) -> None:
"""
Authenticate the client using a personal access token.
@@ -251,41 +228,3 @@ class SpeckleClient:
basepath=self.ws_url,
client=self.wsclient,
)
# Deprecated Resources
self.user = user.Resource(
account=self.account,
basepath=self.url,
client=self.httpclient,
server_version=server_version,
)
self.stream = stream.Resource(
account=self.account,
basepath=self.url,
client=self.httpclient,
server_version=server_version,
)
self.commit = commit.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.branch = branch.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.object = object.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
self.subscribe = subscriptions.Resource(
account=self.account,
basepath=self.ws_url,
client=self.wsclient,
)
def __getattr__(self, name):
try:
attr = getattr(resources, name)
return attr.Resource(
account=self.account, basepath=self.url, client=self.httpclient
)
except AttributeError as ex:
raise SpeckleException(
f"Method {name} is not supported by the SpeckleClient class"
) from ex
src/specklepy/core/api/models/__init__.py
-22
View File
@@ -17,18 +17,6 @@ from specklepy.core.api.models.current import (
UserSearchResultCollection,
Version,
)
from specklepy.core.api.models.deprecated import (
Activity,
ActivityCollection,
Branch,
Branches,
Collaborator,
Commit,
Commits,
Object,
Stream,
Streams,
)
from specklepy.core.api.models.subscription_messages import (
ProjectModelsUpdatedMessage,
ProjectUpdatedMessage,
@@ -58,14 +46,4 @@ __all__ = [
"ProjectModelsUpdatedMessage",
"ProjectUpdatedMessage",
"ProjectVersionsUpdatedMessage",
"Collaborator",
"Commit",
"Commits",
"Object",
"Branch",
"Branches",
"Stream",
"Streams",
"Activity",
"ActivityCollection",
]
src/specklepy/core/api/models/current.py
-2
View File
@@ -2,7 +2,6 @@ from datetime import datetime
from typing import Generic, List, Optional, TypeVar
from specklepy.core.api.enums import ProjectVisibility
from specklepy.core.api.models.deprecated import Streams
from specklepy.core.api.models.graphql_base_model import GraphQLBaseModel
T = TypeVar("T")
@@ -17,7 +16,6 @@ class User(GraphQLBaseModel):
avatar: Optional[str] = None
verified: Optional[bool] = None
role: Optional[str] = None
streams: Optional["Streams"] = None
def __repr__(self):
return (
src/specklepy/core/api/models/deprecated.py
-147
View File
@@ -1,147 +0,0 @@
from datetime import datetime
from typing import List, Optional
from deprecated import deprecated
from pydantic import BaseModel, Field
FE1_DEPRECATION_REASON = (
"Stream/Branch/Commit API is now deprecated, "
"Use the new Project/Model/Version API functions in Client"
)
FE1_DEPRECATION_VERSION = "2.20"
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Collaborator(BaseModel):
id: Optional[str] = None
name: Optional[str] = None
role: Optional[str] = None
avatar: Optional[str] = None
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Commit(BaseModel):
id: Optional[str] = None
message: Optional[str] = None
authorName: Optional[str] = None
authorId: Optional[str] = None
authorAvatar: Optional[str] = None
branchName: Optional[str] = None
createdAt: Optional[datetime] = None
sourceApplication: Optional[str] = None
referencedObject: Optional[str] = None
totalChildrenCount: Optional[int] = None
parents: Optional[List[str]] = None
def __repr__(self) -> str:
return (
f"Commit( id: {self.id}, message: {self.message}, referencedObject:"
f" {self.referencedObject}, authorName: {self.authorName}, branchName:"
f" {self.branchName}, createdAt: {self.createdAt} )"
)
def __str__(self) -> str:
return self.__repr__()
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Commits(BaseModel):
totalCount: Optional[int] = None
cursor: Optional[datetime] = None
items: List[Commit] = []
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Object(BaseModel):
id: Optional[str] = None
speckleType: Optional[str] = None
applicationId: Optional[str] = None
totalChildrenCount: Optional[int] = None
createdAt: Optional[datetime] = None
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Branch(BaseModel):
id: Optional[str] = None
name: Optional[str] = None
description: Optional[str] = None
commits: Optional[Commits] = None
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Branches(BaseModel):
totalCount: Optional[int] = None
cursor: Optional[datetime] = None
items: List[Branch] = []
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Stream(BaseModel):
id: Optional[str] = None
name: Optional[str] = None
role: Optional[str] = None
isPublic: Optional[bool] = None
description: Optional[str] = None
createdAt: Optional[datetime] = None
updatedAt: Optional[datetime] = None
collaborators: List[Collaborator] = Field(default_factory=list)
branches: Optional[Branches] = None
commit: Optional[Commit] = None
object: Optional[Object] = None
commentCount: Optional[int] = None
favoritedDate: Optional[datetime] = None
favoritesCount: Optional[int] = None
def __repr__(self):
return (
f"Stream( id: {self.id}, name: {self.name}, description:"
f" {self.description}, isPublic: {self.isPublic})"
)
def __str__(self) -> str:
return self.__repr__()
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Streams(BaseModel):
totalCount: Optional[int] = None
cursor: Optional[datetime] = None
items: List[Stream] = []
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class Activity(BaseModel):
actionType: Optional[str] = None
info: Optional[dict] = None
userId: Optional[str] = None
streamId: Optional[str] = None
resourceId: Optional[str] = None
resourceType: Optional[str] = None
message: Optional[str] = None
time: Optional[datetime] = None
def __repr__(self) -> str:
return (
f"Activity( streamId: {self.streamId}, actionType: {self.actionType},"
f" message: {self.message}, userId: {self.userId} )"
)
def __str__(self) -> str:
return self.__repr__()
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
class ActivityCollection(BaseModel):
totalCount: Optional[int] = None
items: Optional[List[Activity]] = None
cursor: Optional[datetime] = None
def __repr__(self) -> str:
return (
f"ActivityCollection( totalCount: {self.totalCount}, items:"
f" {len(self.items) if self.items else 0}, cursor:"
f" {self.cursor.isoformat() if self.cursor else None} )"
)
def __str__(self) -> str:
return self.__repr__()
src/specklepy/core/api/models/subscription_messages.py
+1 -1
View File
@@ -31,5 +31,5 @@ class ProjectUpdatedMessage(GraphQLBaseModel):
class ProjectVersionsUpdatedMessage(GraphQLBaseModel):
id: str
type: ProjectVersionsUpdatedMessageType
model_id: Optional[str]
model_id: str
version: Optional[Version]
src/specklepy/core/api/resources/__init__.py
-20
View File
@@ -10,17 +10,6 @@ from specklepy.core.api.resources.current.subscription_resource import (
SubscriptionResource,
)
from specklepy.core.api.resources.current.version_resource import VersionResource
from specklepy.core.api.resources.deprecated import (
active_user,
branch,
commit,
object,
other_user,
server,
stream,
subscriptions,
user,
)
__all__ = [
"ActiveUserResource",
@@ -31,13 +20,4 @@ __all__ = [
"ServerResource",
"SubscriptionResource",
"VersionResource",
"active_user",
"branch",
"commit",
"object",
"other_user",
"server",
"stream",
"subscriptions",
"user",
]
src/specklepy/core/api/resources/current/active_user_resource.py
+2 -224
View File
@@ -1,21 +1,14 @@
from datetime import datetime, timezone
from typing import List, Optional, overload
from typing import List, Optional
from deprecated import deprecated
from gql import gql
from specklepy.core.api.inputs.user_inputs import UserProjectsFilter, UserUpdateInput
from specklepy.core.api.models import (
ActivityCollection,
PendingStreamCollaborator,
Project,
ResourceCollection,
User,
)
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.core.api.responses import DataResponse
from specklepy.logging.exceptions import GraphQLException
@@ -67,7 +60,7 @@ class ActiveUserResource(ResourceBase):
DataResponse[Optional[User]], QUERY, variables
).data
def _update(self, input: UserUpdateInput) -> User:
def update(self, input: UserUpdateInput) -> User:
QUERY = gql(
"""
mutation ActiveUserMutations($input: UserUpdateInput!) {
@@ -93,40 +86,6 @@ class ActiveUserResource(ResourceBase):
DataResponse[DataResponse[User]], QUERY, variables
).data.data
@deprecated("Use UserUpdateInput overload", version=FE1_DEPRECATION_VERSION)
@overload
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
) -> User: ...
@overload
def update(self, *, input: UserUpdateInput) -> User: ...
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
*,
input: Optional[UserUpdateInput] = None,
) -> User:
if isinstance(input, UserUpdateInput):
return self._update(input=input)
else:
return self._update(
input=UserUpdateInput(
name=name,
company=company,
bio=bio,
avatar=avatar,
)
)
def get_projects(
self,
*,
@@ -231,184 +190,3 @@ class ActiveUserResource(ResourceBase):
)
return response.data.data
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
) -> ActivityCollection:
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
If no id argument is provided, will return the current authenticated user's
activity (as extracted from the authorization header).
Note: all timestamps arguments should be `datetime` of any tz as they will be
converted to UTC ISO format strings
user_id {str} -- the id of the user to get the activity from
action_type {str} -- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime} -- latest cutoff for activity
(ie: return all activity _before_ this time)
after {datetime} -- oldest cutoff for activity
(ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
query = gql(
"""
query UserActivity(
$action_type: String,
$before:DateTime,
$after: DateTime,
$cursor: DateTime,
$limit: Int
){
activeUser {
activity(
actionType: $action_type,
before: $before,
after: $after,
cursor: $cursor,
limit: $limit
) {
totalCount
cursor
items {
actionType
info
userId
streamId
resourceId
resourceType
message
time
}
}
}
}
"""
)
params = {
"limit": limit,
"action_type": action_type,
"before": before.astimezone(timezone.utc).isoformat() if before else before,
"after": after.astimezone(timezone.utc).isoformat() if after else after,
"cursor": cursor.astimezone(timezone.utc).isoformat() if cursor else cursor,
}
return self.make_request(
query=query,
params=params,
return_type=["activeUser", "activity"],
schema=ActivityCollection,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_all_pending_invites(self) -> List[PendingStreamCollaborator]:
"""Get all of the active user's pending stream invites
Requires Speckle Server version >= 2.6.4
Returns:
List[PendingStreamCollaborator]
-- a list of pending invites for the current user
"""
self._check_invites_supported()
query = gql(
"""
query StreamInvites {
streamInvites{
id
token
inviteId
streamId
streamName
projectId
projectName
title
role
invitedBy {
id
name
bio
company
avatar
verified
role
}
}
}
"""
)
return self.make_request(
query=query,
return_type="streamInvites",
schema=PendingStreamCollaborator,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_pending_invite(
self, stream_id: str, token: Optional[str] = None
) -> Optional[PendingStreamCollaborator]:
"""Get a particular pending invite for the active user on a given stream.
If no invite_id is provided, any valid invite will be returned.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to look for invites on
token {str} -- the token of the invite to look for (optional)
Returns:
PendingStreamCollaborator
-- the invite for the given stream (or None if it isn't found)
"""
self._check_invites_supported()
query = gql(
"""
query StreamInvite($streamId: String!, $token: String) {
streamInvite(streamId: $streamId, token: $token) {
id
token
inviteId
streamId
streamName
projectId
projectName
title
role
invitedBy {
id
name
bio
company
avatar
verified
role
}
}
}
"""
)
params = {"streamId": stream_id}
if token:
params["token"] = token
return self.make_request(
query=query,
params=params,
return_type="streamInvite",
schema=PendingStreamCollaborator,
)
src/specklepy/core/api/resources/current/other_user_resource.py
+1 -130
View File
@@ -1,21 +1,13 @@
from datetime import datetime, timezone
from typing import List, Optional, Union
from typing import Optional
from deprecated import deprecated
from gql import gql
from specklepy.core.api.models import (
ActivityCollection,
LimitedUser,
UserSearchResultCollection,
)
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.core.api.responses import DataResponse
from specklepy.logging.exceptions import SpeckleException
NAME = "other_user"
@@ -130,124 +122,3 @@ class OtherUserResource(ResourceBase):
return self.make_request_and_parse_response(
DataResponse[UserSearchResultCollection], QUERY, variables
).data
@deprecated(reason="Use user_search instead", version=FE1_DEPRECATION_VERSION)
def search(
self, search_query: str, limit: int = 25
) -> Union[List[LimitedUser], SpeckleException]:
"""Searches for user by name or email. The search query must be at least
3 characters long
Arguments:
search_query {str} -- a string to search for
limit {int} -- the maximum number of results to return
Returns:
List[LimitedUser] -- a list of User objects that match the search query
"""
if len(search_query) < 3:
return SpeckleException(
message="User search query must be at least 3 characters"
)
query = gql(
"""
query UserSearch($search_query: String!, $limit: Int!) {
userSearch(query: $search_query, limit: $limit) {
items {
id
name
bio
company
avatar
verified
role
}
}
}
"""
)
params = {"search_query": search_query, "limit": limit}
return self.make_request(
query=query, params=params, return_type=["userSearch", "items"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
user_id: str,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
) -> ActivityCollection:
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
Note: all timestamps arguments should be `datetime` of
any tz as they will be converted to UTC ISO format strings
user_id {str} -- the id of the user to get the activity from
action_type {str} -- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime} -- latest cutoff for activity
(ie: return all activity _before_ this time)
after {datetime} -- oldest cutoff for activity
(ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
query = gql(
"""
query UserActivity(
$user_id: String!,
$action_type: String,
$before:DateTime,
$after: DateTime,
$cursor: DateTime,
$limit: Int
){
otherUser(id: $user_id) {
activity(
actionType: $action_type,
before: $before,
after: $after,
cursor: $cursor,
limit: $limit
) {
totalCount
cursor
items {
actionType
info
userId
streamId
resourceId
resourceType
message
time
}
}
}
}
"""
)
params = {
"user_id": user_id,
"limit": limit,
"action_type": action_type,
"before": before.astimezone(timezone.utc).isoformat() if before else before,
"after": after.astimezone(timezone.utc).isoformat() if after else after,
"cursor": cursor.astimezone(timezone.utc).isoformat() if cursor else cursor,
}
return self.make_request(
query=query,
params=params,
return_type=["otherUser", "activity"],
schema=ActivityCollection,
)
src/specklepy/core/api/resources/deprecated/active_user.py
-15
View File
@@ -1,15 +0,0 @@
from deprecated import deprecated
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
from specklepy.core.api.resources import ActiveUserResource
@deprecated(
reason="Class renamed to ActiveUserResource", version=FE1_DEPRECATION_VERSION
)
class Resource(ActiveUserResource):
"""
Class renamed to ActiveUserResource
"""
pass
src/specklepy/core/api/resources/deprecated/branch.py
-235
View File
@@ -1,235 +0,0 @@
from typing import Optional
from deprecated import deprecated
from gql import gql
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
Branch,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.logging.exceptions import SpeckleException
NAME = "branch"
class Resource(ResourceBase):
"""
API Access class for branches
Branch resource is deprecated, please use model resource instead
"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
)
self.schema = Branch
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self, stream_id: str, name: str, description: str = "No description provided"
) -> str:
"""Create a new branch on this stream
Arguments:
name {str} -- the name of the new branch
description {str} -- a short description of the branch
Returns:
id {str} -- the newly created branch's id
"""
query = gql(
"""
mutation BranchCreate($branch: BranchCreateInput!) {
branchCreate(branch: $branch)
}
"""
)
if len(name) < 3:
return SpeckleException(message="Branch Name must be at least 3 characters")
params = {
"branch": {
"streamId": stream_id,
"name": name,
"description": description,
}
}
return self.make_request(
query=query, params=params, return_type="branchCreate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, stream_id: str, name: str, commits_limit: int = 10):
"""Get a branch by name from a stream
Arguments:
stream_id {str} -- the id of the stream to get the branch from
name {str} -- the name of the branch to get
commits_limit {int} -- maximum number of commits to get
Returns:
Branch -- the fetched branch with its latest commits
"""
query = gql(
"""
query BranchGet($stream_id: String!, $name: String!, $commits_limit: Int!) {
stream(id: $stream_id) {
branch(name: $name) {
id,
name,
description,
commits (limit: $commits_limit) {
totalCount,
cursor,
items {
id,
referencedObject,
sourceApplication,
totalChildrenCount,
message,
authorName,
authorId,
branchName,
parents,
createdAt
}
}
}
}
}
"""
)
params = {"stream_id": stream_id, "name": name, "commits_limit": commits_limit}
return self.make_request(
query=query, params=params, return_type=["stream", "branch"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_id: str, branches_limit: int = 10, commits_limit: int = 10):
"""Get a list of branches from a given stream
Arguments:
stream_id {str} -- the id of the stream to get the branches from
branches_limit {int} -- maximum number of branches to get
commits_limit {int} -- maximum number of commits to get
Returns:
List[Branch] -- the branches on the stream
"""
query = gql(
"""
query BranchesGet(
$stream_id: String!,
$branches_limit: Int!,
$commits_limit: Int!
) {
stream(id: $stream_id) {
branches(limit: $branches_limit) {
items {
id
name
description
commits(limit: $commits_limit) {
totalCount
items{
id
message
referencedObject
sourceApplication
parents
authorId
authorName
branchName
createdAt
}
}
}
}
}
}
"""
)
params = {
"stream_id": stream_id,
"branches_limit": branches_limit,
"commits_limit": commits_limit,
}
return self.make_request(
query=query, params=params, return_type=["stream", "branches", "items"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(
self,
stream_id: str,
branch_id: str,
name: Optional[str] = None,
description: Optional[str] = None,
):
"""Update a branch
Arguments:
stream_id {str} -- the id of the stream containing the branch to update
branch_id {str} -- the id of the branch to update
name {str} -- optional: the updated branch name
description {str} -- optional: the updated branch description
Returns:
bool -- True if update is successful
"""
query = gql(
"""
mutation BranchUpdate($branch: BranchUpdateInput!) {
branchUpdate(branch: $branch)
}
"""
)
params = {
"branch": {
"streamId": stream_id,
"id": branch_id,
}
}
if name:
params["branch"]["name"] = name
if description:
params["branch"]["description"] = description
return self.make_request(
query=query, params=params, return_type="branchUpdate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, stream_id: str, branch_id: str):
"""Delete a branch
Arguments:
stream_id {str} -- the id of the stream containing the branch to delete
branch_id {str} -- the branch to delete
Returns:
bool -- True if deletion is successful
"""
query = gql(
"""
mutation BranchDelete($branch: BranchDeleteInput!) {
branchDelete(branch: $branch)
}
"""
)
params = {"branch": {"streamId": stream_id, "id": branch_id}}
return self.make_request(
query=query, params=params, return_type="branchDelete", parse_response=False
)
src/specklepy/core/api/resources/deprecated/commit.py
-252
View File
@@ -1,252 +0,0 @@
from typing import List, Optional, Union
from deprecated import deprecated
from gql import gql
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
Commit,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.logging.exceptions import SpeckleException
NAME = "commit"
class Resource(ResourceBase):
"""
API Access class for commits
Commit resource is deprecated, please use version resource instead
"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
)
self.schema = Commit
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, stream_id: str, commit_id: str) -> Commit:
"""
Gets a commit given a stream and the commit id
Arguments:
stream_id {str} -- the stream where we can find the commit
commit_id {str} -- the id of the commit you want to get
Returns:
Commit -- the retrieved commit object
"""
query = gql(
"""
query Commit($stream_id: String!, $commit_id: String!) {
stream(id: $stream_id) {
commit(id: $commit_id) {
id
message
referencedObject
authorId
authorName
authorAvatar
branchName
createdAt
sourceApplication
totalChildrenCount
parents
}
}
}
"""
)
params = {"stream_id": stream_id, "commit_id": commit_id}
return self.make_request(
query=query, params=params, return_type=["stream", "commit"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_id: str, limit: int = 10) -> List[Commit]:
"""
Get a list of commits on a given stream
Arguments:
stream_id {str} -- the stream where the commits are
limit {int} -- the maximum number of commits to fetch (default = 10)
Returns:
List[Commit] -- a list of the most recent commit objects
"""
query = gql(
"""
query Commits($stream_id: String!, $limit: Int!) {
stream(id: $stream_id) {
commits(limit: $limit) {
items {
id
message
referencedObject
authorName
authorId
authorName
authorAvatar
branchName
createdAt
sourceApplication
totalChildrenCount
parents
}
}
}
}
"""
)
params = {"stream_id": stream_id, "limit": limit}
return self.make_request(
query=query, params=params, return_type=["stream", "commits", "items"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self,
stream_id: str,
object_id: str,
branch_name: str = "main",
message: str = "",
source_application: str = "python",
parents: Optional[List[str]] = None,
) -> Union[str, SpeckleException]:
"""
Creates a commit on a branch
Arguments:
stream_id {str} -- the stream you want to commit to
object_id {str} -- the hash of your commit object
branch_name {str}
-- the name of the branch to commit to (defaults to "main")
message {str}
-- optional: a message to give more information about the commit
source_application{str}
-- optional: the application from which the commit was created
(defaults to "python")
parents {List[str]} -- optional: the id of the parent commits
Returns:
str -- the id of the created commit
"""
query = gql(
"""
mutation CommitCreate ($commit: CommitCreateInput!)
{ commitCreate(commit: $commit)}
"""
)
params = {
"commit": {
"streamId": stream_id,
"branchName": branch_name,
"objectId": object_id,
"message": message,
"sourceApplication": source_application,
}
}
if parents:
params["commit"]["parents"] = parents
return self.make_request(
query=query, params=params, return_type="commitCreate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(self, stream_id: str, commit_id: str, message: str) -> bool:
"""
Update a commit
Arguments:
stream_id {str}
-- the id of the stream that contains the commit you'd like to update
commit_id {str} -- the id of the commit you'd like to update
message {str} -- the updated commit message
Returns:
bool -- True if the operation succeeded
"""
query = gql(
"""
mutation CommitUpdate($commit: CommitUpdateInput!)
{ commitUpdate(commit: $commit)}
"""
)
params = {
"commit": {"streamId": stream_id, "id": commit_id, "message": message}
}
return self.make_request(
query=query, params=params, return_type="commitUpdate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, stream_id: str, commit_id: str) -> bool:
"""
Delete a commit
Arguments:
stream_id {str}
-- the id of the stream that contains the commit you'd like to delete
commit_id {str} -- the id of the commit you'd like to delete
Returns:
bool -- True if the operation succeeded
"""
query = gql(
"""
mutation CommitDelete($commit: CommitDeleteInput!)
{ commitDelete(commit: $commit)}
"""
)
params = {"commit": {"streamId": stream_id, "id": commit_id}}
return self.make_request(
query=query, params=params, return_type="commitDelete", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def received(
self,
stream_id: str,
commit_id: str,
source_application: str = "python",
message: Optional[str] = None,
) -> bool:
"""
Mark a commit object a received by the source application.
"""
query = gql(
"""
mutation CommitReceive($receivedInput:CommitReceivedInput!){
commitReceive(input:$receivedInput)
}
"""
)
params = {
"receivedInput": {
"sourceApplication": source_application,
"streamId": stream_id,
"commitId": commit_id,
"message": "message",
}
}
try:
return self.make_request(
query=query,
params=params,
return_type="commitReceive",
parse_response=False,
)
except Exception as ex:
print(ex.with_traceback)
return False
src/specklepy/core/api/resources/deprecated/object.py
-92
View File
@@ -1,92 +0,0 @@
from typing import Dict, List
from gql import gql
from specklepy.core.api.resource import ResourceBase
from specklepy.objects.base import Base
NAME = "object"
class Resource(ResourceBase):
"""API Access class for objects"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
)
self.schema = Base
def get(self, stream_id: str, object_id: str) -> Base:
"""
Get a stream object
Arguments:
stream_id {str} -- the id of the stream for the object
object_id {str} -- the hash of the object you want to get
Returns:
Base -- the returned Base object
"""
query = gql(
"""
query Object($stream_id: String!, $object_id: String!) {
stream(id: $stream_id) {
id
name
object(id: $object_id) {
id
speckleType
applicationId
createdAt
totalChildrenCount
data
}
}
}
"""
)
params = {"stream_id": stream_id, "object_id": object_id}
return self.make_request(
query=query,
params=params,
return_type=["stream", "object", "data"],
)
def create(self, stream_id: str, objects: List[Dict]) -> str:
"""
Not advised - generally, you want to use `operations.send()`.
Create a new object on a stream.
To send a base object, you can prepare it by running it through the
`BaseObjectSerializer.traverse_base()` function to get a valid (serialisable)
object to send.
NOTE: this does not create a commit - you can create one with
`SpeckleClient.commit.create`.
Dynamic fields will be located in the 'data' dict of the received `Base` object
Arguments:
stream_id {str} -- the id of the stream you want to send the object to
objects {List[Dict]}
-- a list of base dictionary objects (NOTE: must be json serialisable)
Returns:
str -- the id of the object
"""
query = gql(
"""
mutation ObjectCreate($object_input: ObjectCreateInput!) {
objectCreate(objectInput: $object_input)
}
"""
)
params = {"object_input": {"streamId": stream_id, "objects": objects}}
return self.make_request(
query=query, params=params, return_type="objectCreate", parse_response=False
)
src/specklepy/core/api/resources/deprecated/other_user.py
-15
View File
@@ -1,15 +0,0 @@
from deprecated import deprecated
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
from specklepy.core.api.resources import OtherUserResource
@deprecated(
reason="Class renamed to OtherUserResource", version=FE1_DEPRECATION_VERSION
)
class Resource(OtherUserResource):
"""
Class renamed to OtherUserResource
"""
pass
src/specklepy/core/api/resources/deprecated/server.py
-11
View File
@@ -1,11 +0,0 @@
from deprecated import deprecated
from specklepy.core.api.models.deprecated import FE1_DEPRECATION_VERSION
from specklepy.core.api.resources import ServerResource
NAME = "server"
@deprecated(reason="Renamed to ServerResource", version=FE1_DEPRECATION_VERSION)
class Resource(ServerResource):
"""API Access class for the server"""
src/specklepy/core/api/resources/deprecated/stream.py
-785
View File
@@ -1,785 +0,0 @@
from datetime import datetime, timezone
from typing import List, Optional
from deprecated import deprecated
from gql import gql
from specklepy.core.api.models import (
ActivityCollection,
PendingStreamCollaborator,
Stream,
)
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.logging.exceptions import SpeckleException, UnsupportedException
NAME = "stream"
class Resource(ResourceBase):
"""
API Access class for streams
Stream resource is deprecated, please use project resource instead
"""
def __init__(self, account, basepath, client, server_version) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
server_version=server_version,
)
self.schema = Stream
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get(self, id: str, branch_limit: int = 10, commit_limit: int = 10) -> Stream:
"""Get the specified stream from the server
Arguments:
id {str} -- the stream id
branch_limit {int} -- the maximum number of branches to return
commit_limit {int} -- the maximum number of commits to return
Returns:
Stream -- the retrieved stream
"""
query = gql(
"""
query Stream($id: String!, $branch_limit: Int!, $commit_limit: Int!) {
stream(id: $id) {
id
name
role
description
isPublic
createdAt
updatedAt
commentCount
favoritesCount
collaborators {
id
name
role
avatar
}
branches(limit: $branch_limit) {
totalCount
cursor
items {
id
name
description
commits(limit: $commit_limit) {
totalCount
cursor
items {
id
message
authorId
createdAt
authorName
referencedObject
sourceApplication
}
}
}
}
}
}
"""
)
params = {"id": id, "branch_limit": branch_limit, "commit_limit": commit_limit}
return self.make_request(query=query, params=params, return_type="stream")
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def list(self, stream_limit: int = 10) -> List[Stream]:
"""Get a list of the user's streams
Arguments:
stream_limit {int} -- The maximum number of streams to return
Returns:
List[Stream] -- A list of Stream objects
"""
query = gql(
"""
query User($stream_limit: Int!) {
activeUser {
id
bio
name
email
avatar
company
verified
profiles
role
streams(limit: $stream_limit) {
totalCount
cursor
items {
id
name
role
isPublic
createdAt
updatedAt
description
commentCount
favoritesCount
collaborators {
id
name
role
}
}
}
}
}
"""
)
params = {"stream_limit": stream_limit}
return self.make_request(
query=query, params=params, return_type=["activeUser", "streams", "items"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def create(
self,
name: str = "Anonymous Python Stream",
description: str = "No description provided",
is_public: bool = True,
) -> str:
"""Create a new stream
Arguments:
name {str} -- the name of the string
description {str} -- a short description of the stream
is_public {bool}
-- whether or not the stream can be viewed by anyone with the id
Returns:
id {str} -- the id of the newly created stream
"""
query = gql(
"""
mutation StreamCreate($stream: StreamCreateInput!) {
streamCreate(stream: $stream)
}
"""
)
if len(name) < 3 and len(name) != 0:
return SpeckleException(message="Stream Name must be at least 3 characters")
params = {
"stream": {"name": name, "description": description, "isPublic": is_public}
}
return self.make_request(
query=query, params=params, return_type="streamCreate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update(
self,
id: str,
name: Optional[str] = None,
description: Optional[str] = None,
is_public: Optional[bool] = None,
) -> bool:
"""Update an existing stream
Arguments:
id {str} -- the id of the stream to be updated
name {str} -- the name of the string
description {str} -- a short description of the stream
is_public {bool}
-- whether or not the stream can be viewed by anyone with the id
Returns:
bool -- whether the stream update was successful
"""
query = gql(
"""
mutation StreamUpdate($stream: StreamUpdateInput!) {
streamUpdate(stream: $stream)
}
"""
)
params = {
"id": id,
"name": name,
"description": description,
"isPublic": is_public,
}
# remove None values so graphql doesn't cry
params = {"stream": {k: v for k, v in params.items() if v is not None}}
return self.make_request(
query=query, params=params, return_type="streamUpdate", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def delete(self, id: str) -> bool:
"""Delete a stream given its id
Arguments:
id {str} -- the id of the stream to delete
Returns:
bool -- whether the deletion was successful
"""
query = gql(
"""
mutation StreamDelete($id: String!) {
streamDelete(id: $id)
}
"""
)
params = {"id": id}
return self.make_request(
query=query, params=params, return_type="streamDelete", parse_response=False
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def search(
self,
search_query: str,
limit: int = 25,
branch_limit: int = 10,
commit_limit: int = 10,
):
"""Search for streams by name, description, or id
Arguments:
search_query {str} -- a string to search for
limit {int} -- the maximum number of results to return
branch_limit {int} -- the maximum number of branches to return
commit_limit {int} -- the maximum number of commits to return
Returns:
List[Stream] -- a list of Streams that match the search query
"""
query = gql(
"""
query StreamSearch(
$search_query: String!,
$limit: Int!,
$branch_limit:Int!,
$commit_limit:Int!
) {
streams(query: $search_query, limit: $limit) {
items {
id
name
role
description
isPublic
createdAt
updatedAt
collaborators {
id
name
role
avatar
}
branches(limit: $branch_limit) {
totalCount
cursor
items {
id
name
description
commits(limit: $commit_limit) {
totalCount
cursor
items {
id
referencedObject
message
authorName
authorId
createdAt
}
}
}
}
}
}
}
"""
)
params = {
"search_query": search_query,
"limit": limit,
"branch_limit": branch_limit,
"commit_limit": commit_limit,
}
return self.make_request(
query=query, params=params, return_type=["streams", "items"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def favorite(self, stream_id: str, favorited: bool = True):
"""Favorite or unfavorite the given stream.
Arguments:
stream_id {str} -- the id of the stream to favorite / unfavorite
favorited {bool}
-- whether to favorite (True) or unfavorite (False) the stream
Returns:
Stream -- the stream with its `id`, `name`, and `favoritedDate`
"""
query = gql(
"""
mutation StreamFavorite($stream_id: String!, $favorited: Boolean!) {
streamFavorite(streamId: $stream_id, favorited: $favorited) {
id
name
favoritedDate
favoritesCount
}
}
"""
)
params = {
"stream_id": stream_id,
"favorited": favorited,
}
return self.make_request(
query=query, params=params, return_type=["streamFavorite"]
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def get_all_pending_invites(
self, stream_id: str
) -> List[PendingStreamCollaborator]:
"""Get all of the pending invites on a stream.
You must be a `stream:owner` to query this.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the stream id from which to get the pending invites
Returns:
List[PendingStreamCollaborator]
-- a list of pending invites for the specified stream
"""
self._check_invites_supported()
query = gql(
"""
query StreamInvites($streamId: String!) {
stream(id: $streamId){
pendingCollaborators {
id
token
inviteId
streamId
streamName
projectName
projectId
title
role
invitedBy{
id
name
bio
company
avatar
verified
role
}
user {
id
name
bio
company
avatar
verified
role
}
}
}
}
"""
)
params = {"streamId": stream_id}
return self.make_request(
query=query,
params=params,
return_type=["stream", "pendingCollaborators"],
schema=PendingStreamCollaborator,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite(
self,
stream_id: str,
email: Optional[str] = None,
user_id: Optional[str] = None,
role: str = "stream:contributor", # should default be reviewer?
message: Optional[str] = None,
):
"""Invite someone to a stream using either their email or user id
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to invite the user to
email {str} -- the email of the user to invite (use this OR `user_id`)
user_id {str} -- the id of the user to invite (use this OR `email`)
role {str}
-- the role to assign to the user (defaults to `stream:contributor`)
message {str}
-- a message to send along with this invite to the specified user
Returns:
bool -- True if the operation was successful
"""
self._check_invites_supported()
if email is None and user_id is None:
raise SpeckleException(
"You must provide either an email or a user id to use the"
" `stream.invite` method"
)
query = gql(
"""
mutation StreamInviteCreate($input: StreamInviteCreateInput!) {
streamInviteCreate(input: $input)
}
"""
)
params = {
"email": email,
"userId": user_id,
"streamId": stream_id,
"message": message,
"role": role,
}
params = {"input": {k: v for k, v in params.items() if v is not None}}
return self.make_request(
query=query,
params=params,
return_type="streamInviteCreate",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_batch(
self,
stream_id: str,
emails: Optional[List[str]] = None,
user_ids: Optional[List[None]] = None,
message: Optional[str] = None,
) -> bool:
"""Invite a batch of users to a specified stream.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to invite the user to
emails {List[str]}
-- the email of the user to invite (use this and/or `user_ids`)
user_id {List[str]}
-- the id of the user to invite (use this and/or `emails`)
message {str}
-- a message to send along with this invite to the specified user
Returns:
bool -- True if the operation was successful
"""
self._check_invites_supported()
if emails is None and user_ids is None:
raise SpeckleException(
"You must provide either an email or a user id to use the"
" `stream.invite` method"
)
query = gql(
"""
mutation StreamInviteBatchCreate($input: [StreamInviteCreateInput!]!) {
streamInviteBatchCreate(input: $input)
}
"""
)
email_invites = [
{"streamId": stream_id, "message": message, "email": email}
for email in (emails if emails is not None else [])
if email is not None
]
user_invites = [
{"streamId": stream_id, "message": message, "userId": user_id}
for user_id in (user_ids if user_ids is not None else [])
if user_id is not None
]
params = {"input": [*email_invites, *user_invites]}
return self.make_request(
query=query,
params=params,
return_type="streamInviteBatchCreate",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_cancel(self, stream_id: str, invite_id: str) -> bool:
"""Cancel an existing stream invite
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream invite
invite_id {str} -- the id of the invite to use
Returns:
bool -- true if the operation was successful
"""
self._check_invites_supported()
query = gql(
"""
mutation StreamInviteCancel($streamId: String!, $inviteId: String!) {
streamInviteCancel(streamId: $streamId, inviteId: $inviteId)
}
"""
)
params = {"streamId": stream_id, "inviteId": invite_id}
return self.make_request(
query=query,
params=params,
return_type="streamInviteCancel",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def invite_use(self, stream_id: str, token: str, accept: bool = True) -> bool:
"""Accept or decline a stream invite
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str}
-- the id of the stream for which the user has a pending invite
token {str} -- the token of the invite to use
accept {bool} -- whether or not to accept the invite (defaults to True)
Returns:
bool -- true if the operation was successful
"""
self._check_invites_supported()
query = gql(
"""
mutation StreamInviteUse(
$accept: Boolean!,
$streamId: String!,
$token: String!
) {
streamInviteUse(accept: $accept, streamId: $streamId, token: $token)
}
"""
)
params = {"streamId": stream_id, "token": token, "accept": accept}
return self.make_request(
query=query,
params=params,
return_type="streamInviteUse",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def update_permission(self, stream_id: str, user_id: str, role: str):
"""Updates permissions for a user on a given stream
Valid for Speckle Server >=2.6.4
Arguments:
stream_id {str} -- the id of the stream to grant permissions to
user_id {str} -- the id of the user to grant permissions for
role {str} -- the role to grant the user
Returns:
bool -- True if the operation was successful
"""
if self.server_version and (
self.server_version != ("dev",) and self.server_version < (2, 6, 4)
):
raise UnsupportedException(
"Server mutation `update_permission` is only supported as of Speckle"
" Server v2.6.4. Please update your Speckle Server to use this method"
" or use the `grant_permission` method instead."
)
query = gql(
"""
mutation StreamUpdatePermission(
$permission_params: StreamUpdatePermissionInput!
) {
streamUpdatePermission(permissionParams: $permission_params)
}
"""
)
params = {
"permission_params": {
"streamId": stream_id,
"userId": user_id,
"role": role,
}
}
return self.make_request(
query=query,
params=params,
return_type="streamUpdatePermission",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def revoke_permission(self, stream_id: str, user_id: str):
"""Revoke permissions from a user on a given stream
Arguments:
stream_id {str} -- the id of the stream to revoke permissions from
user_id {str} -- the id of the user to revoke permissions from
Returns:
bool -- True if the operation was successful
"""
query = gql(
"""
mutation StreamRevokePermission(
$permission_params: StreamRevokePermissionInput!
) {
streamRevokePermission(permissionParams: $permission_params)
}
"""
)
params = {"permission_params": {"streamId": stream_id, "userId": user_id}}
return self.make_request(
query=query,
params=params,
return_type="streamRevokePermission",
parse_response=False,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
def activity(
self,
stream_id: str,
action_type: Optional[str] = None,
limit: int = 20,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
):
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
Note: all timestamps arguments should be `datetime` of any tz
as they will be converted to UTC ISO format strings
stream_id {str} -- the id of the stream to get activity from
action_type {str}
-- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime}
-- latest cutoff for activity (ie: return all activity _before_ this time)
after {datetime}
-- oldest cutoff for activity (ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
query = gql(
"""
query StreamActivity(
$stream_id: String!,
$action_type: String,
$before:DateTime,
$after: DateTime,
$cursor: DateTime,
$limit: Int
){
stream(id: $stream_id) {
activity(
actionType: $action_type,
before: $before,
after: $after,
cursor: $cursor,
limit: $limit
) {
totalCount
cursor
items {
actionType
info
userId
streamId
resourceId
resourceType
message
time
}
}
}
}
"""
)
try:
params = {
"stream_id": stream_id,
"limit": limit,
"action_type": action_type,
"before": (
before.astimezone(timezone.utc).isoformat() if before else before
),
"after": after.astimezone(timezone.utc).isoformat() if after else after,
"cursor": (
cursor.astimezone(timezone.utc).isoformat() if cursor else cursor
),
}
except AttributeError as e:
raise SpeckleException(
"Could not get stream activity - `before`, `after`, and `cursor` must"
" be in `datetime` format if provided",
ValueError(),
) from e
return self.make_request(
query=query,
params=params,
return_type=["stream", "activity"],
schema=ActivityCollection,
)
src/specklepy/core/api/resources/deprecated/subscriptions.py
-144
View File
@@ -1,144 +0,0 @@
from functools import wraps
from typing import Callable, Dict, List, Optional, Union
from deprecated import deprecated
from gql import gql
from graphql import DocumentNode
from specklepy.core.api.models.deprecated import (
FE1_DEPRECATION_REASON,
FE1_DEPRECATION_VERSION,
Stream,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.logging.exceptions import SpeckleException
NAME = "subscribe"
def check_wsclient(function):
@wraps(function)
async def check_wsclient_wrapper(self, *args, **kwargs):
if self.client is None:
raise SpeckleException(
"You must authenticate before you can subscribe to events"
)
else:
return await function(self, *args, **kwargs)
return check_wsclient_wrapper
class Resource(ResourceBase):
"""API Access class for subscriptions"""
def __init__(self, account, basepath, client) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
)
@deprecated(reason=FE1_DEPRECATION_REASON, version=FE1_DEPRECATION_VERSION)
@check_wsclient
async def stream_added(self, callback: Optional[Callable] = None):
"""Subscribes to new stream added event for your profile.
Use this to display an up-to-date list of streams.
Arguments:
callback {Callable[Stream]} -- a function that takes the updated stream
as an argument and executes each time a stream is added
Returns:
Stream -- the update stream
"""
query = gql(
"""
subscription { userStreamAdded }
"""
)
return await self.subscribe(
query=query, callback=callback, return_type="userStreamAdded", schema=Stream
)
@check_wsclient
async def stream_updated(self, id: str, callback: Optional[Callable] = None):
"""
Subscribes to stream updated event.
Use this in clients/components that pertain only to this stream.
Arguments:
id {str} -- the stream id of the stream to subscribe to
callback {Callable[Stream]}
-- a function that takes the updated stream
as an argument and executes each time the stream is updated
Returns:
Stream -- the update stream
"""
query = gql(
"""
subscription Update($id: String!) { streamUpdated(streamId: $id) }
"""
)
params = {"id": id}
return await self.subscribe(
query=query,
params=params,
callback=callback,
return_type="streamUpdated",
schema=Stream,
)
@check_wsclient
async def stream_removed(self, callback: Optional[Callable] = None):
"""Subscribes to stream removed event for your profile.
Use this to display an up-to-date list of streams for your profile.
NOTE: If someone revokes your permissions on a stream,
this subscription will be triggered with an extra value of revokedBy
in the payload.
Arguments:
callback {Callable[Dict]}
-- a function that takes the returned dict as an argument
and executes each time a stream is removed
Returns:
dict -- dict containing 'id' of stream removed and optionally 'revokedBy'
"""
query = gql(
"""
subscription { userStreamRemoved }
"""
)
return await self.subscribe(
query=query,
callback=callback,
return_type="userStreamRemoved",
parse_response=False,
)
@check_wsclient
async def subscribe(
self,
query: DocumentNode,
params: Optional[Dict] = None,
callback: Optional[Callable] = None,
return_type: Optional[Union[str, List]] = None,
schema=None,
parse_response: bool = True,
):
# if self.client.transport.websocket is None:
# TODO: add multiple subs to the same ws connection
async with self.client as session:
async for res in session.subscribe(query, variable_values=params):
res = self._step_into_response(response=res, return_type=return_type)
if parse_response:
res = self._parse_response(response=res, schema=schema)
if callback is not None:
callback(res)
else:
return res
src/specklepy/core/api/resources/deprecated/user.py
-325
View File
@@ -1,325 +0,0 @@
from datetime import datetime, timezone
from typing import List, Optional, Union
from deprecated import deprecated
from gql import gql
from specklepy.core.api.models import (
ActivityCollection,
PendingStreamCollaborator,
User,
)
from specklepy.core.api.resource import ResourceBase
from specklepy.logging.exceptions import SpeckleException
NAME = "user"
DEPRECATION_VERSION = "2.9.0"
DEPRECATION_TEXT = (
"The user resource is deprecated, please use the active_user or other_user"
" resources"
)
class Resource(ResourceBase):
"""API Access class for users"""
def __init__(self, account, basepath, client, server_version) -> None:
super().__init__(
account=account,
basepath=basepath,
client=client,
name=NAME,
server_version=server_version,
)
self.schema = User
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get(self, id: Optional[str] = None) -> User:
"""
Gets the profile of a user.
If no id argument is provided, will return the current authenticated
user's profile (as extracted from the authorization header).
Arguments:
id {str} -- the user id
Returns:
User -- the retrieved user
"""
query = gql(
"""
query User($id: String) {
user(id: $id) {
id
email
name
bio
company
avatar
verified
profiles
role
}
}
"""
)
params = {"id": id}
return self.make_request(query=query, params=params, return_type="user")
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def search(
self, search_query: str, limit: int = 25
) -> Union[List[User], SpeckleException]:
"""
Searches for user by name or email.
The search query must be at least 3 characters long
Arguments:
search_query {str} -- a string to search for
limit {int} -- the maximum number of results to return
Returns:
List[User] -- a list of User objects that match the search query
"""
if len(search_query) < 3:
return SpeckleException(
message="User search query must be at least 3 characters"
)
query = gql(
"""
query UserSearch($search_query: String!, $limit: Int!) {
userSearch(query: $search_query, limit: $limit) {
items {
id
name
bio
company
avatar
verified
}
}
}
"""
)
params = {"search_query": search_query, "limit": limit}
return self.make_request(
query=query, params=params, return_type=["userSearch", "items"]
)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def update(
self,
name: Optional[str] = None,
company: Optional[str] = None,
bio: Optional[str] = None,
avatar: Optional[str] = None,
):
"""Updates your user profile. All arguments are optional.
Arguments:
name {str} -- your name
company {str} -- the company you may or may not work for
bio {str} -- tell us about yourself
avatar {str} -- a nice photo of yourself
Returns:
bool -- True if your profile was updated successfully
"""
query = gql(
"""
mutation UserUpdate($user: UserUpdateInput!) {
userUpdate(user: $user)
}
"""
)
params = {"name": name, "company": company, "bio": bio, "avatar": avatar}
params = {"user": {k: v for k, v in params.items() if v is not None}}
if not params["user"]:
return SpeckleException(
message=(
"You must provide at least one field to update your user profile"
)
)
return self.make_request(
query=query, params=params, return_type="userUpdate", parse_response=False
)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def activity(
self,
user_id: Optional[str] = None,
limit: int = 20,
action_type: Optional[str] = None,
before: Optional[datetime] = None,
after: Optional[datetime] = None,
cursor: Optional[datetime] = None,
):
"""
Get the activity from a given stream in an Activity collection.
Step into the activity `items` for the list of activity.
If no id argument is provided, will return the current authenticated
user's activity (as extracted from the authorization header).
Note: all timestamps arguments should be `datetime` of any tz as
they will be converted to UTC ISO format strings
user_id {str} -- the id of the user to get the activity from
action_type {str} -- filter results to a single action type
(eg: `commit_create` or `commit_receive`)
limit {int} -- max number of Activity items to return
before {datetime}
-- latest cutoff for activity (ie: return all activity _before_ this time)
after {datetime}
-- oldest cutoff for activity (ie: return all activity _after_ this time)
cursor {datetime} -- timestamp cursor for pagination
"""
query = gql(
"""
query UserActivity(
$user_id: String,
$action_type: String,
$before:DateTime,
$after: DateTime,
$cursor: DateTime,
$limit: Int
){
user(id: $user_id) {
activity(
actionType: $action_type,
before: $before,
after: $after,
cursor: $cursor,
limit: $limit
) {
totalCount
cursor
items {
actionType
info
userId
streamId
resourceId
resourceType
message
time
}
}
}
}
"""
)
params = {
"user_id": user_id,
"limit": limit,
"action_type": action_type,
"before": before.astimezone(timezone.utc).isoformat() if before else before,
"after": after.astimezone(timezone.utc).isoformat() if after else after,
"cursor": cursor.astimezone(timezone.utc).isoformat() if cursor else cursor,
}
return self.make_request(
query=query,
params=params,
return_type=["user", "activity"],
schema=ActivityCollection,
)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get_all_pending_invites(self) -> List[PendingStreamCollaborator]:
"""Get all of the active user's pending stream invites
Requires Speckle Server version >= 2.6.4
Returns:
List[PendingStreamCollaborator]
-- a list of pending invites for the current user
"""
self._check_invites_supported()
query = gql(
"""
query StreamInvites {
streamInvites{
id
token
inviteId
streamId
streamName
title
role
invitedBy {
id
name
company
avatar
}
}
}
"""
)
return self.make_request(
query=query,
return_type="streamInvites",
schema=PendingStreamCollaborator,
)
@deprecated(version=DEPRECATION_VERSION, reason=DEPRECATION_TEXT)
def get_pending_invite(
self, stream_id: str, token: Optional[str] = None
) -> Optional[PendingStreamCollaborator]:
"""Get a particular pending invite for the active user on a given stream.
If no invite_id is provided, any valid invite will be returned.
Requires Speckle Server version >= 2.6.4
Arguments:
stream_id {str} -- the id of the stream to look for invites on
token {str} -- the token of the invite to look for (optional)
Returns:
PendingStreamCollaborator
-- the invite for the given stream (or None if it isn't found)
"""
self._check_invites_supported()
query = gql(
"""
query StreamInvite($streamId: String!, $token: String) {
streamInvite(streamId: $streamId, token: $token) {
id
token
streamId
streamName
title
role
invitedBy {
id
name
company
avatar
}
}
}
"""
)
params = {"streamId": stream_id}
if token:
params["token"] = token
return self.make_request(
query=query,
params=params,
return_type="streamInvite",
schema=PendingStreamCollaborator,
)
src/specklepy/core/api/wrapper.py
+4 -4
View File
@@ -18,7 +18,7 @@ class StreamWrapper:
The `StreamWrapper` gives you some handy helpers to deal with urls and
get authenticated clients and transports.
Construct a `StreamWrapper` with a stream, branch, commit, or object URL.
Construct a `StreamWrapper` with a URL of a model, version, or object.
The corresponding ids will be stored
in the wrapper. If you have local accounts on the machine,
you can use the `get_account` and `get_client` methods
@@ -29,8 +29,8 @@ class StreamWrapper:
```py
from specklepy.api.wrapper import StreamWrapper
# provide any stream, branch, commit, object, or globals url
wrapper = StreamWrapper("https://app.speckle.systems/streams/3073b96e86/commits/604bea8cc6")
# provide a url for a model, version, or object
wrapper = StreamWrapper("https://app.speckle.systems/projects/3073b96e86/models/0fe47c9dca@604bea8cc6")
# get an authenticated SpeckleClient if you have a local account for the server
client = wrapper.get_client()
@@ -163,7 +163,7 @@ class StreamWrapper:
if not self.stream_id:
raise SpeckleException(
f"Cannot parse {url} into a stream wrapper class - no {key_stream} ",
f"Cannot parse {url} into a stream wrapper class - no {key_stream} "
"id found.",
)
src/specklepy/transports/server/server.py
+15 -9
View File
@@ -24,24 +24,31 @@ class ServerTransport(AbstractTransport):
```py
from specklepy.api import operations
from specklepy.api.client import SpeckleClient
from specklepy.core.api.inputs.version_inputs import CreateVersionInput
from specklepy.transports.server import ServerTransport
# here's the data you want to send
block = Block(length=2, height=4)
# next create the server transport - this is the vehicle through which
# here's the project and model you want to send to
project_id = "abcdefghi"
model_id = "ihgfedcba"
# next, create the server transport - this is the vehicle through which
# you will send and receive
transport = ServerTransport(stream_id=new_stream_id, client=client)
transport = ServerTransport(stream_id=project_id, client=client)
# this serialises the block and sends it to the transport
hash = operations.send(base=block, transports=[transport])
# you can now create a commit on your stream with this object
commit_id = client.commit.create(
stream_id=new_stream_id,
obj_id=hash,
message="this is a block I made in speckle-py",
# you can now create tag this version of the model with this object
input = CreateVersionInput(
objectId = hash,
modelId = model_id,
projectId = project_id,
)
version = client.version.create(input)
```
"""
@@ -127,8 +134,7 @@ class ServerTransport(AbstractTransport):
raise SpeckleException(
"Getting a single object using `ServerTransport.get_object()` is not"
" implemented. To get an object from the server, please use the"
" `SpeckleClient.object.get()` route",
" implemented.",
NotImplementedError(),
)
tests/conftest.py
+6 -4
View File
@@ -6,7 +6,7 @@ from specklepy.objects.base import Base
@pytest.fixture(scope="session")
def base():
def base() -> Base:
base = Base()
base.name = "my_base"
base.units = "millimetres"
@@ -15,7 +15,9 @@ def base():
base.tuple = (1, 2, "3")
base.set = {1, 2, "3"}
base.vertices = [random.uniform(0, 10) for _ in range(1, 120)]
base.test_bases = [Base(name=i) for i in range(1, 22)]
base["@detach"] = Base(name="detached base")
base["@revit_thing"] = Base.of_type("SpecialRevitFamily", name="secret tho")
base.test_bases = [Base(applicationId=str(i)) for i in range(1, 22)]
base["@detach"] = Base(applicationId="detached base")
base["@revit_thing"] = Base.of_type(
"SpecialRevitFamily", applicationId="secret tho"
)
return base
tests/integration/client/deprecated/test_active_user.py
-45
View File
@@ -1,45 +0,0 @@
import pytest
from specklepy.api.client import SpeckleClient
from specklepy.api.models import Activity, ActivityCollection, User
from specklepy.logging.exceptions import GraphQLException
@pytest.mark.run(order=2)
class TestUser:
def test_user_get_self(self, client: SpeckleClient, user_dict):
fetched_user = client.active_user.get()
assert isinstance(fetched_user, User)
assert fetched_user.name == user_dict["name"]
assert fetched_user.email == user_dict["email"]
user_dict["id"] = fetched_user.id
def test_user_update(self, client: SpeckleClient):
bio = "i am a ghost in the machine"
with pytest.raises(GraphQLException):
client.active_user.update(bio=None)
updated = client.active_user.update(bio=bio)
assert isinstance(updated, User)
assert isinstance(updated, User)
assert updated.bio == bio
def test_user_activity(self, client: SpeckleClient, second_user_dict):
my_activity = client.active_user.activity(limit=10)
their_activity = client.other_user.activity(second_user_dict["id"])
assert isinstance(my_activity, ActivityCollection)
assert my_activity.items
assert isinstance(my_activity.items[0], Activity)
assert my_activity.totalCount
assert isinstance(their_activity, ActivityCollection)
older_activity = client.active_user.activity(before=my_activity.items[0].time)
assert isinstance(older_activity, ActivityCollection)
assert older_activity.totalCount
assert older_activity.totalCount < my_activity.totalCount
tests/integration/client/deprecated/test_branch.py
-83
View File
@@ -1,83 +0,0 @@
import pytest
from specklepy.api import operations
from specklepy.api.models import Branch, Commit, Stream
from specklepy.transports.server import ServerTransport
class TestBranch:
@pytest.fixture(scope="module")
def branch(self):
return Branch(name="olive branch 🌿", description="a test branch")
@pytest.fixture(scope="module")
def updated_branch(self):
return Branch(name="eucalyptus branch 🌿", description="an updated test branch")
@pytest.fixture(scope="module")
def stream(self, client):
stream = Stream(
name="a sample stream for testing",
description="a stream created for testing",
isPublic=True,
)
stream.id = client.stream.create(
stream.name, stream.description, stream.isPublic
)
return stream
def test_branch_create(self, client, stream, branch):
branch.id = client.branch.create(
stream_id=stream.id, name=branch.name, description=branch.description
)
assert isinstance(branch.id, str)
def test_branch_get(self, client, mesh, stream, branch):
transport = ServerTransport(client=client, stream_id=stream.id)
mesh.id = operations.send(mesh, transports=[transport])
client.commit.create(
stream_id=stream.id,
branch_name=branch.name,
object_id=mesh.id,
message="a commit for testing branch get",
)
fetched_branch = client.branch.get(stream_id=stream.id, name=branch.name)
assert isinstance(fetched_branch, Branch)
assert fetched_branch.name == branch.name
assert fetched_branch.description == branch.description
assert isinstance(fetched_branch.commits.items, list)
assert isinstance(fetched_branch.commits.items[0], Commit)
def test_branch_list(self, client, stream, branch):
branches = client.branch.list(stream_id=stream.id)
print(branches)
assert isinstance(branches, list)
assert len(branches) == 2
assert isinstance(branches[0], Branch)
assert branches[1].name == branch.name
def test_branch_update(self, client, stream, branch, updated_branch):
updated = client.branch.update(
stream_id=stream.id,
branch_id=branch.id,
name=updated_branch.name,
description=updated_branch.description,
)
fetched_branch = client.branch.get(
stream_id=stream.id, name=updated_branch.name
)
assert updated is True
assert fetched_branch.name == updated_branch.name
assert fetched_branch.description == updated_branch.description
def test_branch_delete(self, client, stream, branch):
deleted = client.branch.delete(stream_id=stream.id, branch_id=branch.id)
assert deleted is True
tests/integration/client/deprecated/test_commit.py
-88
View File
@@ -1,88 +0,0 @@
import pytest
from specklepy.api import operations
from specklepy.api.models import Commit, Stream
from specklepy.transports.server.server import ServerTransport
@pytest.mark.run(order=6)
class TestCommit:
@pytest.fixture(scope="module")
def commit(self):
return Commit(message="a fun little test commit")
@pytest.fixture(scope="module")
def updated_commit(
self,
):
return Commit(message="a fun little updated commit")
@pytest.fixture(scope="module")
def stream(self, client):
stream = Stream(
name="a sample stream for testing",
description="a stream created for testing",
isPublic=True,
)
stream.id = client.stream.create(
stream.name, stream.description, stream.isPublic
)
return stream
def test_commit_create(self, client, stream, mesh, commit):
transport = ServerTransport(client=client, stream_id=stream.id)
mesh.id = operations.send(mesh, transports=[transport])
commit.id = client.commit.create(
stream_id=stream.id, object_id=mesh.id, message=commit.message
)
assert isinstance(commit.id, str)
def test_commit_get(self, client, stream, mesh, commit):
fetched_commit = client.commit.get(stream_id=stream.id, commit_id=commit.id)
assert fetched_commit.message == commit.message
assert fetched_commit.referencedObject == mesh.id
def test_commit_list(self, client, stream):
commits = client.commit.list(stream_id=stream.id)
assert isinstance(commits, list)
assert isinstance(commits[0], Commit)
def test_commit_update(self, client, stream, commit, updated_commit):
updated = client.commit.update(
stream_id=stream.id, commit_id=commit.id, message=updated_commit.message
)
fetched_commit = client.commit.get(stream_id=stream.id, commit_id=commit.id)
assert updated is True
assert fetched_commit.message == updated_commit.message
def test_commit_delete(self, client, stream, mesh):
commit_id = client.commit.create(
stream_id=stream.id, object_id=mesh.id, message="a great commit to delete"
)
deleted = client.commit.delete(stream_id=stream.id, commit_id=commit_id)
assert deleted is True
def test_commit_marked_as_received(self, client, stream, mesh) -> None:
commit = Commit(message="this commit should be received")
commit.id = client.commit.create(
stream_id=stream.id,
object_id=mesh.id,
message=commit.message,
)
commit_marked_received = client.commit.received(
stream.id,
commit.id,
source_application="pytest",
message="testing received",
)
assert commit_marked_received is True
tests/integration/client/deprecated/test_other_user.py
-49
View File
@@ -1,49 +0,0 @@
import pytest
from specklepy.api.client import SpeckleClient
from specklepy.api.models import Activity, ActivityCollection, LimitedUser
@pytest.mark.run(order=4)
class TestOtherUser:
def test_user_get_self(self, client):
"""
Test, that a limited user query cannot query the active user.
"""
with pytest.raises(TypeError):
client.other_user.get()
def test_user_search(self, client: SpeckleClient, second_user_dict):
search_results = client.other_user.search(
search_query=second_user_dict["name"][:5]
)
assert isinstance(search_results, list)
assert len(search_results) > 0
result_user = search_results[0]
assert isinstance(result_user, LimitedUser)
assert result_user.name == second_user_dict["name"]
second_user_dict["id"] = result_user.id
assert getattr(result_user, "email", None) is None
def test_user_get(self, client: SpeckleClient, second_user_dict):
fetched_user = client.other_user.get(id=second_user_dict["id"])
assert isinstance(fetched_user, LimitedUser)
assert fetched_user.name == second_user_dict["name"]
# changed in the server, now you cannot get emails of other users
# not checking this, since the first user could or could not be an admin
# on the server, admins can get emails of others, regular users can't
# assert fetched_user.email == None
second_user_dict["id"] = fetched_user.id
def test_user_activity(self, client: SpeckleClient, second_user_dict):
their_activity = client.other_user.activity(second_user_dict["id"])
assert isinstance(their_activity, ActivityCollection)
assert isinstance(their_activity.items, list)
assert isinstance(their_activity.items[0], Activity)
assert their_activity.totalCount
assert their_activity.totalCount > 0
tests/integration/client/deprecated/test_server.py
-52
View File
@@ -1,52 +0,0 @@
import pytest
from specklepy.api.client import SpeckleClient
from specklepy.api.models import ServerInfo
class TestServer:
@pytest.fixture(scope="module")
def token_info(self):
return {
"token": None,
"name": "super secret token",
"scopes": ["streams:read", "streams:write"],
"lifespan": 9001,
}
def test_server_get(self, client: SpeckleClient):
server = client.server.get()
assert isinstance(server, ServerInfo)
assert isinstance(server.frontend2, bool)
def test_server_version(self, client: SpeckleClient):
version = client.server.version()
assert isinstance(version, tuple)
if len(version) == 1:
assert version[0] == "dev"
else:
assert isinstance(version[0], int)
assert len(version) >= 3
def test_server_apps(self, client: SpeckleClient):
apps = client.server.apps()
assert isinstance(apps, list)
assert len(apps) >= 1
assert any(app["name"] == "Speckle Web Manager" for app in apps)
def test_server_create_token(self, client, token_info):
token_info["token"] = client.server.create_token(
name=token_info["name"],
scopes=token_info["scopes"],
lifespan=token_info["lifespan"],
)
assert isinstance(token_info["token"], str)
def test_server_revoke_token(self, client, token_info):
revoked = client.server.revoke_token(token=token_info["token"])
assert revoked is True
tests/integration/client/deprecated/test_stream.py
-224
View File
@@ -1,224 +0,0 @@
import pytest
from specklepy.api.client import SpeckleClient
from specklepy.api.models import (
Activity,
ActivityCollection,
PendingStreamCollaborator,
Stream,
User,
)
from specklepy.logging.exceptions import GraphQLException, SpeckleException
@pytest.mark.run(order=3)
class TestStream:
@pytest.fixture(scope="session")
def stream(self):
return Stream(
name="a wonderful stream",
description="a stream created for testing",
isPublic=True,
)
@pytest.fixture(scope="module")
def updated_stream(
self,
):
return Stream(
name="a wonderful updated stream",
description="an updated stream description for testing",
isPublic=False,
)
@pytest.fixture(scope="module")
def second_user(self, second_client: SpeckleClient):
return second_client.active_user.get()
def test_stream_create(self, client, stream, updated_stream):
stream.id = updated_stream.id = client.stream.create(
name=stream.name,
description=stream.description,
is_public=stream.isPublic,
)
assert isinstance(stream.id, str)
def test_stream_create_short_name(self, client, stream, updated_stream):
new_stream_id = client.stream.create(
name="x",
description=stream.description,
is_public=stream.isPublic,
)
assert isinstance(new_stream_id, SpeckleException)
def test_stream_get(self, client, stream):
fetched_stream = client.stream.get(stream.id)
assert fetched_stream.name == stream.name
assert fetched_stream.description == stream.description
assert fetched_stream.isPublic == stream.isPublic
def test_stream_update(self, client, updated_stream):
updated = client.stream.update(
id=updated_stream.id,
name=updated_stream.name,
description=updated_stream.description,
is_public=updated_stream.isPublic,
)
fetched_stream = client.stream.get(updated_stream.id)
assert updated is True
assert fetched_stream.name == updated_stream.name
assert fetched_stream.description == updated_stream.description
assert fetched_stream.isPublic == updated_stream.isPublic
def test_stream_list(self, client):
client.stream.create(name="a second wonderful stream")
client.stream.create(name="a third fantastic stream")
streams = client.stream.list()
assert len(streams) >= 3
def test_stream_search(self, client, updated_stream):
search_results = client.stream.search(updated_stream.name)
assert len(search_results) == 1
assert search_results[0].name == updated_stream.name
def test_stream_favorite(self, client, stream):
favorited = client.stream.favorite(stream.id)
assert isinstance(favorited, Stream)
assert favorited.favoritedDate is not None
unfavorited = client.stream.favorite(stream.id, False)
assert isinstance(favorited, Stream)
assert unfavorited.favoritedDate is None
def test_stream_invite(
self, client: SpeckleClient, stream: Stream, second_user_dict: dict
):
invited = client.stream.invite(
stream_id=stream.id,
email=second_user_dict["email"],
role="stream:reviewer",
message="welcome to my stream!",
)
assert invited is True
# fail if no email or id
with pytest.raises(SpeckleException):
client.stream.invite(stream_id=stream.id)
def test_stream_invite_get_all_for_user(
self, second_client: SpeckleClient, stream: Stream
):
# NOTE: these are user queries, but testing here to contain the flow
invites = second_client.active_user.get_all_pending_invites()
assert isinstance(invites, list)
assert isinstance(invites[0], PendingStreamCollaborator)
assert len(invites) == 1
invite = second_client.active_user.get_pending_invite(stream_id=stream.id)
assert isinstance(invite, PendingStreamCollaborator)
def test_stream_invite_use(self, second_client: SpeckleClient, stream: Stream):
invite: PendingStreamCollaborator = (
second_client.active_user.get_all_pending_invites()[0]
)
accepted = second_client.stream.invite_use(
stream_id=stream.id, token=invite.token
)
assert accepted is True
def test_stream_update_permission(
self, client: SpeckleClient, stream: Stream, second_user: User
):
updated = client.stream.update_permission(
stream_id=stream.id, user_id=second_user.id, role="stream:contributor"
)
assert updated is True
def test_stream_revoke_permission(self, client, stream, second_user):
revoked = client.stream.revoke_permission(
stream_id=stream.id, user_id=second_user.id
)
fetched_stream = client.stream.get(stream.id)
assert revoked is True
assert len(fetched_stream.collaborators) == 1
def test_stream_invite_cancel(
self,
client: SpeckleClient,
stream: Stream,
second_user: User,
):
invited = client.stream.invite(
stream_id=stream.id,
user_id=second_user.id,
message="welcome to my stream!",
)
assert invited is True
invites = client.stream.get_all_pending_invites(stream_id=stream.id)
cancelled = client.stream.invite_cancel(
invite_id=invites[0].invite_id, stream_id=stream.id
)
assert cancelled is True
def test_stream_invite_batch(
self, client: SpeckleClient, stream: Stream, second_user: User
):
# NOTE: only works for server admins
# invited = client.stream.invite_batch(
# stream_id=stream.id,
# emails=["userA@example.org", "userB@example.org"],
# user_ids=[second_user.id],
# message="yeehaw 🤠",
# )
# assert invited is True
# invited_only_email = client.stream.invite_batch(
# stream_id=stream.id,
# emails=["userC@example.org"],
# message="yeehaw 🤠",
# )
# assert invited_only_email is True
# fail if no emails or user ids
with pytest.raises(SpeckleException):
client.stream.invite_batch(stream_id=stream.id)
def test_stream_activity(self, client: SpeckleClient, stream: Stream):
activity = client.stream.activity(stream.id)
older_activity = client.stream.activity(
stream.id, before=activity.items[0].time
)
assert isinstance(activity, ActivityCollection)
assert isinstance(older_activity, ActivityCollection)
assert older_activity.totalCount < activity.totalCount
assert activity.items is not None
assert isinstance(activity.items[0], Activity)
def test_stream_delete(self, client, stream):
deleted = client.stream.delete(stream.id)
stream_get = client.stream.get(stream.id)
assert deleted is True
assert isinstance(stream_get, GraphQLException)
tests/integration/client/deprecated/test_user.py
-74
View File
@@ -1,74 +0,0 @@
import pytest
from specklepy.api.client import SpeckleClient
from specklepy.api.models import Activity, ActivityCollection, User
from specklepy.logging.exceptions import SpeckleException
@pytest.mark.run(order=1)
class TestUser:
def test_user_get_self(self, client: SpeckleClient, user_dict):
with pytest.deprecated_call():
fetched_user = client.user.get()
assert isinstance(fetched_user, User)
assert fetched_user.name == user_dict["name"]
assert fetched_user.email == user_dict["email"]
user_dict["id"] = fetched_user.id
def test_user_search(self, client, second_user_dict):
with pytest.deprecated_call():
search_results = client.user.search(
search_query=second_user_dict["name"][:5]
)
assert isinstance(search_results, list)
assert isinstance(search_results[0], User)
assert search_results[0].name == second_user_dict["name"]
second_user_dict["id"] = search_results[0].id
def test_user_get(self, client, second_user_dict):
with pytest.deprecated_call():
fetched_user = client.user.get(id=second_user_dict["id"])
assert isinstance(fetched_user, User)
assert fetched_user.name == second_user_dict["name"]
# changed in the server, now you cannot get emails of other users
# not checking this, since the first user could or could not be an admin
# on the server, admins can get emails of others, regular users can't
# assert fetched_user.email == None
second_user_dict["id"] = fetched_user.id
def test_user_update(self, client):
bio = "i am a ghost in the machine"
with pytest.deprecated_call():
failed_update = client.user.update()
assert isinstance(failed_update, SpeckleException)
with pytest.deprecated_call():
updated = client.user.update(bio=bio)
assert updated is True
with pytest.deprecated_call():
me = client.user.get()
assert me.bio == bio
def test_user_activity(self, client: SpeckleClient, second_user_dict):
with pytest.deprecated_call():
my_activity = client.user.activity(limit=10)
their_activity = client.user.activity(second_user_dict["id"])
assert isinstance(my_activity, ActivityCollection)
assert my_activity.items
assert isinstance(my_activity.items[0], Activity)
assert my_activity.totalCount
assert isinstance(their_activity, ActivityCollection)
older_activity = client.user.activity(before=my_activity.items[0].time)
assert isinstance(older_activity, ActivityCollection)
assert older_activity.totalCount
assert older_activity.totalCount < my_activity.totalCount
tests/integration/conftest.py
+13 -11
View File
@@ -8,8 +8,11 @@ import requests
from specklepy.api.client import SpeckleClient
from specklepy.core.api import operations
from specklepy.core.api.enums import ProjectVisibility
from specklepy.core.api.inputs.project_inputs import ProjectCreateInput
from specklepy.core.api.inputs.version_inputs import CreateVersionInput
from specklepy.core.api.models import Stream, Version
from specklepy.core.api.models import Version
from specklepy.core.api.models.current import Project
from specklepy.logging import metrics
from specklepy.objects.base import Base
from specklepy.objects.geometry import Point
@@ -107,14 +110,13 @@ def second_client(host: str, second_user_dict: Dict[str, str]):
@pytest.fixture(scope="session")
def sample_stream(client: SpeckleClient) -> Stream:
stream = Stream(
def sample_project(client: SpeckleClient) -> Project:
input = ProjectCreateInput(
name="a sample stream for testing",
description="a stream created for testing",
isPublic=True,
visibility=ProjectVisibility.PUBLIC,
)
stream.id = client.stream.create(stream.name, stream.description, stream.isPublic)
return stream
return client.project.create(input)
@pytest.fixture(scope="session")
@@ -126,14 +128,14 @@ def mesh() -> FakeMesh:
mesh["@(100)colours"] = [random.uniform(0, 10) for _ in range(1, 210)]
mesh["@()default_chunk"] = [random.uniform(0, 10) for _ in range(1, 210)]
mesh.cardinal_dir = FakeDirection.WEST
mesh.test_bases = [Base(name=f"test {i}") for i in range(1, 22)]
mesh.detach_this = Base(name="predefined detached base")
mesh["@detach"] = Base(name="detached base")
mesh.test_bases = [Base(applicationId=f"test {i}") for i in range(1, 22)]
mesh.detach_this = Base(applicationId="predefined detached base")
mesh["@detach"] = Base(applicationId="detached base")
mesh["@detached_list"] = [
42,
"some text",
[1, 2, 3],
Base(name="detached within a list"),
Base(applicationId="detached within a list"),
]
mesh.origin = Point(x=4, y=2)
mesh.origin = Point(x=4, y=2, z=0, units="m")
return mesh
tests/integration/test_serialization.py
+12 -8
View File
@@ -3,6 +3,8 @@ import json
import pytest
from specklepy.api import operations
from specklepy.api.client import SpeckleClient
from specklepy.core.api.models.current import Project
from specklepy.objects.base import Base
from specklepy.objects.geometry import Point
from specklepy.transports.memory import MemoryTransport
@@ -13,7 +15,7 @@ from .fakemesh import FakeMesh
@pytest.mark.run(order=5)
class TestSerialization:
def test_serialize(self, base):
def test_serialize(self, base: Base):
serialized = operations.serialize(base)
deserialized = operations.deserialize(serialized)
@@ -21,9 +23,9 @@ class TestSerialization:
assert base.units == "millimetres"
assert isinstance(base.test_bases[0], Base)
assert base["@revit_thing"].speckle_type == "SpecialRevitFamily"
assert base["@detach"].name == deserialized["@detach"].name
assert base["@detach"].applicationId == deserialized["@detach"].applicationId
def test_detaching(self, mesh):
def test_detaching(self, mesh: FakeMesh):
transport = MemoryTransport()
serialized = operations.serialize(mesh, [transport])
deserialized = operations.deserialize(serialized, transport)
@@ -36,7 +38,7 @@ class TestSerialization:
assert serialized_dict["@detached_list"][-1]["speckle_type"] == "reference"
assert mesh.get_id() == deserialized.get_id()
def test_chunking(self, mesh):
def test_chunking(self, mesh: FakeMesh):
transport = MemoryTransport()
serialized = operations.serialize(mesh, [transport])
deserialized = operations.deserialize(serialized, transport)
@@ -51,13 +53,15 @@ class TestSerialization:
assert serialized_dict["@()default_chunk"][0]["speckle_type"] == "reference"
assert mesh.get_id() == deserialized.get_id()
def test_send_and_receive(self, client, sample_stream, mesh):
transport = ServerTransport(client=client, stream_id=sample_stream.id)
def test_send_and_receive(
self, client: SpeckleClient, sample_project: Project, mesh: FakeMesh
):
transport = ServerTransport(client=client, stream_id=sample_project.id)
hash = operations.send(mesh, transports=[transport])
# also try constructing server transport with token and url
transport = ServerTransport(
stream_id=sample_stream.id, token=client.account.token, url=client.url
stream_id=sample_project.id, token=client.account.token, url=client.url
)
# use a fresh memory transport to force receiving from remote
received = operations.receive(
@@ -72,7 +76,7 @@ class TestSerialization:
mesh.id = hash # populate with decomposed id for use in proceeding tests
def test_receive_local(self, client, mesh):
def test_receive_local(self, client: SpeckleClient, mesh: FakeMesh):
hash = operations.send(mesh) # defaults to SQLiteTransport
received = operations.receive(hash)
tests/unit/test_base.py
+37 -20
View File
@@ -5,8 +5,9 @@ from typing import Dict, List, Optional, Union
import pytest
from specklepy.api import operations
from specklepy.logging.exceptions import SpeckleException, SpeckleInvalidUnitException
from specklepy.logging.exceptions import SpeckleException
from specklepy.objects.base import Base
from specklepy.objects.interfaces import IHasUnits
from specklepy.objects.models.units import Units
@@ -42,7 +43,8 @@ def test_new_type_registration() -> None:
def test_fake_base_serialization() -> None:
fake_model = FakeModel(foo="bar")
fake_model = FakeModel()
fake_model.foo = "bar"
serialized = operations.serialize(fake_model)
deserialized = operations.deserialize(serialized)
@@ -79,24 +81,32 @@ def test_attribute_name_validation(
def test_speckle_type_cannot_be_set(base: Base) -> None:
assert base.speckle_type == "Base"
base.speckle_type = "unset"
base.speckle_type = "unset" # type: ignore
assert base.speckle_type == "Base"
class FakeUnitBase(Base, IHasUnits, speckle_type="UnityBase"):
pass
def test_setting_units():
b = Base(units="foot")
b = FakeUnitBase()
b.units = "foot"
assert b.units == "foot"
# with pytest.raises(SpeckleInvalidUnitException):
b.units = "big"
assert b.units == "big"
with pytest.raises(SpeckleInvalidUnitException):
b.units = 7 # invalid args are skipped
# invalid args are skipped
with pytest.raises(SpeckleException):
b.units = 7 # type: ignore
assert b.units == "big"
b.units = None # None should be a valid arg
assert b.units is None
# None should be not be a valid arg
with pytest.raises(SpeckleException):
b.units = None # type: ignore
assert b.units == "big"
b.units = Units.none
assert b.units == "none"
@@ -106,9 +116,9 @@ def test_setting_units():
def test_base_of_custom_speckle_type() -> None:
b1 = Base.of_type("BirdHouse", name="Tweety's Crib")
b1 = Base.of_type("BirdHouse", applicationId="Tweety's Crib")
assert b1.speckle_type == "BirdHouse"
assert b1.name == "Tweety's Crib"
assert b1.applicationId == "Tweety's Crib"
class DietaryRestrictions(Enum):
@@ -133,22 +143,22 @@ def test_type_checking() -> None:
order = FrozenYoghurt()
order.servings = 2
order.price = "7" # will get converted
order.price = "7" # type: ignore - it will get converted
order.customer = "izzy"
order.dietary = DietaryRestrictions.VEGAN
order.tag = "preorder"
order.tag = 4411
with pytest.raises(SpeckleException):
order.flavours = "not a list"
order.flavours = "not a list" # type: ignore
with pytest.raises(SpeckleException):
order.servings = "five"
order.servings = "five" # type: ignore
with pytest.raises(SpeckleException):
order.add_ons = ["sprinkles"]
order.add_ons = ["sprinkles"] # type: ignore
with pytest.raises(SpeckleException):
order.dietary = "no nuts plz"
order.dietary = "no nuts plz" # type: ignore
with pytest.raises(SpeckleException):
order.tag = ["tag01", "tag02"]
order.tag = ["tag01", "tag02"] # type: ignore
order.add_ons = {"sprinkles": 0.2, "chocolate": 1.0}
order.flavours = ["strawberry", "lychee", "peach", "pineapple"]
@@ -158,14 +168,21 @@ def test_type_checking() -> None:
def test_cached_deserialization() -> None:
material = Base(color="blue", opacity=0.5)
material = Base()
material.color = "blue"
material.opacity = 0.5
a = Base(name="a")
a = Base()
a.name = "a"
a["@material"] = material
b = Base(name="b")
b = Base()
b.name = "b"
b["@material"] = material
root = Base(a=a, b=b)
root = Base()
root.a = a
root.b = b
serialized = operations.serialize(root)
deserialized = operations.deserialize(serialized)