Nick DeCraene
10/06/2022, 4:47 PMNone
instead of what is hinted at in the method. Does anyone know of a work-around besides ignoring the errors?@task(retries=5, retry_delay_seconds=30)
def extract(ctx: RunContext, today: Optional[dt.datetime] = None) -> List[int]:
@task()
def get_ctx() -> RunContext:
return RunContext()
@flow()
def my_flow(date: Optional[dt.dateime]):
ctx = get_ctx()
data = extract(ctx, date)
for d in data:
do_something(d)
will issue
error: No overload variant of "__call__" of "Task" matches argument types "None", "Optional[datetime]" [call-overload]
extract
is getting a None
type passed for the ctx argRob Freedy
10/07/2022, 1:45 PMNick DeCraene
10/07/2022, 2:19 PMList[Tuple[int, str]]
insteadRob Freedy
10/07/2022, 5:59 PMNick DeCraene
10/11/2022, 8:35 PM@task(retries=3, retry_delay_seconds=10)
def get_cids(
cid_to_run: Optional[str] = None,
org_to_run: Optional[int] = None,
all_orgs: bool = False,
) -> List[Tuple[int, str]]:
"""Get a list cids and org ids to pull data for [(org, cid), (org, cid), ...]"""
with ro_session_scope() as session:
info = get_info(session, cid_to_run, org_to_run, all_orgs)
return [(i[0], i[1]) for i in info]
@flow(name="my_flow")
def run_experiments(
cid_to_run: Optional[str] = None,
org_to_run: Optional[int] = None,
):
org_cids: List[Tuple[int, str]] = get_cids(cid_to_run, org_to_run, all_orgs=True)
# rest of the flow here
get_cids pretty much just grabs some info from a db, but always returns a List[Tuple[int, str]]. The mypy issue occurs when assigning org_cids
as None
can't be cast to List
Rob Freedy
10/12/2022, 7:06 PM