fix gcs inputs

[sharvenm]

fix gcs inputs in generic file source
update sample dag inputs

[eugenemiretsky]

Are we able to add unit tests?

[rabdulatipoff]

DAGs config looks fine, though I'm not sure what using an empty bucket prefix would mean (would it be a filesystem or a Cloud Ctorage lookup?), also I'd use os.path routines instead of concatenating path strings manually -- looks fine otherwise

[soyarym]

DAGs config looks fine, though I'm not sure what using an empty bucket prefix would mean (would it be a filesystem or a Cloud Ctorage lookup?), also I'd use os.path routines instead of concatenating path strings manually -- looks fine otherwise

I agree.
@eugenemiretsky @sharvenm could you please give an example of bucket prefix we are expecting?

[sharvenm]

@rabdulatipoff @soyarym an empty bucket prefix would mean that the file (blob since really we're only dealing with gcs for these buckets) we are searching for is not located inside any "folders". So if our file name was "users.csv" and our bucketname was "gcs-airflow"

The path to that blob should be: gs://gcs-airflow/users.csv

If instead the blob was located under "folders" (gcs doesn't actually have folders, but we can think of blob names like that) and the full path to the blob was: gs://gcs-airflow/some-folder/users.csv then we would need the gcs-prefix to be "some-folder"

I like the idea of using os.path.join so I've added that as you guys recommended.

[soyarym]

@sharvenm
Oh, okay, got it.
I guess if you like the idea of using os.path.join then maybe would better to use it in the other part of the script instead of '/' in the same cases?