from prefect.tasks.secrets import PrefectSecret

from prefect.client import Secret

from prefect import task, Flow

import prefect

from prefect.tasks import aws

from prefect.utilities.aws import get_boto_client

from prefect.utilities.tasks import defaults_from_attrs

import datetime

import json

#Connect to Prefect

client = prefect.Client()

access_key = Secret("ACCESS_KEY").get()

secret_access_key = Secret("SECRET_ACCESS_KEY").get()

creds= {"ACCESS_KEY":access_key, "SECRET_ACCESS_KEY":secret_access_key}

aws_credentials= json.dumps(creds)

print(aws_credentials)

#Connect to AWS

# lambda_client = boto3.client('lambda', aws_access_key_id=access_key, aws_secret_access_key=secret_access_key)

@task(max_retries=2, retry_delay=datetime.timedelta(seconds=5))

def InvokeLambdaFunction():

prefect_lambda= prefect.tasks.aws.lambda_function.LambdaList(master_region="ap-southeast-2", function_version="ALL", marker=None, max_items=50)

prefect_lambda.run(credentials=aws_credentials)

# prefect_s3 = prefect.tasks.aws.s3.S3List(bucket='oes-cloud-trail-logs')

# prefect_s3.run(credentials=aws_credentials)

flow = Flow("Invoke-lambda", tasks=[InvokeLambdaFunction])

# flow.run() We could run our flow locally using the flow's run method but we'll be running this from Cloud!

flow.register(project_name="Test PoC")

flow.run()