prefect-email

prefect-email

smtplib.SMTPAuthenticationError: (535, b'5.7.8 Username and Password not accepted. Learn more at\n5.7.8  <https://support.google.com/mail/?p=BadCredentials> d6-20020a170903230600b0016efc27ca98sm3023696plh.169 - gsmtp'

Result

FileSystems

file not found

/tmp

prefect-gcp

gcs_block = GCS.load("gcs-dev")

@flow()
def example_cloud_storage_upload_blob_from_file_flow():
    gcp_credentials = GcpCredentials(service_account_info=gcs_block.service_account_info)
    test_upload_file = "test_upload.txt"
    blob = cloud_storage_upload_blob_from_file(test_upload_path, gcs_block.bucket_path, "test_upload.txt", gcp_credentials)
    return blob

@task
def actual_task(arg1, arg2, arg3):
   #does stuff

task_list = [
    (1, "a", "b"),
    (2, "c", "d"),
    (3, "e", "f"),
    (4, "g", "h"),
]

def task_map(task):
    return actual_task(task[0], task[1], task[2])

with Flow("my_flow") as flow:
    task_results = apply_map(task_map, task_list)

task.fn(function_to_call())

AttributeError: 'function' object has no attribute 'fn'

task.fn()

retries

flow.storage = S3(
   bucket=DEPLOYMENT_BUCKET, stored_as_script=False, add_default_labels=False
)
flow.register(
   PROJECT_NAME,
   add_default_labels=False,
   idempotency_key=flow.serialized_hash(),
)

.submit

.submit()

from typing import Optional
from prefect.utilities.asyncutils import sync_compatible
from prefect.blocks.notifications import NotificationBlock

class MyEmail(NotificationBlock):

    _block_type_name = "My Email"
    _block_type_slug = "my-email"
    _block_schema_capabilities = ["notify"]
    
    @sync_compatible
    async def notify(self,body: str,subject: Optional[str] = None):
        await print( f"In my email notify subject: {subject}\nbody: {body}" )

- project
        └── flows
            └── flow1.py
            └── flow2.py
        └── util
            └── util.py

some/dir/project> prefect deployment build flows/flow1.py:flow_func -n my-flow -ib kubernetes-job/my-job -sb gcs/my-bucket -t k8s

FileNotFoundError: [Errno 2] No such file or directory: '/opt/prefect/flows/flow1.py'

upstream_tasks

Task Runners

10:21:24.290 | INFO    | prefect.agent - Submitting flow run 'cfc4f262-4f05-4685-882e-364192297107'
10:21:24.474 | INFO    | prefect.infrastructure.process - Opening process 'blond-mammoth'...
10:21:24.482 | INFO    | prefect.agent - Completed submission of flow run 'cfc4f262-4f05-4685-882e-364192297107'
10:21:27.334 | ERROR   | Flow run 'blond-mammoth' - Flow could not be retrieved from deployment.
Traceback (most recent call last):
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/prefect/engine.py", line 247, in retrieve_flow_then_begin_flow_run
    flow = await load_flow_from_flow_run(flow_run, client=client)
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/prefect/client.py", line 104, in with_injected_client
    return await fn(*args, **kwargs)
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/prefect/deployments.py", line 47, in load_flow_from_flow_run
    await storage_block.get_directory(from_path=None, local_path=".")
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/prefect/filesystems.py", line 373, in get_directory
    return await self.filesystem.get_directory(
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/prefect/filesystems.py", line 251, in get_directory
    return self.filesystem.get(from_path, local_path, recursive=True)
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/fsspec/spec.py", line 801, in get
    self.get_file(rpath, lpath, **kwargs)
  File "/opt/micromamba/envs/main/lib/python3.8/site-packages/fsspec/spec.py", line 769, in get_file
    outfile = open(lpath, "wb")
FileNotFoundError: [Errno 2] No such file or directory: '/tmp/tmp3er_ugnvprefect/S3-BUCKET-NAME/alpha/flow.py'
10:21:27.727 | INFO    | prefect.infrastructure.process - Process 'blond-mammoth' exited cleanly.
... 
$ aws s3 ls <s3://S3-BUCKET-NAME/alpha/>
2022-08-06 10:20:49       6473 flow.py
2022-08-06 10:20:49       3204 example_flow-manifest.json

# prefect deployment build ./flow.py:example_flow --name example-flow-alpha --tag alpha --storage-block s3/S3-BUCKET-NAME
# prefect deployment apply example-flow-alpha.yaml