For anyone else struggling with the same, I was overcomplicating this. 
basically next_snapshot runs as a loop itself. so you don't need to provide a separate loop for it to work.

Please check the below code:

def next_snapshot(latest_version):
  """
    latest_version is the version of the latest snapshot
  """
  print("***** Start next_snapshot() ", latest_version)
  if latest_version is None:
    version = versions[0]
  else:
    version_idx = versions.index(latest_version) + 1
    if version_idx > len(versions)-1:
      return None
    version = versions[version_idx]
  
  snapshot_path = f"{snapshot_root}"
  if exist(snapshot_path):
    return (spark.read.format("parquet").load(snapshot_path).select("*", "_metadata"), version)
  else:
    return None




dlt.create_streaming_table(
  name =(table_name),
  table_properties={"quality": "bronze","delta.enableChangeDataFeed":"true"}
)

  
dlt.apply_changes_from_snapshot(
  target = table_name,
  snapshot_and_version = next_snapshot,
  keys = ["id"],
  stored_as_scd_type = 2,
  # track_history_column_list = ["TrackingCol"] ### if you want to track history of specific columns
  track_history_except_column_list=["TrackingCol","TrackingCol","TrackingCol"]  ### if you want to exclude specific columns
)