8
8
import sys
9
9
import typing as t
10
10
from pathlib import Path
11
+ from typing import Union
11
12
12
13
from colorama import init
13
14
14
15
from vcspull ._internal import logger
15
16
from vcspull .config import load_config
17
+ from vcspull .config .migration import migrate_all_configs , migrate_config_file
16
18
from vcspull .config .models import VCSPullConfig
17
19
from vcspull .operations import (
18
20
apply_lock ,
@@ -49,6 +51,7 @@ def cli(argv: list[str] | None = None) -> int:
49
51
add_detect_command (subparsers )
50
52
add_lock_command (subparsers )
51
53
add_apply_lock_command (subparsers )
54
+ add_migrate_command (subparsers )
52
55
53
56
args = parser .parse_args (argv if argv is not None else sys .argv [1 :])
54
57
@@ -67,6 +70,8 @@ def cli(argv: list[str] | None = None) -> int:
67
70
return lock_command (args )
68
71
if args .command == "apply-lock" :
69
72
return apply_lock_command (args )
73
+ if args .command == "migrate" :
74
+ return migrate_command (args )
70
75
71
76
return 0
72
77
@@ -247,6 +252,64 @@ def add_apply_lock_command(subparsers: argparse._SubParsersAction[t.Any]) -> Non
247
252
)
248
253
249
254
255
+ def add_migrate_command (subparsers : argparse ._SubParsersAction [t .Any ]) -> None :
256
+ """Add the migrate command to the parser.
257
+
258
+ Parameters
259
+ ----------
260
+ subparsers : argparse._SubParsersAction
261
+ Subparsers action to add the command to
262
+ """
263
+ parser = subparsers .add_parser (
264
+ "migrate" ,
265
+ help = "Migrate configuration files to the latest format" ,
266
+ description = (
267
+ "Migrate VCSPull configuration files from old format to new "
268
+ "Pydantic-based format"
269
+ ),
270
+ )
271
+ parser .add_argument (
272
+ "config_paths" ,
273
+ nargs = "*" ,
274
+ help = (
275
+ "Paths to configuration files to migrate (defaults to standard "
276
+ "paths if not provided)"
277
+ ),
278
+ )
279
+ parser .add_argument (
280
+ "-o" ,
281
+ "--output" ,
282
+ help = (
283
+ "Path to save the migrated configuration (if not specified, "
284
+ "overwrites the original)"
285
+ ),
286
+ )
287
+ parser .add_argument (
288
+ "-n" ,
289
+ "--no-backup" ,
290
+ action = "store_true" ,
291
+ help = "Don't create backup files of original configurations" ,
292
+ )
293
+ parser .add_argument (
294
+ "-f" ,
295
+ "--force" ,
296
+ action = "store_true" ,
297
+ help = "Force migration even if files are already in the latest format" ,
298
+ )
299
+ parser .add_argument (
300
+ "-d" ,
301
+ "--dry-run" ,
302
+ action = "store_true" ,
303
+ help = "Show what would be migrated without making changes" ,
304
+ )
305
+ parser .add_argument (
306
+ "-c" ,
307
+ "--color" ,
308
+ action = "store_true" ,
309
+ help = "Colorize output" ,
310
+ )
311
+
312
+
250
313
def info_command (args : argparse .Namespace ) -> int :
251
314
"""Handle the info command.
252
315
@@ -628,3 +691,143 @@ def filter_repositories_by_paths(
628
691
setattr (filtered_config , attr_name , getattr (config , attr_name ))
629
692
630
693
return filtered_config
694
+
695
+
696
+ def migrate_command (args : argparse .Namespace ) -> int :
697
+ """Migrate configuration files to the latest format.
698
+
699
+ Parameters
700
+ ----------
701
+ args : argparse.Namespace
702
+ Parsed command line arguments
703
+
704
+ Returns
705
+ -------
706
+ int
707
+ Exit code
708
+ """
709
+ from colorama import Fore , Style
710
+
711
+ use_color = args .color
712
+
713
+ def format_status (success : bool ) -> str :
714
+ """Format success status with color if enabled."""
715
+ if not use_color :
716
+ return "Success" if success else "Failed"
717
+
718
+ if success :
719
+ return f"{ Fore .GREEN } Success{ Style .RESET_ALL } "
720
+ return f"{ Fore .RED } Failed{ Style .RESET_ALL } "
721
+
722
+ # Determine paths to process
723
+ if args .config_paths :
724
+ # Convert to strings to satisfy Union[str, Path] typing requirement
725
+ paths_to_process : list [str | Path ] = list (args .config_paths )
726
+ else :
727
+ # Use default paths if none provided
728
+ default_paths = [
729
+ Path ("~/.config/vcspull" ).expanduser (),
730
+ Path ("~/.vcspull" ).expanduser (),
731
+ Path .cwd (),
732
+ ]
733
+ paths_to_process = [str (p ) for p in default_paths if p .exists ()]
734
+
735
+ # Show header
736
+ if args .dry_run :
737
+ print ("Dry run: No files will be modified" )
738
+ print ()
739
+
740
+ create_backups = not args .no_backup
741
+
742
+ # Process single file if output specified
743
+ if args .output and len (paths_to_process ) == 1 :
744
+ path_obj = Path (paths_to_process [0 ])
745
+ if path_obj .is_file ():
746
+ source_path = path_obj
747
+ output_path = Path (args .output )
748
+
749
+ try :
750
+ if args .dry_run :
751
+ from vcspull .config .migration import detect_config_version
752
+
753
+ version = detect_config_version (source_path )
754
+ needs_migration = version == "v1" or args .force
755
+ print (f"Would migrate: { source_path } " )
756
+ print (f" - Format: { version } " )
757
+ print (f" - Output: { output_path } " )
758
+ print (f" - Needs migration: { 'Yes' if needs_migration else 'No' } " )
759
+ else :
760
+ success , message = migrate_config_file (
761
+ source_path ,
762
+ output_path ,
763
+ create_backup = create_backups ,
764
+ force = args .force ,
765
+ )
766
+ status = format_status (success )
767
+ print (f"{ status } : { message } " )
768
+
769
+ return 0
770
+ except Exception as e :
771
+ logger .exception (f"Error migrating { source_path } " )
772
+ print (f"Error: { e } " )
773
+ return 1
774
+
775
+ # Process multiple files or directories
776
+ try :
777
+ if args .dry_run :
778
+ from vcspull .config .loader import find_config_files
779
+ from vcspull .config .migration import detect_config_version
780
+
781
+ config_files = find_config_files (paths_to_process )
782
+ if not config_files :
783
+ print ("No configuration files found" )
784
+ return 0
785
+
786
+ print (f"Found { len (config_files )} configuration file(s):" )
787
+
788
+ # Process files outside the loop to avoid try-except inside loop
789
+ configs_to_process = []
790
+ for file_path in config_files :
791
+ try :
792
+ version = detect_config_version (file_path )
793
+ needs_migration = version == "v1" or args .force
794
+ configs_to_process .append ((file_path , version , needs_migration ))
795
+ except Exception as e :
796
+ if use_color :
797
+ print (f"{ Fore .RED } Error{ Style .RESET_ALL } : { file_path } - { e } " )
798
+ else :
799
+ print (f"Error: { file_path } - { e } " )
800
+
801
+ # Display results
802
+ for file_path , version , needs_migration in configs_to_process :
803
+ status = "Would migrate" if needs_migration else "Already migrated"
804
+
805
+ if use_color :
806
+ status_color = Fore .YELLOW if needs_migration else Fore .GREEN
807
+ print (
808
+ f"{ status_color } { status } { Style .RESET_ALL } : { file_path } ({ version } )"
809
+ )
810
+ else :
811
+ print (f"{ status } : { file_path } ({ version } )" )
812
+ else :
813
+ results = migrate_all_configs (
814
+ paths_to_process ,
815
+ create_backups = create_backups ,
816
+ force = args .force ,
817
+ )
818
+
819
+ if not results :
820
+ print ("No configuration files found" )
821
+ return 0
822
+
823
+ # Print results
824
+ print (f"Processed { len (results )} configuration file(s):" )
825
+ for file_path , success , message in results :
826
+ status = format_status (success )
827
+ print (f"{ status } : { file_path } - { message } " )
828
+
829
+ return 0
830
+ except Exception as e :
831
+ logger .exception (f"Error processing configuration files" )
832
+ print (f"Error: { e } " )
833
+ return 1
0 commit comments