ecs = ECSTask(
        name=f"prefect2-ecs-task-{flow_name}",
        aws_credentials=AwsCredentials.load("aws-credentials"),
        image=f"{ecr_repository}:{image_tag}",
        cpu=cpu,
        memory=memory,
        stream_output=True,
        configure_cloudwatch_logs=True,
        cloudwatch_logs_options={
            "awslogs-group": flows_log_group,
        },
        cluster=ecs_cluster,
        task_role_arn=task_run_role,
        execution_role_arn=execution_role,
        vpc_id=vpc_id,
        task_customizations=[
            {
                "op": "add",
                "path": "/networkConfiguration/awsvpcConfiguration/subnets",
                "value": private_subnet_ids,
            },
            {
                "op": "add",
                "path": "/networkConfiguration/awsvpcConfiguration/securityGroups",
                "value": [security_group_id],
            },
        ],
        task_start_timeout_seconds=300,
        task_definition_arn=task_definition_arn
    )
    ecs.save(f"ecs-task-{flow_name}", overwrite=True)

prefect_flow_task_definition = ecs.TaskDefinition(
    self,
    "Prefect-Flow-Task-Definition",
    family=f"flow-task-definition-{environment}",
    cpu="4096",
    memory_mib="16384",
    compatibility=ecs.Compatibility.FARGATE,
    network_mode=ecs.NetworkMode.AWS_VPC,
)

prefect_flow_definition.add_container(
    "Prefect-Flow-Container",
    container_name=f"container-{environment}",
    image=ecs.ContainerImage.from_registry("prefecthq/prefect:2.8.2-python3.10"),
)