@Marvin What's a secure way to use the AWS Credentials block if you've got temporary ones? (ie, the kind where you need to supply a Token). Normally I just create the

ECSTask

block fresh every time, but that's not quite working with the ECS Push worker.

To use temporary AWS credentials with the ECS Push worker, you'll need to supply the AWS Access Key ID, AWS Secret Access Key, and the AWS Session Token. These credentials are typically obtained from AWS STS (Security Token Service). However, it seems like you're encountering an issue with this setup. Could you provide more details about the problem you're facing? This will help me provide a more accurate solution. Also, please note that the Prefect ECS Task Runner does not currently support temporary AWS credentials directly. You might need to manually handle the refreshing of these credentials if they expire within the lifetime of your task. For more detailed information, you might want to check out the Prefect documentation or community discussions. Here are a few relevant links I found: 1. No class found for dispatch key s3 bucket in registry for type block working with deployments 2. Push Work to Serverless Computing Infrastructure Please let me know if you need more help!

can you show what you do currently to get your fresh token in an instance of AwsCredentials?

@Nate I don't use it at the moment. Right now when I deploy, I generate a dictionary with a bunch of args and then pass it to

ECSTask

like so

full_ecs_args = make_ec2_args(memory, user, full_image, cpu)
    ecs_task_block = ECSTask(**full_ecs_args)

and then pass that to the

infrastructure

arg of

Deployment.build_from_flow

gotcha gotcha it feels like there could be some sort of optional (callable) handler on AwsCredentials to load the token value at instantiation time. is your token always accessible from a certain place? i.e. the place you create an instance ... ah i think you answered the question i was typing

I did try

creds = boto3.session.Session().get_credentials().get_frozen_credentials()
credentials = AwsCredentials(
    **{
        "aws_access_key_id": creds.access_key,
        "aws_secret_access_key": creds.secret_key,
        "aws_session_token": creds.token,
        "region_name": "us-west-2"
    }
)

and then passing the object as

aws_credentials

to the dictionary I pass to

job_variables

in

.deploy

(this doesnt currently exist but) if we added some validator like

class AwsCredentials:

   refresh_token_fn: Optional[Callable] = None

   @validator
   def get_refreshed_token(cls, v):
      # load from somewhere at instantiation time
      if refresh_token_fn:
         return refresh_token_fn()
      return v

which one could test quickly by subclassing

AwsCredentials

and tacking these things on

Interesting!

maybe? idk - i can come back to this later with fresh eyes

Thanks!