MCPcopy Create free account
hub / github.com/AutoForgeAI/autoforge / migrate_json_to_sqlite

Function migrate_json_to_sqlite

api/migration.py:19–115  ·  view source on GitHub ↗

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,
)

Source from the content-addressed store, hash-verified

17
18
19def 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

Callers 1

server_lifespanFunction · 0.90

Calls 2

FeatureClass · 0.90
closeMethod · 0.45

Tested by

no test coverage detected