Linting

Author: tconley1428

scripts/gen_protos.py

@@ -201,6 +201,7 @@ def generate_protos(output_dir: Path):
             / v,
         )
 
+
 if __name__ == "__main__":
     check_proto_toolchain_versions()
     print("Generating protos...", file=sys.stderr)

scripts/gen_visitors.py

@@ -13,6 +13,7 @@
 
 base_dir = Path(__file__).parent.parent
 
+
 def gen_workflow_activation_payload_visitor_code() -> str:
     """
     Generate Python source code that, given a function f(Payload) -> Payload,
@@ -23,29 +24,44 @@ def gen_workflow_activation_payload_visitor_code() -> str:
     for repeated fields and map entries, and a convenience entrypoint
     function `visit_workflow_activation_payloads`.
     """
+
     def name_for(desc: Descriptor) -> str:
         # Use fully-qualified name to avoid collisions; replace dots with underscores
-        return desc.full_name.replace('.', '_')
-
-    def emit_loop(lines: list[str], field_name: str, iter_expr: str, var_name: str, child_method: str) -> None:
+        return desc.full_name.replace(".", "_")
+
+    def emit_loop(
+        lines: list[str],
+        field_name: str,
+        iter_expr: str,
+        var_name: str,
+        child_method: str,
+    ) -> None:
         # Helper to emit a for-loop over a collection with optional headers guard
         if field_name == "headers":
             lines.append("        if not self.skip_headers:")
             lines.append(f"            for {var_name} in {iter_expr}:")
-            lines.append(f"                await self.visit_{child_method}(f, {var_name})")
+            lines.append(
+                f"                await self.visit_{child_method}(f, {var_name})"
+            )
         else:
             lines.append(f"        for {var_name} in {iter_expr}:")
             lines.append(f"            await self.visit_{child_method}(f, {var_name})")
 
-    def emit_singular(lines: list[str], field_name: str, access_expr: str, child_method: str) -> None:
+    def emit_singular(
+        lines: list[str], field_name: str, access_expr: str, child_method: str
+    ) -> None:
         # Helper to emit a singular field visit with presence check and optional headers guard
         if field_name == "headers":
             lines.append("        if not self.skip_headers:")
             lines.append(f"            if o.HasField('{field_name}'):")
-            lines.append(f"                await self.visit_{child_method}(f, {access_expr})")
+            lines.append(
+                f"                await self.visit_{child_method}(f, {access_expr})"
+            )
         else:
             lines.append(f"        if o.HasField('{field_name}'):")
-            lines.append(f"            await self.visit_{child_method}(f, {access_expr})")
+            lines.append(
+                f"            await self.visit_{child_method}(f, {access_expr})"
+            )
 
     # Track which message descriptors have visitor methods generated
     generated: dict[str, bool] = {}
@@ -83,36 +99,65 @@ def walk(desc: Descriptor) -> bool:
 
             # Repeated fields (including maps which are represented as repeated messages)
             if field.label == FieldDescriptor.LABEL_REPEATED:
-                if field.message_type is not None and field.message_type.GetOptions().map_entry:
+                if (
+                    field.message_type is not None
+                    and field.message_type.GetOptions().map_entry
+                ):
                     entry_desc = field.message_type
                     key_fd = entry_desc.fields_by_name.get("key")
                     val_fd = entry_desc.fields_by_name.get("value")
 
-                    if val_fd is not None and val_fd.type == FieldDescriptor.TYPE_MESSAGE:
+                    if (
+                        val_fd is not None
+                        and val_fd.type == FieldDescriptor.TYPE_MESSAGE
+                    ):
                         child_desc = val_fd.message_type
                         child_needed = walk(child_desc)
                         needed |= child_needed
                         if child_needed:
-                            emit_loop(lines, field.name, f"o.{field.name}.values()", "v", name_for(child_desc))
-
-                    if key_fd is not None and key_fd.type == FieldDescriptor.TYPE_MESSAGE:
+                            emit_loop(
+                                lines,
+                                field.name,
+                                f"o.{field.name}.values()",
+                                "v",
+                                name_for(child_desc),
+                            )
+
+                    if (
+                        key_fd is not None
+                        and key_fd.type == FieldDescriptor.TYPE_MESSAGE
+                    ):
                         key_desc = key_fd.message_type
                         child_needed = walk(key_desc)
                         needed |= child_needed
                         if child_needed:
-                            emit_loop(lines, field.name, f"o.{field.name}.keys()", "k", name_for(key_desc))
+                            emit_loop(
+                                lines,
+                                field.name,
+                                f"o.{field.name}.keys()",
+                                "k",
+                                name_for(key_desc),
+                            )
                 else:
                     child_desc = field.message_type
                     child_needed = walk(child_desc)
                     needed |= child_needed
                     if child_needed:
-                        emit_loop(lines, field.name, f"o.{field.name}", "v", name_for(child_desc))
+                        emit_loop(
+                            lines,
+                            field.name,
+                            f"o.{field.name}",
+                            "v",
+                            name_for(child_desc),
+                        )
             else:
                 child_desc = field.message_type
                 child_needed = walk(child_desc)
                 needed |= child_needed
                 if child_needed:
-                    emit_singular(lines, field.name, f"o.{field.name}", name_for(child_desc))
+                    emit_singular(
+                        lines, field.name, f"o.{field.name}", name_for(child_desc)
+                    )
 
         generated[key] = needed
         in_progress.discard(key)
@@ -134,7 +179,7 @@ def walk(desc: Descriptor) -> bool:
         walk(r)
 
     header = (
-        "from typing import Awaitable, Callable, Any\n\n"
+        "from typing import Any, Awaitable, Callable\n\n"
         "from temporalio.api.common.v1.message_pb2 import Payload\n\n\n"
         "class PayloadVisitor:\n"
         "    def __init__(self, *, skip_search_attributes: bool = False, skip_headers: bool = False):\n"
@@ -156,7 +201,7 @@ def write_generated_visitors_into_visitor_generated_py() -> None:
     code = gen_workflow_activation_payload_visitor_code()
     out_path.write_text(code)
 
+
 if __name__ == "__main__":
     print("Generating temporalio/bridge/visitor_generated.py...", file=sys.stderr)
     write_generated_visitors_into_visitor_generated_py()
-

temporalio/bridge/visitor.py

@@ -9,7 +9,9 @@
 
 
 class PayloadVisitor:
-    def __init__(self, *, skip_search_attributes: bool = False, skip_headers: bool = False):
+    def __init__(
+        self, *, skip_search_attributes: bool = False, skip_headers: bool = False
+    ):
         self.skip_search_attributes = skip_search_attributes
         self.skip_headers = skip_headers
 
@@ -31,8 +33,10 @@ async def visit_payloads(
             for o in root:
                 await self.visit_payloads(f, o)
         elif isinstance(root, Message):
-            await self.visit_message(f, root,)
-
+            await self.visit_message(
+                f,
+                root,
+            )
 
     async def visit_message(
         self, f: Callable[[Payload], Awaitable[Payload]], root: Message
@@ -44,7 +48,10 @@ async def visit_message(
             # Repeated fields (including maps which are represented as repeated messages)
             if field.label == FieldDescriptor.LABEL_REPEATED:
                 value = getattr(root, field.name)
-                if field.message_type is not None and field.message_type.GetOptions().map_entry:
+                if (
+                    field.message_type is not None
+                    and field.message_type.GetOptions().map_entry
+                ):
                     for k, v in value.items():
                         await self.visit_payloads(f, k)
                         await self.visit_payloads(f, v)
@@ -53,6 +60,8 @@ async def visit_message(
                         await self.visit_payloads(f, item)
             else:
                 # Only descend into singular message fields if present
-                if field.type == FieldDescriptor.TYPE_MESSAGE and root.HasField(field.name):
+                if field.type == FieldDescriptor.TYPE_MESSAGE and root.HasField(
+                    field.name
+                ):
                     value = getattr(root, field.name)
                     await self.visit_payloads(f, value)

temporalio/bridge/visitor_generated.py

@@ -1,4 +1,4 @@
-from typing import Awaitable, Callable, Any
+from typing import Any, Awaitable, Callable
 
 from temporalio.api.common.v1.message_pb2 import Payload
 

temporalio/bridge/worker.py

@@ -372,18 +372,25 @@ async def decode_activation(
     decode_headers: bool,
 ) -> None:
     """Decode the given activation with the codec."""
+
     async def visitor(payload: Payload) -> Payload:
         return (await codec.decode([payload]))[0]
 
-    await PayloadVisitor(skip_search_attributes=True, skip_headers=not decode_headers).visit_message(visitor, act)
+    await PayloadVisitor(
+        skip_search_attributes=True, skip_headers=not decode_headers
+    ).visit_message(visitor, act)
+
 
 async def encode_completion(
     comp: temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion,
     codec: temporalio.converter.PayloadCodec,
     encode_headers: bool,
 ) -> None:
     """Recursively encode the given completion with the codec."""
+
     async def visitor(payload: Payload) -> Payload:
         return (await codec.encode([payload]))[0]
 
-    await PayloadVisitor(skip_search_attributes=True, skip_headers=not encode_headers).visit_message(visitor, comp)
+    await PayloadVisitor(
+        skip_search_attributes=True, skip_headers=not encode_headers
+    ).visit_message(visitor, comp)

temporalio/worker/_workflow_instance.py

@@ -146,6 +146,7 @@ class WorkflowInstanceDetails:
     worker_level_failure_exception_types: Sequence[Type[BaseException]]
     last_completion_result: Payloads
 
+
 class WorkflowInstance(ABC):
     """Instance of a workflow that can handle activations."""
 
@@ -1689,16 +1690,28 @@ def workflow_set_current_details(self, details: str):
         self._assert_not_read_only("set current details")
         self._current_details = details
 
-    def workflow_last_completion_result(self, type_hint: Optional[Type]) -> Optional[Any]:
-        print("workflow_last_completion_result: ", self._last_completion_result, type(self._last_completion_result), "payload length:", len(self._last_completion_result.payloads))
+    def workflow_last_completion_result(
+        self, type_hint: Optional[Type]
+    ) -> Optional[Any]:
+        print(
+            "workflow_last_completion_result: ",
+            self._last_completion_result,
+            type(self._last_completion_result),
+            "payload length:",
+            len(self._last_completion_result.payloads),
+        )
         if len(self._last_completion_result.payloads) == 0:
             return None
         elif len(self._last_completion_result.payloads) > 1:
-            warnings.warn(f"Expected single last completion result, got {len(self._last_completion_result.payloads)}")
+            warnings.warn(
+                f"Expected single last completion result, got {len(self._last_completion_result.payloads)}"
+            )
             return None
 
         print("Payload:", self._last_completion_result.payloads[0])
-        return self._payload_converter.from_payload(self._last_completion_result.payloads[0], type_hint)
+        return self._payload_converter.from_payload(
+            self._last_completion_result.payloads[0], type_hint
+        )
 
     #### Calls from outbound impl ####
     # These are in alphabetical order and all start with "_outbound_".

temporalio/workflow.py

@@ -898,7 +898,9 @@ def workflow_get_current_details(self) -> str: ...
     def workflow_set_current_details(self, details: str): ...
 
     @abstractmethod
-    def workflow_last_completion_result(self, type_hint: Optional[Type]) -> Optional[Any]: ...
+    def workflow_last_completion_result(
+        self, type_hint: Optional[Type]
+    ) -> Optional[Any]: ...
 
 
 _current_update_info: contextvars.ContextVar[UpdateInfo] = contextvars.ContextVar(

tests/test_client.py

@@ -1541,4 +1541,3 @@ async def test_schedule_last_completion_result(
 
         await handle.delete()
         assert False
-

tests/worker/test_visitor.py

@@ -44,9 +44,7 @@ async def test_workflow_activation_completion():
                         schedule_to_close_timeout=Duration(seconds=5),
                         priority=Priority(),
                     ),
-                    user_metadata=UserMetadata(
-                        summary=Payload(data=b"Summary")
-                    ),
+                    user_metadata=UserMetadata(summary=Payload(data=b"Summary")),
                 )
             ],
         ),
@@ -78,9 +76,7 @@ async def test_workflow_activation():
                         Payload(data=b"repeated1"),
                         Payload(data=b"repeated2"),
                     ],
-                    headers={
-                        "header":Payload(data=b"map")
-                    },
+                    headers={"header": Payload(data=b"map")},
                     last_completion_result=Payloads(
                         payloads=[
                             Payload(data=b"obj1"),
@@ -89,9 +85,9 @@ async def test_workflow_activation():
                     ),
                     search_attributes=SearchAttributes(
                         indexed_fields={
-                            "sakey":Payload(data=b"saobj"),
+                            "sakey": Payload(data=b"saobj"),
                         }
-                    )
+                    ),
                 ),
             )
         ]
@@ -110,13 +106,29 @@ async def visitor(payload: Payload) -> Payload:
     assert act.jobs[0].initialize_workflow.arguments[0].metadata["visited"]
     assert act.jobs[0].initialize_workflow.arguments[1].metadata["visited"]
     assert act.jobs[0].initialize_workflow.headers["header"].metadata["visited"]
-    assert act.jobs[0].initialize_workflow.last_completion_result.payloads[0].metadata["visited"]
-    assert act.jobs[0].initialize_workflow.last_completion_result.payloads[1].metadata["visited"]
-    assert act.jobs[0].initialize_workflow.search_attributes.indexed_fields["sakey"].metadata["visited"]
+    assert (
+        act.jobs[0]
+        .initialize_workflow.last_completion_result.payloads[0]
+        .metadata["visited"]
+    )
+    assert (
+        act.jobs[0]
+        .initialize_workflow.last_completion_result.payloads[1]
+        .metadata["visited"]
+    )
+    assert (
+        act.jobs[0]
+        .initialize_workflow.search_attributes.indexed_fields["sakey"]
+        .metadata["visited"]
+    )
 
     act = original.__deepcopy__()
     await PayloadVisitor(skip_search_attributes=True).visit(visitor, act)
-    assert not act.jobs[0].initialize_workflow.search_attributes.indexed_fields["sakey"].metadata["visited"]
+    assert (
+        not act.jobs[0]
+        .initialize_workflow.search_attributes.indexed_fields["sakey"]
+        .metadata["visited"]
+    )
 
     act = original.__deepcopy__()
     await PayloadVisitor(skip_headers=True).visit(visitor, act)
@@ -198,6 +210,7 @@ async def visitor(payload: Payload) -> Payload:
     ur = cmds[4].update_response
     assert ur.completed.data == b"visited:e1"
 
+
 async def test_code_gen():
     # Smoke test the generated visitor on a simple activation containing payloads
     act = WorkflowActivation(
@@ -221,4 +234,4 @@ async def _f(p: Payload) -> Payload:
     init = act.jobs[0].initialize_workflow
     assert init.arguments[0].data == b"v:x1"
     assert init.arguments[1].data == b"v:x2"
-    assert init.headers["h"].data == b"v:x3"
+    assert init.headers["h"].data == b"v:x3"

tests/worker/test_workflow.py

@@ -8329,4 +8329,4 @@ async def test_workflow_headers_with_codec(
             assert headers["foo"].data == b"bar"
         else:
             assert headers["foo"].data != b"bar"
-    assert False
+    assert False