Workflow implementation (#21)

.gitignore

@@ -4,6 +4,7 @@ __pycache__
 /dist
 /docs/_autosummary
 /docs/_build
+/docs/_build_pydoctor
 temporalio/api/*
 !temporalio/api/__init__.py
 temporalio/bridge/proto/*

docs/index.rst

@@ -12,6 +12,7 @@ Modules
 
    temporalio.client
    temporalio.worker
+   temporalio.workflow
    temporalio.activity
    temporalio.converter
    temporalio.exceptions

pyproject.toml

@@ -37,9 +37,10 @@ black = "^21.12b0"
 furo = "^2022.3.4"
 grpcio-tools = "^1.43.0"
 isort = "^5.10.1"
-mypy = "^0.931"
+mypy = "^0.950"
 mypy-protobuf = "^3.2.0"
 pydocstyle = "^6.1.1"
+pydoctor = "^22.5.1"
 pytest = "^7.1.1"
 pytest-asyncio = "^0.18.0"
 pytest-timeout = "^2.1.0"
@@ -56,14 +57,17 @@ build-bridge-develop = "python scripts/setup_bridge.py develop"
 fix-wheel = "python scripts/fix_wheel.py"
 format = [{cmd = "black ."}, {cmd = "isort ."}]
 gen-docs = "sphinx-build docs docs/_build"
+gen-docs-pydoctor = "pydoctor"
 gen-protos = "python scripts/gen_protos.py"
 lint = [
   {cmd = "black --check ."},
   {cmd = "isort --check-only ."},
   {ref = "lint-types"},
   {ref = "lint-docs"},
 ]
-lint-docs = "pydocstyle"
+# TODO(cretz): Why does pydocstyle complain about @overload missing docs after
+# https://github.com/PyCQA/pydocstyle/pull/511?
+lint-docs = "pydocstyle --ignore-decorators=overload"
 lint-types = "mypy ."
 test = "pytest"
 
@@ -86,7 +90,7 @@ asyncio_mode = "auto"
 log_cli = true
 log_cli_level = "INFO"
 log_cli_format = "%(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)"
-timeout = 60
+timeout = 600
 timeout_func_only = true
 
 [tool.isort]
@@ -111,6 +115,15 @@ add_ignore = [
   "D205", "D415"
 ]
 
+[tool.pydoctor]
+add-package = ["temporalio"]
+docformat = "google"
+html-output = "docs/_build_pydoctor"
+intersphinx = ["https://docs.python.org/3/objects.inv", "https://googleapis.dev/python/protobuf/latest/objects.inv"]
+privacy = ["PRIVATE:temporalio.bridge", "HIDDEN:**.*_pb2*"]
+project-name = "Temporal"
+sidebar-expand-depth = 40
+
 [build-system]
 build-backend = "poetry.core.masonry.api"
 requires = ["poetry-core>=1.0.0", "setuptools", "wheel", "setuptools-rust"]

scripts/gen_protos.py

@@ -68,9 +68,17 @@ def fix_generated_output(base_path: Path):
                 f.write(content)
     # Write init
     with (base_path / "__init__.py").open("w") as f:
+        # Imports
+        message_names = []
         for stem, messages in imports.items():
             for message in messages:
                 f.write(f"from .{stem} import {message}\n")
+                message_names.append(message)
+        # __all__
+        if message_names:
+            f.write(
+                f'\n__all__ = [\n    "' + '",\n    "'.join(message_names) + '",\n]\n'
+            )
 
 
 if __name__ == "__main__":

temporalio/activity.py

@@ -11,6 +11,7 @@
 
 import asyncio
 import contextvars
+import inspect
 import logging
 import threading
 from dataclasses import dataclass
@@ -57,15 +58,8 @@ def defn(fn: Optional[ActivityFunc] = None, *, name: Optional[str] = None):
     """
 
     def with_name(name: str, fn: ActivityFunc) -> ActivityFunc:
-        # Validate the activity
-        if not callable(fn):
-            raise TypeError("Activity is not callable")
-        elif not fn.__code__:
-            raise TypeError("Activity callable missing __code__")
-        elif fn.__code__.co_kwonlyargcount:
-            raise TypeError("Activity cannot have keyword-only arguments")
-        # Set the name
-        setattr(fn, "__temporal_activity_name", name)
+        # This performs validation
+        _Definition._apply_to_callable(fn, name)
         return fn
 
     # If name option is available, return decorator function
@@ -320,7 +314,7 @@ def process(
     ) -> Tuple[Any, MutableMapping[str, Any]]:
         """Override to add activity details."""
         msg, kwargs = super().process(msg, kwargs)
-        if self.activity_info_on_extra or self.activity_info_on_extra:
+        if self.activity_info_on_message or self.activity_info_on_extra:
             context = _current_context.get(None)
             if context:
                 if self.activity_info_on_message:
@@ -342,3 +336,45 @@ def base_logger(self) -> logging.Logger:
 
 #: Logger that will have contextual activity details embedded.
 logger = LoggerAdapter(logging.getLogger(__name__), None)
+
+
+@dataclass
+class _Definition:
+    name: str
+    fn: Callable
+    is_async: bool
+
+    @staticmethod
+    def from_callable(fn: Callable) -> Optional[_Definition]:
+        return getattr(fn, "__temporal_activity_definition", None)
+
+    @staticmethod
+    def must_from_callable(fn: Callable) -> _Definition:
+        ret = _Definition.from_callable(fn)
+        if ret:
+            return ret
+        fn_name = getattr(fn, "__name__", "<unknown>")
+        raise TypeError(
+            f"Activity {fn_name} missing attributes, was it decorated with @activity.defn?"
+        )
+
+    @staticmethod
+    def _apply_to_callable(fn: Callable, activity_name: str) -> None:
+        # Validate the activity
+        if hasattr(fn, "__temporal_activity_definition"):
+            raise ValueError("Function already contains activity definition")
+        elif not callable(fn):
+            raise TypeError("Activity is not callable")
+        elif not fn.__code__:
+            raise TypeError("Activity callable missing __code__")
+        elif fn.__code__.co_kwonlyargcount:
+            raise TypeError("Activity cannot have keyword-only arguments")
+        setattr(
+            fn,
+            "__temporal_activity_definition",
+            _Definition(
+                name=activity_name,
+                fn=fn,
+                is_async=inspect.iscoroutinefunction(fn),
+            ),
+        )

temporalio/bridge/proto/__init__.py

@@ -1 +1,6 @@
 from .core_interface_pb2 import ActivityHeartbeat, ActivityTaskCompletion
+
+__all__ = [
+    "ActivityHeartbeat",
+    "ActivityTaskCompletion",
+]

temporalio/bridge/src/client.rs

@@ -4,15 +4,15 @@ use pyo3_asyncio::tokio::future_into_py;
 use std::collections::HashMap;
 use std::time::Duration;
 use temporal_client::{
-    ConfiguredClient, RetryConfig, RetryGateway, ServerGatewayOptions, ServerGatewayOptionsBuilder,
-    TlsConfig, WorkflowService, WorkflowServiceClientWithMetrics,
+    ClientOptions, ClientOptionsBuilder, ConfiguredClient, RetryClient, RetryConfig, TlsConfig,
+    WorkflowService, WorkflowServiceClientWithMetrics,
 };
 use tonic;
 use url::Url;
 
 pyo3::create_exception!(temporal_sdk_bridge, RPCError, PyException);
 
-type Client = RetryGateway<ConfiguredClient<WorkflowServiceClientWithMetrics>>;
+type Client = RetryClient<ConfiguredClient<WorkflowServiceClientWithMetrics>>;
 
 #[pyclass]
 pub struct ClientRef {
@@ -51,7 +51,7 @@ struct ClientRetryConfig {
 
 pub fn connect_client(py: Python, config: ClientConfig) -> PyResult<&PyAny> {
     // TODO(cretz): Add metrics_meter?
-    let opts: ServerGatewayOptions = config.try_into()?;
+    let opts: ClientOptions = config.try_into()?;
     future_into_py(py, async move {
         Ok(ClientRef {
             retry_client: opts.connect_no_namespace(None).await.map_err(|err| {
@@ -226,11 +226,11 @@ where
     }
 }
 
-impl TryFrom<ClientConfig> for temporal_client::ServerGatewayOptions {
+impl TryFrom<ClientConfig> for ClientOptions {
     type Error = PyErr;
 
     fn try_from(opts: ClientConfig) -> PyResult<Self> {
-        let mut gateway_opts = ServerGatewayOptionsBuilder::default();
+        let mut gateway_opts = ClientOptionsBuilder::default();
         gateway_opts
             .target_url(
                 Url::parse(&opts.target_url)