Nicholas
task
mypy
from typing import Any, cast import httpx from prefect import flow, task @task(retries=3) def call(url: str) -> dict[str, Any]: response = httpx.get(url) if response.status_code >= 400: raise Exception() return cast(dict[str, Any], response.json()) @task def parse(response: dict[str, Any]) -> str: return cast(str, response["fact"]) @flow def run_demo_prefect(url: str) -> None: fact_response = call(url) parse(fact_response) if __name__ == "__main__": run_demo_prefect("<https://catfact.ninja/fact>")
example.py:23: error: No overload variant of "__call__" of "Task" matches argument type "None" [call-overload] example.py:23: note: Possible overload variants: example.py:23: note: def __call__(self, response: Dict[str, Any]) -> None example.py:23: note: def __call__(self, response: Dict[str, Any]) -> str example.py:23: note: def __call__(self, *args: Any, return_state: Literal[True], **kwargs: Any) -> State[str]
Diego Escobedo
Nate
❯ mypy mypytest.py --follow-imports skip Success: no issues found in 1 source file
Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.