now able to use --upload arg as an option to enable lakefs versioning…

…, requires a repo name

src/lake_fs.py

@@ -18,12 +18,12 @@ def execute_command(command):
             logging.error("Error executing command: %s", e.stderr)
             return e
 
-def lakefs_merge_into_main():
+def lakefs_merge_into_main(repo):
         logging.info("Uploading files from data directory to branch...")
         command = [
             "lakectl", "merge",
-            f"lakefs://example-repo/ingestion/",
-            f"lakefs://example-repo/main/",
+            f"lakefs://{repo}/ingestion/",
+            f"lakefs://{repo}/main/",
             "-m", "Merge ingestion branch into main"
         ]
         execute_command(command)

src/main.py

@@ -23,7 +23,7 @@ def main():
     parser.add_argument("--credentials_file", default="lakectl.cfg")
     parser.add_argument("--serial", default=False, action='store_true')
     parser.add_argument("--endpoint_url", default="http://localhost:8000")
-    parser.add_argument("--upload", default=False, action="store_true")
+    parser.add_argument("--upload", nargs='?', const=False, default=False)
     parser.add_argument("--metadata_dir", default="data/uda")
     parser.add_argument("--force", action="store_true")
     parser.add_argument("--signal_names", nargs="*", default=[])
@@ -36,6 +36,7 @@ def main():
         config = LakeFSUploadConfig(
             credentials_file=args.credentials_file,
             endpoint_url=args.endpoint_url,
+            repository=args.upload
         )
         workflow_cls = partial(LakeFSIngestionWorkflow, upload_config=config)
     else:
@@ -63,7 +64,7 @@ def main():
         logging.info(f"Finished source {source}")
 
     if args.upload:
-        lakefs_merge_into_main()
+        lakefs_merge_into_main(args.upload)
 
 if __name__ == "__main__":
     main()

src/task.py

@@ -44,7 +44,7 @@ def __call__(self):
         logging.info(f"Uploading {self.local_file} to LakeFS.")
         args = [
         "lakectl", "fs", "upload",
-        f"lakefs://example-repo/ingestion/{self.shot_name}",
+        f"lakefs://{self.config.repository}/ingestion/{self.shot_name}",
         "-s", str(self.local_file), "--recursive"
         ]
 
@@ -78,7 +78,7 @@ def __call__(self):
         logging.info(f"Commit {self.local_file} to branch.")
         args = [
             "lakectl", "commit",
-            f"lakefs://example-repo/ingestion/",
+            f"lakefs://{self.config.repository}/ingestion/",
             "-m", f"Commit file {self.local_file}"
         ]
 

src/uploader.py

@@ -11,3 +11,4 @@ class UploadConfig:
 class LakeFSUploadConfig:
     endpoint_url: str
     credentials_file: str
+    repository: str