MyPy Fixes for Worker, Workflow metaclass, Context, and some others
#273
Conversation
|
|
||
|
|
||
| @dataclass |
There was a problem hiding this comment.
Let me know if I'm wrong on this and this should be a dataclass!
* feat: fix mypy for Workflow * fix: type stub * fix: remove any in a couple places * fix: attempting using `Union` * fix: try with `.Value` * fix: name attr * fix: remove more type: ignore * MyPy for Context, etc. (#276) * fix: types for config, etc. * fix: queue type hint * fix: typeddict items
hatchet-temporary
changed the title
WorkerWorker, Workflow metaclass, Context, and some others
| @@ -54,7 +63,7 @@ def _prepare_workflow_options( | |||
| if options is not None and "additional_metadata" in options: | |||
| meta = options["additional_metadata"] | |||
|
|
|||
| trigger_options: TriggerWorkflowOptions = { | |||
| trigger_options: TriggerWorkflowOptions = { # type: ignore[typeddict-item] | |||
There was a problem hiding this comment.
this is here because namespace isn't set - if we do set it, it causes problems downstream. I think Pydantic here would be helpful instead of using typed dicts
| for child_workflow_run in child_workflow_runs: | ||
| workflow_name = child_workflow_run["workflow_name"] | ||
| input = child_workflow_run["input"] | ||
|
|
||
| key = child_workflow_run.get("key") | ||
| options = child_workflow_run.get("options", {}) | ||
|
|
||
| trigger_options = self._prepare_workflow_options(key, options, worker_id) | ||
| ## TODO: figure out why this is failing | ||
| trigger_options = self._prepare_workflow_options(key, options, worker_id) # type: ignore[arg-type] |
There was a problem hiding this comment.
weird type error here on the options that I'll figure out later
| ## TODO: Use Protocol here to help with MyPy errors | ||
| func._step_name = name.lower() or str(func.__name__).lower() # type: ignore[attr-defined] | ||
| func._step_parents = parents # type: ignore[attr-defined] | ||
| func._step_timeout = timeout # type: ignore[attr-defined] | ||
| func._step_retries = retries # type: ignore[attr-defined] | ||
| func._step_rate_limits = limits # type: ignore[attr-defined] | ||
| func._step_name = name.lower() or str(func.__name__).lower() | ||
| func._step_parents = parents | ||
| func._step_timeout = timeout | ||
| func._step_retries = retries | ||
| func._step_rate_limits = limits |
There was a problem hiding this comment.
using the protocol lets us get rid of all of these type issues
| config: ClientConfig = {}, | ||
| config: ClientConfig = ClientConfig(), |
There was a problem hiding this comment.
pretty sure this default ended up being overridden downstream, but just checking the dictionary was the wrong type
| def thread_action_func(self, context, action_func, action: Action): | ||
| ## TODO: Stricter type hinting here | ||
| def thread_action_func( | ||
| self, context: Context, action_func: Callable[..., Any], action: Action |
There was a problem hiding this comment.
one note for the future - it'd be awesome to make this action_func conform to something stricter than just Callable[..., Any], but we can address that later I think
| @@ -267,7 +277,7 @@ def cleanup_run_id(self, run_id: str): | |||
| def create_context( | |||
| self, action: Action, action_func: Callable[..., Any] | None | |||
| ) -> Context | DurableContext: | |||
| if hasattr(action_func, "durable") and action_func.durable: | |||
| if hasattr(action_func, "durable") and getattr(action_func, "durable"): | |||
There was a problem hiding this comment.
escape hatch
| """Terminate a python threading.Thread.""" | ||
| try: | ||
| if not thread.is_alive(): | ||
| return | ||
|
|
||
| logger.info(f"Forcefully terminating thread {thread.ident}") | ||
| ident = cast(int, thread.ident) |
There was a problem hiding this comment.
according to the package docs, this is always an int. but according to their type hints, it's not 🤷 going with what the docs say and casting here, but lmk if this could ever be None
There was a problem hiding this comment.
no idea, I've only ever seen int
| action_function.is_coroutine = True | ||
| setattr(action_function, "is_coroutine", True) | ||
| else: | ||
| action_function.is_coroutine = False | ||
| setattr(action_function, "is_coroutine", False) |
There was a problem hiding this comment.
escape hatch for now - I think we can probably define another Protocol for the action_func, but will do that later
| steps: stepsType = [ | ||
| (getattr(func, "_step_name"), attrs.pop(func_name)) | ||
| for func_name, func in list(attrs.items()) | ||
| if hasattr(func, "_step_name") | ||
| ] |
There was a problem hiding this comment.
factored out into a helper
| poetry run black . --color | ||
| poetry run isort . | ||
| poetry run mypy --config-file=pyproject.toml |
There was a problem hiding this comment.
modified this to run the linters on all the files, since it only takes a couple seconds
A couple things here:
Workerdoesn't need to be adataclass, and this was making things a bit more complicated (b/c ofpost_init, etc.)hatchet_sdk/worker/worker.pyto the included MyPy paths and fixed all the type hints (except for a couple tricky ones with theWorkflowMetametaclass - we can handle those latersetattrto make the type checker happyProtocolfor theWorkflowMetametaclass for the same reasonWorkflowmetaclass andContextWill try to annotate!