Detect existing feature_list.json, import to SQLite, rename to backup. This function: 1. Checks if feature_list.json exists 2. Checks if database already has data (skips if so) 3. Imports all features from JSON 4. Renames JSON file to feature_list.json.backup.
(
project_dir: Path,
session_maker: sessionmaker,
)
| 17 | |
| 18 | |
| 19 | def migrate_json_to_sqlite( |
| 20 | project_dir: Path, |
| 21 | session_maker: sessionmaker, |
| 22 | ) -> bool: |
| 23 | """ |
| 24 | Detect existing feature_list.json, import to SQLite, rename to backup. |
| 25 | |
| 26 | This function: |
| 27 | 1. Checks if feature_list.json exists |
| 28 | 2. Checks if database already has data (skips if so) |
| 29 | 3. Imports all features from JSON |
| 30 | 4. Renames JSON file to feature_list.json.backup.<timestamp> |
| 31 | |
| 32 | Args: |
| 33 | project_dir: Directory containing the project |
| 34 | session_maker: SQLAlchemy session maker |
| 35 | |
| 36 | Returns: |
| 37 | True if migration was performed, False if skipped |
| 38 | """ |
| 39 | json_file = project_dir / "feature_list.json" |
| 40 | |
| 41 | if not json_file.exists(): |
| 42 | return False # No JSON file to migrate |
| 43 | |
| 44 | # Check if database already has data |
| 45 | session: Session = session_maker() |
| 46 | try: |
| 47 | existing_count = session.query(Feature).count() |
| 48 | if existing_count > 0: |
| 49 | print( |
| 50 | f"Database already has {existing_count} features, skipping migration" |
| 51 | ) |
| 52 | return False |
| 53 | finally: |
| 54 | session.close() |
| 55 | |
| 56 | # Load JSON data |
| 57 | try: |
| 58 | with open(json_file, "r", encoding="utf-8") as f: |
| 59 | features_data = json.load(f) |
| 60 | except json.JSONDecodeError as e: |
| 61 | print(f"Error parsing feature_list.json: {e}") |
| 62 | return False |
| 63 | except IOError as e: |
| 64 | print(f"Error reading feature_list.json: {e}") |
| 65 | return False |
| 66 | |
| 67 | if not isinstance(features_data, list): |
| 68 | print("Error: feature_list.json must contain a JSON array") |
| 69 | return False |
| 70 | |
| 71 | # Import features into database |
| 72 | session = session_maker() |
| 73 | try: |
| 74 | imported_count = 0 |
| 75 | for i, feature_dict in enumerate(features_data): |
| 76 | # Handle both old format (no id/priority/name) and new format |
no test coverage detected