Using dict union operator

alpha-unito · Oct 19, 2024 · e142780 · e142780
1 parent 68dd844
commit e142780
Show file tree

Hide file tree

Showing 25 changed files with 403 additions and 511 deletions.
diff --git a/.github/workflows/ci-tests.yaml b/.github/workflows/ci-tests.yaml
@@ -147,7 +147,7 @@ jobs:
           python -m pip install -r docs/requirements.txt
       - name: "Build documentation and check for consistency"
         env:
-          CHECKSUM: "66ec8ce6a6a09bb8bc95a2bae09405c075ddbd2343a7f8c93bb961b337d4bde4"
+          CHECKSUM: "bedaf099ee8247691295d0be167c943499658b140cf4b188129f3f279ee5ea1b"
         run: |
           cd docs
           HASH="$(make checksum | tail -n1)"

diff --git a/docs/source/conf.py b/docs/source/conf.py
@@ -204,7 +204,7 @@ def patched_run(self, schema, pointer=''):
             if prop in schema['properties']:
                 props[prop] = schema['properties'][prop]
                 del schema['properties'][prop]
-        schema['properties'] = {**props, **schema['properties']}
+        schema['properties'] = props | schema['properties']
     return original_run(self, schema, pointer)
 
 
@@ -226,7 +226,7 @@ def patched_get_json_data(self):
                 target_file = os.path.join(os.path.dirname(filename), obj['$ref'])
                 target_schema, _ = self.from_file(target_file)
                 target_schema = self.ordered_load(target_schema)
-                schema['properties'] = {**target_schema.get('properties', {}), **schema['properties']}
+                schema['properties'] = target_schema.get('properties', {}) | schema['properties']
         del schema['allOf']
         schema['properties'] = dict(sorted(schema['properties'].items()))
     return schema, source, pointer

diff --git a/docs/source/ext/database.rst b/docs/source/ext/database.rst
@@ -209,32 +209,32 @@ The ``Database`` interface, defined in the ``streamflow.core.persistence`` modul
         ...
 
     async def update_deployment(
-        self, deployment_id: int, updates: MutableMapping[str, Any]
+        self, deployment_id: int, updates: dict[str, Any]
     ) -> int:
         ...
 
     async def update_execution(
-        self, execution_id: int, updates: MutableMapping[str, Any]
+        self, execution_id: int, updates: dict[str, Any]
     ) -> int:
         ...
 
     async def update_port(
-        self, port_id: int, updates: MutableMapping[str, Any]
+        self, port_id: int, updates: dict[str, Any]
     ) -> int:
         ...
 
     async def update_step(
-        self, step_id: int, updates: MutableMapping[str, Any]
+        self, step_id: int, updates: dict[str, Any]
     ) -> int:
         ...
 
     async def update_target(
-        self, target_id: str, updates: MutableMapping[str, Any]
+        self, target_id: str, updates: dict[str, Any]
     ) -> int:
         ...
 
     async def update_workflow(
-        self, workflow_id: int, updates: MutableMapping[str, Any]
+        self, workflow_id: int, updates: dict[str, Any]
     ) -> int:
         ...
 

diff --git a/streamflow/core/command.py b/streamflow/core/command.py
@@ -53,7 +53,7 @@ async def _load(
 
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]:
+    ) -> dict[str, Any]:
         return {}
 
 
@@ -109,7 +109,9 @@ async def _load(
             ),
         )
 
-    async def _save_additional_params(self, context: StreamFlowContext):
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> dict[str, Any]:
         if self.target:
             await self.target.save(context)
         return {
@@ -169,7 +171,7 @@ async def _load(
 
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]:
+    ) -> dict[str, Any]:
         return {"name": self.name}
 
     @abstractmethod
@@ -232,9 +234,8 @@ async def _load(
     async def _save_additional_params(
         self, context: StreamFlowContext
     ) -> MutableMapping[str, Any]:
-        return {
-            **await super()._save_additional_params(context),
-            **{"processor": await self.processor.save(context)},
+        return await super()._save_additional_params(context) | {
+            "processor": await self.processor.save(context)
         }
 
     def _update_options(self, options: CommandOptions, token: Token) -> CommandOptions:
@@ -319,22 +320,19 @@ async def _load(
     async def _save_additional_params(
         self, context: StreamFlowContext
     ) -> MutableMapping[str, Any]:
-        return {
-            **await super()._save_additional_params(context),
-            **{
-                "processors": {
-                    name: token
-                    for name, token in zip(
-                        self.processors.keys(),
-                        await asyncio.gather(
-                            *(
-                                asyncio.create_task(t.save(context))
-                                for t in self.processors.values()
-                            )
-                        ),
-                    )
-                }
-            },
+        return await super()._save_additional_params(context) | {
+            "processors": {
+                name: token
+                for name, token in zip(
+                    self.processors.keys(),
+                    await asyncio.gather(
+                        *(
+                            asyncio.create_task(t.save(context))
+                            for t in self.processors.values()
+                        )
+                    ),
+                )
+            }
         }
 
     def _update_options(self, options: CommandOptions, token: Token) -> CommandOptions:
@@ -409,13 +407,10 @@ async def _load(
     async def _save_additional_params(
         self, context: StreamFlowContext
     ) -> MutableMapping[str, Any]:
-        return {
-            **await super()._save_additional_params(context),
-            **{
-                "processors": await asyncio.gather(
-                    *(asyncio.create_task(t.save(context)) for t in self.processors)
-                )
-            },
+        return await super()._save_additional_params(context) | {
+            "processors": await asyncio.gather(
+                *(asyncio.create_task(t.save(context)) for t in self.processors)
+            )
         }
 
     def _update_options(self, options: CommandOptions, token: Token) -> CommandOptions:
@@ -483,14 +478,11 @@ async def _load(
     async def _save_additional_params(
         self, context: StreamFlowContext
     ) -> MutableMapping[str, Any]:
-        return {
-            **await super()._save_additional_params(context),
-            **{
-                "processors": await asyncio.gather(
-                    *(
-                        asyncio.create_task(processor.save(context))
-                        for processor in self.processors
-                    )
+        return await super()._save_additional_params(context) | {
+            "processors": await asyncio.gather(
+                *(
+                    asyncio.create_task(processor.save(context))
+                    for processor in self.processors
                 )
-            },
+            )
         }
diff --git a/streamflow/core/persistence.py b/streamflow/core/persistence.py
@@ -274,35 +274,27 @@ async def get_workflows_list(
 
     @abstractmethod
     async def update_deployment(
-        self, deployment_id: int, updates: MutableMapping[str, Any]
+        self, deployment_id: int, updates: dict[str, Any]
     ) -> int: ...
 
     @abstractmethod
     async def update_execution(
-        self, execution_id: int, updates: MutableMapping[str, Any]
+        self, execution_id: int, updates: dict[str, Any]
     ) -> int: ...
 
     @abstractmethod
-    async def update_filter(
-        self, filter_id: int, updates: MutableMapping[str, Any]
-    ) -> int: ...
+    async def update_filter(self, filter_id: int, updates: dict[str, Any]) -> int: ...
 
     @abstractmethod
-    async def update_port(
-        self, port_id: int, updates: MutableMapping[str, Any]
-    ) -> int: ...
+    async def update_port(self, port_id: int, updates: dict[str, Any]) -> int: ...
 
     @abstractmethod
-    async def update_step(
-        self, step_id: int, updates: MutableMapping[str, Any]
-    ) -> int: ...
+    async def update_step(self, step_id: int, updates: dict[str, Any]) -> int: ...
 
     @abstractmethod
-    async def update_target(
-        self, target_id: str, updates: MutableMapping[str, Any]
-    ) -> int: ...
+    async def update_target(self, target_id: str, updates: dict[str, Any]) -> int: ...
 
     @abstractmethod
     async def update_workflow(
-        self, workflow_id: int, updates: MutableMapping[str, Any]
+        self, workflow_id: int, updates: dict[str, Any]
     ) -> int: ...
diff --git a/streamflow/core/scheduling.py b/streamflow/core/scheduling.py
@@ -172,7 +172,7 @@ async def _load(
     @abstractmethod
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]: ...
+    ) -> dict[str, Any]: ...
 
     @abstractmethod
     def eval(self, job: Job) -> Hardware: ...

diff --git a/streamflow/core/workflow.py b/streamflow/core/workflow.py
@@ -83,7 +83,9 @@ async def _load(
             tmp_directory=row["tmp_directory"],
         )
 
-    async def _save_additional_params(self, context: StreamFlowContext):
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> dict[str, Any]:
         await asyncio.gather(
             *(asyncio.create_task(t.save(context)) for t in self.inputs.values())
         )
@@ -140,7 +142,7 @@ async def _load(
 
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]:
+    ) -> dict[str, Any]:
         return {}
 
     def close(self, consumer: str):
@@ -244,7 +246,7 @@ async def _load(
 
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]:
+    ) -> dict[str, Any]:
         return {}
 
     async def _set_status(self, status: Status):
@@ -461,7 +463,9 @@ async def _load(
             workflow=await loading_context.load_workflow(context, row["workflow"]),
         )
 
-    async def _save_additional_params(self, context: StreamFlowContext):
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> dict[str, Any]:
         return {"name": self.name, "workflow": self.workflow.persistent_id}
 
     @classmethod
@@ -519,7 +523,7 @@ async def _load(
 
     async def _save_additional_params(
         self, context: StreamFlowContext
-    ) -> MutableMapping[str, Any]:
+    ) -> dict[str, Any]:
         return {"config": self.config, "output_ports": self.output_ports}
 
     def create_port(self, cls: type[P] = Port, name: str = None, **kwargs) -> P:

diff --git a/streamflow/cwl/combinator.py b/streamflow/cwl/combinator.py
@@ -54,14 +54,13 @@ async def _load(
             flatten=row["flatten"],
         )
 
-    async def _save_additional_params(self, context: StreamFlowContext):
-        return {
-            **await super()._save_additional_params(context),
-            **{
-                "input_names": self.input_names,
-                "output_name": self.output_name,
-                "flatten": self.flatten,
-            },
+    async def _save_additional_params(
+        self, context: StreamFlowContext
+    ) -> dict[str, Any]:
+        return await super()._save_additional_params(context) | {
+            "input_names": self.input_names,
+            "output_name": self.output_name,
+            "flatten": self.flatten,
         }
 
     async def combine(