Add centralized path management and sample database generator

Co-authored-by: elisiariocouto <818914+elisiariocouto@users.noreply.github.com>

leggen/commands/generate_sample_db.py

@@ -0,0 +1,65 @@
+"""Generate sample database command."""
+
+import click
+from pathlib import Path
+
+from leggen.utils.paths import path_manager
+
+
+@click.command()
+@click.option(
+    "--database",
+    type=click.Path(path_type=Path),
+    help="Path to database file (default: uses LEGGEN_DATABASE_PATH or ~/.config/leggen/leggen-dev.db)",
+)
+@click.option(
+    "--accounts",
+    type=int,
+    default=3,
+    help="Number of sample accounts to generate (default: 3)",
+)
+@click.option(
+    "--transactions",
+    type=int,
+    default=50,
+    help="Number of transactions per account (default: 50)",
+)
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Overwrite existing database without confirmation",
+)
+@click.pass_context
+def generate_sample_db(ctx: click.Context, database: Path, accounts: int, transactions: int, force: bool):
+    """Generate a sample database with realistic financial data for testing."""
+    
+    # Import here to avoid circular imports
+    import sys
+    import subprocess
+    from pathlib import Path as PathlibPath
+    
+    # Get the script path
+    script_path = PathlibPath(__file__).parent.parent.parent / "scripts" / "generate_sample_db.py"
+    
+    # Build command arguments
+    cmd = [sys.executable, str(script_path)]
+    
+    if database:
+        cmd.extend(["--database", str(database)])
+    
+    cmd.extend(["--accounts", str(accounts)])
+    cmd.extend(["--transactions", str(transactions)])
+    
+    if force:
+        cmd.append("--force")
+    
+    # Execute the script
+    try:
+        subprocess.run(cmd, check=True)
+    except subprocess.CalledProcessError as e:
+        click.echo(f"Error generating sample database: {e}")
+        ctx.exit(1)
+
+
+# Export the command
+generate_sample_db = generate_sample_db

leggen/database/sqlite.py

@@ -5,14 +5,13 @@ from sqlite3 import IntegrityError
 import click
 
 from leggen.utils.text import success, warning
+from leggen.utils.paths import path_manager
 
 
 def persist_balances(ctx: click.Context, balance: dict):
     # Connect to SQLite database
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
-    db_path.parent.mkdir(parents=True, exist_ok=True)
+    db_path = path_manager.get_database_path()
+    path_manager.ensure_database_dir_exists()
     conn = sqlite3.connect(str(db_path))
     cursor = conn.cursor()
 
@@ -108,10 +107,8 @@ def persist_balances(ctx: click.Context, balance: dict):
 
 def persist_transactions(ctx: click.Context, account: str, transactions: list) -> list:
     # Connect to SQLite database
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
-    db_path.parent.mkdir(parents=True, exist_ok=True)
+    db_path = path_manager.get_database_path()
+    path_manager.ensure_database_dir_exists()
     conn = sqlite3.connect(str(db_path))
     cursor = conn.cursor()
 
@@ -216,9 +213,7 @@ def get_transactions(
     search=None,
 ):
     """Get transactions from SQLite database with optional filtering"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return []
     conn = sqlite3.connect(str(db_path))
@@ -288,9 +283,7 @@ def get_transactions(
 
 def get_balances(account_id=None):
     """Get latest balances from SQLite database"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return []
     conn = sqlite3.connect(str(db_path))
@@ -329,9 +322,7 @@ def get_balances(account_id=None):
 
 def get_account_summary(account_id):
     """Get basic account info from transactions table (avoids GoCardless API call)"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return None
     conn = sqlite3.connect(str(db_path))
@@ -365,9 +356,7 @@ def get_account_summary(account_id):
 
 def get_transaction_count(account_id=None, **filters):
     """Get total count of transactions matching filters"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return 0
     conn = sqlite3.connect(str(db_path))
@@ -414,10 +403,8 @@ def get_transaction_count(account_id=None, **filters):
 
 def persist_account(account_data: dict):
     """Persist account details to SQLite database"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
-    db_path.parent.mkdir(parents=True, exist_ok=True)
+    db_path = path_manager.get_database_path()
+    path_manager.ensure_database_dir_exists()
     conn = sqlite3.connect(str(db_path))
     cursor = conn.cursor()
 
@@ -485,9 +472,7 @@ def persist_account(account_data: dict):
 
 def get_accounts(account_ids=None):
     """Get account details from SQLite database"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return []
     conn = sqlite3.connect(str(db_path))
@@ -519,9 +504,7 @@ def get_accounts(account_ids=None):
 
 def get_account(account_id: str):
     """Get specific account details from SQLite database"""
-    from pathlib import Path
-
-    db_path = Path.home() / ".config" / "leggen" / "leggen.db"
+    db_path = path_manager.get_database_path()
     if not db_path.exists():
         return None
     conn = sqlite3.connect(str(db_path))

leggen/main.py

@@ -7,6 +7,7 @@ import click
 
 from leggen.utils.config import load_config
 from leggen.utils.text import error
+from leggen.utils.paths import path_manager
 
 cmd_folder = os.path.abspath(os.path.join(os.path.dirname(__file__), "commands"))
 
@@ -77,7 +78,7 @@ class Group(click.Group):
     "-c",
     "--config",
     type=click.Path(dir_okay=False),
-    default=Path.home() / ".config" / "leggen" / "config.toml",
+    default=lambda: str(path_manager.get_config_file_path()),
     show_default=True,
     callback=load_config,
     is_eager=True,
@@ -86,6 +87,20 @@ class Group(click.Group):
     show_envvar=True,
     help="Path to TOML configuration file",
 )
+@click.option(
+    "--config-dir",
+    type=click.Path(exists=False, file_okay=False, path_type=Path),
+    envvar="LEGGEN_CONFIG_DIR",
+    show_envvar=True,
+    help="Directory containing configuration files (default: ~/.config/leggen)",
+)
+@click.option(
+    "--database",
+    type=click.Path(dir_okay=False, path_type=Path),
+    envvar="LEGGEN_DATABASE_PATH",
+    show_envvar=True,
+    help="Path to SQLite database file (default: <config-dir>/leggen.db)",
+)
 @click.option(
     "--api-url",
     type=str,
@@ -100,7 +115,7 @@ class Group(click.Group):
 )
 @click.version_option(package_name="leggen")
 @click.pass_context
-def cli(ctx: click.Context, api_url: str):
+def cli(ctx: click.Context, config_dir: Path, database: Path, api_url: str):
     """
     Leggen: An Open Banking CLI
     """
@@ -109,5 +124,11 @@ def cli(ctx: click.Context, api_url: str):
     if "--help" in sys.argv[1:] or "-h" in sys.argv[1:]:
         return
 
+    # Set up path manager with user-provided paths
+    if config_dir:
+        path_manager.set_config_dir(config_dir)
+    if database:
+        path_manager.set_database_path(database)
+
     # Store API URL in context for commands to use
     ctx.obj["api_url"] = api_url

leggen/utils/paths.py

@@ -0,0 +1,67 @@
+"""Centralized path management for Leggen."""
+
+import os
+from pathlib import Path
+from typing import Optional
+
+
+class PathManager:
+    """Manages configurable paths for config and database files."""
+    
+    def __init__(self):
+        self._config_dir: Optional[Path] = None
+        self._database_path: Optional[Path] = None
+    
+    def get_config_dir(self) -> Path:
+        """Get the configuration directory."""
+        if self._config_dir is not None:
+            return self._config_dir
+        
+        # Check environment variable first
+        config_dir = os.environ.get("LEGGEN_CONFIG_DIR")
+        if config_dir:
+            return Path(config_dir)
+        
+        # Default to ~/.config/leggen
+        return Path.home() / ".config" / "leggen"
+    
+    def set_config_dir(self, path: Path) -> None:
+        """Set the configuration directory."""
+        self._config_dir = Path(path)
+    
+    def get_config_file_path(self) -> Path:
+        """Get the configuration file path."""
+        return self.get_config_dir() / "config.toml"
+    
+    def get_database_path(self) -> Path:
+        """Get the database file path."""
+        if self._database_path is not None:
+            return self._database_path
+        
+        # Check environment variable first
+        database_path = os.environ.get("LEGGEN_DATABASE_PATH")
+        if database_path:
+            return Path(database_path)
+        
+        # Default to config_dir/leggen.db
+        return self.get_config_dir() / "leggen.db"
+    
+    def set_database_path(self, path: Path) -> None:
+        """Set the database file path."""
+        self._database_path = Path(path)
+    
+    def get_auth_file_path(self) -> Path:
+        """Get the authentication file path."""
+        return self.get_config_dir() / "auth.json"
+    
+    def ensure_config_dir_exists(self) -> None:
+        """Ensure the configuration directory exists."""
+        self.get_config_dir().mkdir(parents=True, exist_ok=True)
+    
+    def ensure_database_dir_exists(self) -> None:
+        """Ensure the database directory exists."""
+        self.get_database_path().parent.mkdir(parents=True, exist_ok=True)
+
+
+# Global instance for the application
+path_manager = PathManager()