The operations module provides the core functions for working with Speckle data: sending objects to transports, receiving them back, and serializing/deserializing for custom workflows.
Copy
from specklepy.api import operationsfrom specklepy.objects.geometry import Pointfrom specklepy.transports.server import ServerTransport# Send an objectpoint = Point(x=1, y=2, z=3)transport = ServerTransport(stream_id="abc", client=client)object_id = operations.send(point, [transport])# Receive it backreceived = operations.receive(object_id, remote_transport=transport)
from specklepy.api import operationsfrom specklepy.transports.server import ServerTransportfrom specklepy.transports.memory import MemoryTransportfrom specklepy.objects.geometry import Pointpoint = Point(x=1, y=2, z=3)# Send to both server and memory transportserver_transport = ServerTransport(stream_id="project_id", client=client)memory_transport = MemoryTransport()object_id = operations.send( point, [server_transport, memory_transport])# Object now available in both transports
Copy
from specklepy.api import operationsfrom specklepy.objects.geometry import Pointpoint = Point(x=1, y=2, z=3)# Send to local cache only (no remote transport)object_id = operations.send(point)print(f"Cached locally: {object_id}")
When you send an object, all nested Base objects are automatically sent too. You don’t need to send them separately.
The received object, fully recomposed with all children
Examples:
Receive from Server
Receive with Local Cache
Receive from Cache Only
Copy
from specklepy.api import operationsfrom specklepy.api.client import SpeckleClientfrom specklepy.transports.server import ServerTransport# Setupclient = SpeckleClient(host="https://app.speckle.systems")client.authenticate_with_token(token)# Get the object ID from a versionversion = client.version.get(project_id, version_id)object_id = version.referencedObject# Create transport and receivetransport = ServerTransport(stream_id=project_id, client=client)obj = operations.receive(object_id, remote_transport=transport)print(f"Received: {obj.speckle_type}")
Copy
from specklepy.api import operationsfrom specklepy.transports.server import ServerTransportfrom specklepy.transports.sqlite import SQLiteTransportobject_id = "abc123..."# Receive from server, cache locallyremote = ServerTransport(stream_id="project_id", client=client)local = SQLiteTransport()obj = operations.receive( object_id, remote_transport=remote, local_transport=local)# Second receive is faster (reads from local cache)obj2 = operations.receive( object_id, remote_transport=remote, local_transport=local)
Copy
from specklepy.api import operationsfrom specklepy.transports.sqlite import SQLiteTransport# If object is in cache, no remote transport neededobject_id = "abc123..."try: obj = operations.receive( object_id, local_transport=SQLiteTransport() ) print("Found in cache!")except Exception: print("Not in cache, need remote transport")
receive() automatically reconstructs the entire object tree. All referenced child objects are fetched and recomposed.
Note: The serialize() and deserialize() functions below are included for completeness, but most developers won’t need them directly. The send() and receive() operations handle serialization and deserialization implicitly. Use these functions only when you need custom workflows like saving objects to files or working with JSON representations directly.
from specklepy.api import operationsfrom specklepy.transports.memory import MemoryTransport# Assume we serialized with detachment earliertransport = MemoryTransport()# ... (serialize with transport) ...# Deserialize with same transport to resolve referencesobj = operations.deserialize(json_str, read_transport=transport)# All references are resolvedprint(obj.displayValue.vertices_count)
Copy
from specklepy.api import operations# Load from filewith open("point.json", "r") as f: json_str = f.read()point = operations.deserialize(json_str)print(f"Loaded: {point.speckle_type}")
If the JSON contains references (like "@displayValue": "hash123..."), you must provide a read_transport that contains those referenced objects.
Large objects are automatically optimized during send:
Copy
from specklepy.api import operationsfrom specklepy.objects.geometry import Meshfrom specklepy.transports.server import ServerTransport# Create a large meshmesh = Mesh( vertices=[...], # 1 million vertices faces=[...], # 500k faces)# Mesh vertices and faces are automatically chunkedtransport = ServerTransport(stream_id="project_id", client=client)object_id = operations.send(mesh, [transport])# The mesh and its chunks are all stored separately for efficiency
Objects like Mesh have predefined chunking and detachment rules. See the Mesh documentation for details.
# Good - uses local cache by defaultobject_id = operations.send(obj, [remote_transport])# Also good - explicit local cachereceived = operations.receive( object_id, remote_transport=remote_transport, local_transport=SQLiteTransport() # Explicit)# Bad - disabling cache hurts performanceobject_id = operations.send( obj, [remote_transport], use_default_cache=False # Don't do this)
# Only deserialize JSON from trusted sourcesdef safe_deserialize(json_str: str): try: # Validate it's actually Speckle data obj = operations.deserialize(json_str) if not hasattr(obj, 'speckle_type'): raise ValueError("Not a valid Speckle object") return obj except Exception as e: print(f"Deserialization failed: {e}") return None
Use appropriate transports
Choose the right transport for your use case:
Copy
from specklepy.transports.server import ServerTransportfrom specklepy.transports.memory import MemoryTransportfrom specklepy.transports.sqlite import SQLiteTransport# For sending to Speckleserver = ServerTransport(stream_id="id", client=client)operations.send(obj, [server])# For temporary storagememory = MemoryTransport()operations.send(obj, [memory])# For local persistencesqlite = SQLiteTransport()operations.send(obj, [sqlite])