def _create()

in custom-resource/dms_task_conf/task.py [0:0]

• 13 lines of code
• 3 McCabe index (conditional complexity)

def _create(replication_task_arn):
    try:
        with open("task_settings.json") as f:
            data = json.load(f)
        client.modify_replication_task(
            ReplicationTaskArn=replication_task_arn,
            ReplicationTaskSettings=json.dumps(data),
        )
    except FileNotFoundError as e:
        logger.error("Unable to load task config")
        logger.error(e)
    except ClientError as e:
        logger.error(e)