Parallelize typed object uploads and harden mutations

packages/evo-objects/README.md

@@ -105,6 +105,46 @@ Attribute descriptions for ordinary attributes round-trip through `DataFrame.att
 Collection coordinate units come only from `DistanceCollection.unit` or `IntervalCollection.unit`; unit metadata on the
 distance or `from` column is ignored.
 
+### Upload concurrency
+
+Typed objects upload independent tables and attributes concurrently, preserving input order and applying mutations only
+after all required uploads succeed. Configure the upload limit for a block, including its child tasks:
+
+```python
+from evo.objects.utils import upload_concurrency
+
+with upload_concurrency(max_concurrent=8):
+    dhc = await DownholeCollection.create(context, data)
+```
+
+The default limit is **4** and is shared by uploads on the same event loop, including direct `ObjectDataClient` calls.
+Unlike uploads, `download_tables(max_concurrency=...)` applies a separate limit to each call. The upload limit covers
+preparation and transfers; a storage retry may hold a slot. Nested scopes restore the previous setting.
+HTTP connection limits are configured separately with `AioTransport(num_pools=...)`.
+
+Uploads retry transient failures up to three times by default; pass `retry=Retry(...)` to an `ObjectDataClient` upload
+method to customize that policy. On exhaustion, the last failure is raised and the retry history is chained as its
+cause. Storage transfers make one attempt per outer retry, preventing retry layers from multiplying requests.
+For custom interception, override `upload_dataframe` or `upload_category_dataframe` for DataFrame inputs; these methods
+do not dispatch through the table upload methods.
+
+To configure a separate HTTP connection limit, pass a configured transport to your context using your existing
+authorizer, cache, and environment:
+
+```python
+from evo.aio import AioTransport
+from evo.common import StaticContext
+
+transport = AioTransport(user_agent="my-app", num_pools=8)
+context = StaticContext(
+    transport=transport, authorizer=authorizer, cache=cache,
+    hub_url=environment.hub_url, org_id=environment.org_id, workspace_id=environment.workspace_id,
+)
+```
+
+`ServiceManager` and `ServiceManagerWidget` also accept a transport in their constructors.
+The `ServiceManagerWidget.with_auth_code(...)` convenience factory currently uses the default connection limit.
+
 ## Contributing
 
 For instructions on contributing to the development of this library, please refer to the [evo-python-sdk documentation](https://github.com/seequentevo/evo-python-sdk).

packages/evo-objects/pyproject.toml

@@ -10,7 +10,7 @@ authors = [
 ]
 
 dependencies = [
-    "evo-sdk-common[jmespath]>=0.5.31",
+    "evo-sdk-common[jmespath]>=0.5.32",
     "pydantic>=2,<3",
 ]
 

packages/evo-objects/src/evo/objects/typed/_data.py

@@ -17,8 +17,9 @@
 
 from evo.common import IFeedback
 from evo.common.interfaces import IContext
-from evo.common.utils import NoFeedback
+from evo.common.utils import NoFeedback, split_feedback
 from evo.objects.typed.attributes import Attributes
+from evo.objects.utils._concurrency import gather_cancelling
 from evo.objects.utils.table_formats import KnownTableFormat
 
 from ._model import SchemaBuilder, SchemaLocation, SchemaModel
@@ -49,9 +50,15 @@ async def from_dataframe(self, df: pd.DataFrame, fb: IFeedback = NoFeedback) ->
         :param df: DataFrame containing the new values for this table.
         :param fb: Optional feedback object to report upload progress.
         """
-        self._document.update(await self._data_to_schema(df, self._context, fb=fb))
+        self._apply_uploaded_table(df, await self._upload(df, fb))
 
-        # Mark the context as modified so loading data is not allowed
+    async def _upload(self, df: pd.DataFrame, fb: IFeedback = NoFeedback) -> dict[str, Any]:
+        """Upload replacement values without changing the table or its derived metadata."""
+        return await self._data_to_schema(df, self._obj, fb=fb)
+
+    def _apply_uploaded_table(self, df: pd.DataFrame, document: dict[str, Any]) -> None:
+        """Commit uploaded values and derived metadata; override to extend table replacement behavior."""
+        self._document.update(document)
         self._context.mark_modified(self._data)
 
     @classmethod
@@ -118,11 +125,17 @@ async def from_dataframe(self, df: pd.DataFrame, fb: IFeedback = NoFeedback) ->
         """
         table_df, attr_df = self._split_dataframe(df, self._table.data_columns)
 
-        await self._table.from_dataframe(table_df, fb=fb)
-        if attr_df is not None:
-            await self.attributes.set_attributes(attr_df, fb=fb)
-        else:
-            self.attributes.clear()
+        if attr_df is None:
+            attr_df = pd.DataFrame(index=table_df.index)
+        table_fb, attr_fb = split_feedback(fb, [1.0, float(len(attr_df.columns))])
+        descriptions, attribute_types = await self.attributes._validate(attr_df)
+        table_upload, attribute_upload = (
+            self._table._upload(table_df, table_fb),
+            self.attributes._upload(attr_df, descriptions, attribute_types, attr_fb),
+        )
+        table_document, attribute_documents = await gather_cancelling(table_upload, attribute_upload)
+        self._table._apply_uploaded_table(table_df, table_document)
+        self.attributes._apply(attribute_documents)
 
     def validate(self) -> None:
         """Validate that all attributes have the correct length."""
@@ -152,8 +165,7 @@ async def _data_to_schema(cls, data: Any, context: IContext) -> Any:
         table_df, attr_df = cls._split_dataframe(data, table_type.data_columns)
 
         builder = SchemaBuilder(cls, context)
-        await builder.set_sub_model_value("_table", table_df)
-        await builder.set_sub_model_value("attributes", attr_df)
+        await builder.set_sub_model_values([("_table", table_df), ("attributes", attr_df)])
 
         # As we specially handle the _table and attributes sub-models above, we skip them in the default data processing of build_from_data.
         return await builder.build_from_data(data, skip_sub_models={"_table", "attributes"})

packages/evo-objects/src/evo/objects/typed/_model.py

@@ -21,6 +21,7 @@
 from evo import jmespath
 from evo.common import IContext
 from evo.objects import DownloadedObject
+from evo.objects.utils._concurrency import gather_cancelling
 
 from ._utils import (
     assign_jmespath_value,
@@ -185,14 +186,6 @@ def set_property(self, name: str, value: Any) -> None:
         schema_property = self._properties[name]
         _set_property_value(schema_property, self.document, value)
 
-    async def set_sub_model_value(self, name: str, data: Any) -> None:
-        metadata = self._sub_models[name]
-        sub_document = await metadata.model_type._data_to_schema(data, context=self._context)
-        if metadata.jmespath_expr:
-            assign_jmespath_value(self.document, metadata.jmespath_expr, sub_document)
-        else:
-            self.document.update(sub_document)
-
     async def build_from_data(self, data: Any, skip_sub_models: set[str] | None = None) -> dict[str, Any]:
         """Build the schema document by applying all properties and sub-models."""
         for key, prop in self._properties.items():
@@ -202,17 +195,37 @@ async def build_from_data(self, data: Any, skip_sub_models: set[str] | None = No
                 value = getattr(data, key, None)
                 self.set_property(key, value)
 
+        pending: list[tuple[str, Any]] = []
         for name, metadata in self._sub_models.items():
             if skip_sub_models and name in skip_sub_models:
                 continue
             if metadata.data_field:
                 sub_data = getattr(data, metadata.data_field, None)
             else:
                 sub_data = data
-            await self.set_sub_model_value(name, sub_data)
+            pending.append((name, sub_data))
+        await self.set_sub_model_values(pending)
 
         return self.document
 
+    async def set_sub_model_values(self, values: Sequence[tuple[str, Any]]) -> None:
+        """Convert several sub-models concurrently, then assign the results in the given order.
+
+        Sub-models are independent, so their data uploads can run in parallel.
+        """
+        sub_documents = await gather_cancelling(
+            *(self._sub_models[name].model_type._data_to_schema(data, context=self._context) for name, data in values)
+        )
+        for (name, _), sub_document in zip(values, sub_documents):
+            self._assign_sub_document(name, sub_document)
+
+    def _assign_sub_document(self, name: str, sub_document: Any) -> None:
+        metadata = self._sub_models[name]
+        if metadata.jmespath_expr:
+            assign_jmespath_value(self.document, metadata.jmespath_expr, sub_document)
+        else:
+            self.document.update(sub_document)
+
 
 def _get_base_type(annotation: Any) -> tuple[Any, SchemaLocation | None, DataLocation | None]:
     """Extract the base type and SchemaLocation from an annotation.
@@ -497,4 +510,4 @@ async def _data_to_schema(cls, data: Any, context: IContext) -> list[Any]:
             return []
         if not isinstance(data, Sequence):
             raise TypeError(f"Expected a sequence for SchemaList data, got {type(data).__name__}")
-        return [await cls._item_type._data_to_schema(item, context) for item in data]
+        return await gather_cancelling(*(cls._item_type._data_to_schema(item, context) for item in data))