Adam Green
10/12/2022, 12:50 AMfrom prefect.deployments import Deployment
from prefect.filesystems import S3
s3_block = S3(
aws_access_key_id=aws_key,
aws_secret_access_key=aws_secret,
bucket_path=context["prefect_flows_bucket"],
)
s3_block.save("s3", overwrite=True)
Deployment.build_from_flow(
name="alpha",
work_queue_name="alpha",
flow=healthcheck,
storage=S3.load("s3"),
infrastructure=Process(),
apply=True,
)
When we run mypy on this code, it complains about things not being typed as async. Is it possible to type this code without converting to async?Anna Geller
10/12/2022, 12:59 AMAdam Green
10/12/2022, 1:37 AMdataflow-ops
a lot - it’s great ❤️
why don’t you leverage the CLI insteadWe started out using the CLI but have ended up doing everything in Python. We will reconsider this decision in light of the async requirement for static typing of Prefect deployments.
Ryan Peden
10/12/2022, 4:19 AMresult = s3_block.save("s3", overwrite=True)
assert result is not None
and then something similar for Deployment.build_from_flow
.
I'd expect both methods to throw an exception if something happened that would cause them to return None
. But doing it this way (if you don't find it to be a hassle) should make MyPy happy and also give you an extra bit of validation if you run your deployments with assertions enabled.
You could also skip the assert, but then your tooling might be unhappy about the unused result
variable.
To add a bit of detail on what is happening: it is fine to run both those methods non asynchronously because they use the sync_compatible decorator, but MyPy doesn't know it's okay to run them that way and so it complains if you just call them without either awaiting them or assigning their result to something.