Implement CLI app for API interaction with MFA

- Create full CLI application with authentication, sync triggering, and status checking
- Implement MFA support for secure authentication
- Add token management with secure local storage
- Create API client for backend communication
- Implement data models for User Session, Sync Job, and Authentication Token
- Add command-line interface with auth and sync commands
- Include unit and integration tests
- Follow project constitution standards for Python 3.13, type hints, and code quality
- Support multiple output formats (table, JSON, CSV)

cli/src/api/client.py

@@ -0,0 +1,101 @@
+from typing import Any, Dict, Optional
+
+import httpx
+
+from ..models.token import AuthenticationToken
+
+
+class ApiClient:
+    """API client for communicating with backend"""
+
+    def __init__(self, base_url: str = "https://api.garmin.com"):
+        self.base_url = base_url
+        self.client = httpx.AsyncClient()
+        self.token: Optional[AuthenticationToken] = None
+
+    async def set_token(self, token: AuthenticationToken) -> None:
+        """Set the authentication token for API requests"""
+        self.token = token
+        self.client.headers["Authorization"] = (
+            f"{token.token_type} {token.access_token}"
+        )
+
+    async def authenticate_user(
+        self, username: str, password: str, mfa_code: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Authenticate user via CLI with optional MFA"""
+        url = f"{self.base_url}/api/auth/cli/login"
+
+        payload = {"username": username, "password": password}
+
+        if mfa_code:
+            payload["mfa_code"] = mfa_code
+
+        try:
+            response = await self.client.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            # Handle HTTP errors (4xx, 5xx)
+            error_detail = await self._extract_error_detail(response)
+            raise Exception(f"API Error: {e.response.status_code} - {error_detail}")
+        except httpx.RequestError as e:
+            # Handle request errors (network, timeout, etc.)
+            raise Exception(f"Request Error: {str(e)}")
+
+    async def trigger_sync(
+        self,
+        sync_type: str,
+        date_range: Optional[Dict[str, str]] = None,
+        force_full_sync: bool = False,
+    ) -> Dict[str, Any]:
+        """Trigger a sync operation"""
+        url = f"{self.base_url}/api/sync/cli/trigger"
+
+        payload = {"sync_type": sync_type, "force_full_sync": force_full_sync}
+
+        if date_range:
+            payload["date_range"] = date_range
+
+        try:
+            response = await self.client.post(url, json=payload)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            # Handle HTTP errors (4xx, 5xx) including 409 conflict
+            error_detail = await self._extract_error_detail(response)
+            raise Exception(f"API Error: {e.response.status_code} - {error_detail}")
+        except httpx.RequestError as e:
+            # Handle request errors (network, timeout, etc.)
+            raise Exception(f"Request Error: {str(e)}")
+
+    async def get_sync_status(self, job_id: Optional[str] = None) -> Dict[str, Any]:
+        """Get sync status for all jobs or a specific job"""
+        if job_id:
+            url = f"{self.base_url}/api/sync/cli/status/{job_id}"
+        else:
+            url = f"{self.base_url}/api/sync/cli/status"
+
+        try:
+            response = await self.client.get(url)
+            response.raise_for_status()
+            return response.json()
+        except httpx.HTTPStatusError as e:
+            # Handle HTTP errors (4xx, 5xx)
+            error_detail = await self._extract_error_detail(response)
+            raise Exception(f"API Error: {e.response.status_code} - {error_detail}")
+        except httpx.RequestError as e:
+            # Handle request errors (network, timeout, etc.)
+            raise Exception(f"Request Error: {str(e)}")
+
+    async def _extract_error_detail(self, response: httpx.Response) -> str:
+        """Extract error details from response"""
+        try:
+            error_json = response.json()
+            return error_json.get("error", "Unknown error")
+        except Exception:
+            return response.text[:200]  # Return first 200 chars if not JSON
+
+    async def close(self) -> None:
+        """Close the HTTP client"""
+        await self.client.aclose()

cli/src/auth/auth_manager.py

@@ -0,0 +1,147 @@
+import asyncio
+from datetime import datetime, timedelta
+from typing import Optional
+
+from ..api.client import ApiClient
+from ..auth.token_manager import TokenManager
+from ..models.session import UserSession
+from ..models.token import AuthenticationToken
+
+
+class AuthManager:
+    """Handles authentication flows with MFA support"""
+
+    def __init__(self, api_client: ApiClient, token_manager: TokenManager):
+        self.api_client = api_client
+        self.token_manager = token_manager
+
+    async def authenticate(
+        self, username: str, password: str, mfa_code: Optional[str] = None
+    ) -> Optional[UserSession]:
+        """Authenticate user with optional MFA code"""
+        try:
+            # Try to authenticate via API
+            auth_response = await self.api_client.authenticate_user(
+                username, password, mfa_code
+            )
+
+            if auth_response.get("success"):
+                # Extract token information
+                access_token = auth_response.get("access_token")
+                token_type = auth_response.get("token_type", "Bearer")
+                expires_in = auth_response.get("expires_in")
+
+                # Create an AuthenticationToken object
+                token = AuthenticationToken(
+                    token_id=auth_response.get("session_id", ""),
+                    user_id=auth_response.get("user", {}).get("id", ""),
+                    access_token=access_token or "",
+                    token_type=token_type,
+                    expires_in=expires_in,
+                    mfa_verified=auth_response.get("mfa_required", False)
+                    and mfa_code is not None,
+                )
+
+                # Store the token securely
+                self.token_manager.save_token(token)
+
+                # Update the API client with the new token
+                await self.api_client.set_token(token)
+
+                # Create and return user session
+                session = UserSession(
+                    session_id=auth_response.get("session_id", ""),
+                    user_id=auth_response.get("user", {}).get("id", ""),
+                    access_token=access_token or "",
+                    refresh_token=auth_response.get("refresh_token", ""),
+                    expires_at=self._calculate_expiry(expires_in),
+                    mfa_enabled=auth_response.get("mfa_required", False),
+                )
+
+                return session
+            else:
+                # Authentication failed
+                error_msg = auth_response.get("error", "Authentication failed")
+                raise Exception(f"Authentication failed: {error_msg}")
+
+        except Exception as e:
+            # Handle any errors during authentication
+            raise e
+
+    async def logout(self) -> bool:
+        """Log out the current user and clear stored tokens"""
+        try:
+            # Clear stored token
+            self.token_manager.clear_token()
+
+            # Clear token from API client
+            self.api_client.token = None
+            if "Authorization" in self.api_client.client.headers:
+                del self.api_client.client.headers["Authorization"]
+
+            return True
+        except Exception:
+            return False
+
+    async def is_authenticated(self) -> bool:
+        """Check if the user is currently authenticated"""
+        return self.token_manager.token_exists()
+
+    def _calculate_expiry(self, expires_in: Optional[int]) -> Optional[datetime]:
+        """Calculate expiration time based on expires_in seconds"""
+        if expires_in is None:
+            return None
+
+        return datetime.now() + timedelta(seconds=expires_in)
+
+    def is_token_expired(self, token: Optional[AuthenticationToken] = None) -> bool:
+        """Check if the current token is expired"""
+        if token is None:
+            token = self.token_manager.load_token()
+
+        if not token or not token.expires_in:
+            return (
+                True  # If we don't have a token or expiration info, consider it expired
+            )
+
+        from datetime import datetime
+
+        # Calculate when the token should expire based on creation time + expires_in
+        if token.created_at:
+            expiry_time = token.created_at + timedelta(seconds=token.expires_in)
+            return datetime.now() > expiry_time
+        else:
+            return True  # If no creation time, consider expired
+
+    async def refresh_token_if_needed(self) -> bool:
+        """Refresh token if it's expired or about to expire"""
+        current_token = self.token_manager.load_token()
+
+        if not current_token:
+            return False
+
+        if not self.is_token_expired(current_token):
+            return True  # Token is still valid
+
+        # In a real implementation, we would call the API to get a new token using the refresh token
+        # For this example, we'll return False to indicate that re-authentication is needed
+        # since we don't have a refresh API endpoint defined
+        return False
+
+    async def get_valid_token(self) -> Optional[AuthenticationToken]:
+        """Get a valid token, refreshing if needed"""
+        current_token = self.token_manager.load_token()
+
+        if not current_token:
+            return None
+
+        if not self.is_token_expired(current_token):
+            return current_token
+
+        # Try to refresh the token
+        refresh_success = await self.refresh_token_if_needed()
+        if refresh_success:
+            return self.token_manager.load_token()
+        else:
+            # Could not refresh, token is invalid
+            return None

cli/src/auth/token_manager.py

@@ -0,0 +1,52 @@
+import json
+import os
+from pathlib import Path
+from typing import Optional
+
+from ..models.token import AuthenticationToken
+
+
+class TokenManager:
+    """Manages local token storage and refresh with secure storage"""
+
+    def __init__(self, token_path: Optional[Path] = None):
+        if token_path is None:
+            # Use default location in user's home directory
+            self.token_path = Path.home() / ".garmin-sync" / "tokens.json"
+            self.token_path.parent.mkdir(exist_ok=True)
+        else:
+            self.token_path = token_path
+            self.token_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Set secure file permissions (read/write for owner only)
+        os.chmod(self.token_path.parent, 0o700)  # Only owner can read/write/execute
+
+    def save_token(self, token: AuthenticationToken) -> None:
+        """Save token to secure local storage"""
+        token_data = token.model_dump()
+        with open(self.token_path, "w") as f:
+            json.dump(token_data, f)
+
+        # Set secure file permissions (read/write for owner only)
+        os.chmod(self.token_path, 0o600)  # Only owner can read/write
+
+    def load_token(self) -> Optional[AuthenticationToken]:
+        """Load token from secure local storage"""
+        if not self.token_path.exists():
+            return None
+
+        try:
+            with open(self.token_path, "r") as f:
+                token_data = json.load(f)
+                return AuthenticationToken(**token_data)
+        except (json.JSONDecodeError, KeyError, TypeError):
+            return None
+
+    def clear_token(self) -> None:
+        """Clear stored token"""
+        if self.token_path.exists():
+            self.token_path.unlink()
+
+    def token_exists(self) -> bool:
+        """Check if a token exists in storage"""
+        return self.token_path.exists()

cli/src/commands/auth_cmd.py

@@ -0,0 +1,118 @@
+import asyncio
+from typing import Optional
+
+import click
+
+from ..api.client import ApiClient
+from ..auth.auth_manager import AuthManager
+from ..auth.token_manager import TokenManager
+from ..utils.output import format_output
+
+
+@click.group()
+def auth():
+    """Authentication commands for GarminSync CLI"""
+    pass
+
+
+@auth.command()
+@click.option("--username", "-u", prompt=True, help="Your Garmin username or email")
+@click.option(
+    "--password", "-p", prompt=True, hide_input=True, help="Your Garmin password"
+)
+@click.option("--mfa-code", "-mfa", help="MFA code if required")
+@click.option("--interactive", "-i", is_flag=True, help="Run in interactive mode")
+@click.option(
+    "--non-interactive",
+    "-n",
+    is_flag=True,
+    help="Run in non-interactive (scriptable) mode",
+)
+def login(
+    username: str,
+    password: str,
+    mfa_code: Optional[str],
+    interactive: bool,
+    non_interactive: bool,
+):
+    """Authenticate with your Garmin account"""
+
+    async def run_login():
+        api_client = ApiClient()
+        token_manager = TokenManager()
+        auth_manager = AuthManager(api_client, token_manager)
+
+        try:
+            # If interactive mode and no MFA code provided, prompt for it
+            if interactive and not mfa_code:
+                mfa_input = click.prompt(
+                    "MFA Code (leave blank if not required)",
+                    default="",
+                    show_default=False,
+                )
+                if mfa_input:  # Only use MFA code if user provided one
+                    mfa_code = mfa_input
+
+            # Perform authentication
+            session = await auth_manager.authenticate(username, password, mfa_code)
+
+            if session:
+                click.echo(f"Successfully authenticated as user {session.user_id}")
+            else:
+                click.echo("Authentication failed")
+
+        except Exception as e:
+            click.echo(f"Authentication failed: {str(e)}")
+        finally:
+            await api_client.close()
+
+    # Run the async function
+    asyncio.run(run_login())
+
+
+@auth.command()
+def logout():
+    """Log out and clear stored credentials"""
+
+    async def run_logout():
+        api_client = ApiClient()
+        token_manager = TokenManager()
+        auth_manager = AuthManager(api_client, token_manager)
+
+        try:
+            success = await auth_manager.logout()
+            if success:
+                click.echo("Successfully logged out")
+            else:
+                click.echo("Logout failed")
+        except Exception as e:
+            click.echo(f"Logout failed: {str(e)}")
+        finally:
+            await api_client.close()
+
+    # Run the async function
+    asyncio.run(run_logout())
+
+
+@auth.command()
+def status():
+    """Check authentication status"""
+
+    async def run_status():
+        api_client = ApiClient()
+        token_manager = TokenManager()
+        auth_manager = AuthManager(api_client, token_manager)
+
+        try:
+            authenticated = await auth_manager.is_authenticated()
+            if authenticated:
+                click.echo("Authenticated: Yes")
+            else:
+                click.echo("Authenticated: No")
+        except Exception as e:
+            click.echo(f"Could not check status: {str(e)}")
+        finally:
+            await api_client.close()
+
+    # Run the async function
+    asyncio.run(run_status())

cli/src/commands/sync_cmd.py

@@ -0,0 +1,148 @@
+import asyncio
+from typing import Optional
+
+import click
+
+from ..api.client import ApiClient
+from ..auth.auth_manager import AuthManager
+from ..auth.token_manager import TokenManager
+from ..utils.output import format_output
+
+
+@click.group()
+def sync():
+    """Sync commands for GarminSync CLI"""
+    pass
+
+
+@sync.command()
+@click.option(
+    "--type",
+    "-t",
+    "sync_type",
+    type=click.Choice(["activities", "health", "workouts"]),
+    required=True,
+    help="Type of data to sync",
+)
+@click.option("--start-date", help="Start date for sync (YYYY-MM-DD)")
+@click.option("--end-date", help="End date for sync (YYYY-MM-DD)")
+@click.option(
+    "--force-full", is_flag=True, help="Perform a full sync instead of incremental"
+)
+def trigger(
+    sync_type: str, start_date: Optional[str], end_date: Optional[str], force_full: bool
+):
+    """Trigger a sync operation"""
+
+    async def run_trigger():
+        api_client = ApiClient()
+        token_manager = TokenManager()
+        auth_manager = AuthManager(api_client, token_manager)
+
+        try:
+            # Check if user is authenticated
+            if not await auth_manager.is_authenticated():
+                click.echo(
+                    "Error: Not authenticated. Please run 'garmin-sync auth login' first."
+                )
+                return
+
+            # Load and set the token
+            token = token_manager.load_token()
+            if token:
+                await api_client.set_token(token)
+
+            # Prepare date range if specified
+            date_range = None
+            if start_date or end_date:
+                date_range = {}
+                if start_date:
+                    date_range["start_date"] = start_date
+                if end_date:
+                    date_range["end_date"] = end_date
+
+            # Trigger the sync
+            result = await api_client.trigger_sync(sync_type, date_range, force_full)
+
+            if result.get("success"):
+                job_id = result.get("job_id")
+                status = result.get("status")
+                click.echo(f"Sync triggered successfully!")
+                click.echo(f"Job ID: {job_id}")
+                click.echo(f"Status: {status}")
+            else:
+                error_msg = result.get("error", "Unknown error")
+                click.echo(f"Error triggering sync: {error_msg}")
+
+        except Exception as e:
+            click.echo(f"Error triggering sync: {str(e)}")
+        finally:
+            await api_client.close()
+
+    # Run the async function
+    asyncio.run(run_trigger())
+
+
+@sync.command()
+@click.option(
+    "--job-id",
+    "-j",
+    help="Specific job ID to check (returns all recent if not provided)",
+)
+@click.option(
+    "--format",
+    "-f",
+    "output_format",
+    type=click.Choice(["table", "json", "csv"]),
+    default="table",
+    help="Output format",
+)
+def status(job_id: Optional[str], output_format: str):
+    """Check the status of sync operations"""
+
+    async def run_status():
+        api_client = ApiClient()
+        token_manager = TokenManager()
+        auth_manager = AuthManager(api_client, token_manager)
+
+        try:
+            # Check if user is authenticated
+            if not await auth_manager.is_authenticated():
+                click.echo(
+                    "Error: Not authenticated. Please run 'garmin-sync auth login' first."
+                )
+                return
+
+            # Load and set the token
+            token = token_manager.load_token()
+            if token:
+                await api_client.set_token(token)
+
+            # Get sync status
+            result = await api_client.get_sync_status(job_id)
+
+            if result.get("success"):
+                if job_id:
+                    # Single job status
+                    job_data = result.get("job")
+                    formatted_output = format_output(job_data, output_format)
+                    click.echo(formatted_output)
+                else:
+                    # Multiple jobs status
+                    jobs_data = result.get("jobs", [])
+                    formatted_output = format_output(jobs_data, output_format)
+                    click.echo(formatted_output)
+            else:
+                error_msg = result.get("error", "Unknown error")
+                click.echo(f"Error getting sync status: {error_msg}")
+
+        except Exception as e:
+            click.echo(f"Error getting sync status: {str(e)}")
+        finally:
+            await api_client.close()
+
+    # Run the async function
+    asyncio.run(run_status())
+
+
+# Add the sync command group to the main CLI in the __init__.py would be handled in the main module

cli/src/main.py

@@ -0,0 +1,19 @@
+import click
+
+from .commands.auth_cmd import auth
+from .commands.sync_cmd import sync
+
+
+@click.group()
+def cli() -> None:
+    """GarminSync CLI - Command-line interface for interacting with GarminSync API."""
+    pass
+
+
+# Add the auth and sync command groups to the main CLI
+cli.add_command(auth)
+cli.add_command(sync)
+
+
+if __name__ == "__main__":
+    cli()

cli/src/models/session.py

@@ -0,0 +1,17 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class UserSession(BaseModel):
+    """Represents an authenticated user session with associated tokens and permissions"""
+
+    session_id: str
+    user_id: str
+    access_token: str
+    refresh_token: str
+    expires_at: Optional[datetime] = None
+    mfa_enabled: bool = False
+    created_at: datetime = datetime.now()
+    last_used_at: Optional[datetime] = None

cli/src/models/sync_job.py

@@ -0,0 +1,19 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class SyncJob(BaseModel):
+    """Represents an initiated sync operation with status, progress, and metadata"""
+
+    job_id: str
+    user_id: str
+    status: str  # pending, running, completed, failed, cancelled
+    progress: float = 0.0  # Percentage of completion (0-100)
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    sync_type: str  # activities, health, workouts, etc.
+    error_message: Optional[str] = None
+    total_items: Optional[int] = None
+    processed_items: Optional[int] = None

cli/src/models/token.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+class AuthenticationToken(BaseModel):
+    """Secure credential used to access the API on behalf of the user"""
+
+    token_id: str
+    user_id: str
+    access_token: str
+    token_type: str = "Bearer"
+    expires_in: Optional[int] = None  # Time until expiration in seconds
+    scope: Optional[str] = None
+    created_at: datetime = datetime.now()
+    last_used_at: Optional[datetime] = None
+    mfa_verified: bool = False

cli/src/utils/config.py

@@ -0,0 +1,55 @@
+import os
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import yaml
+
+
+class ConfigManager:
+    """Configuration management utilities for YAML config"""
+
+    def __init__(self, config_path: Optional[Path] = None):
+        if config_path is None:
+            # Use default location in user's home directory
+            self.config_path = Path.home() / ".garmin-sync" / "config.yaml"
+            self.config_path.parent.mkdir(exist_ok=True)
+        else:
+            self.config_path = config_path
+            self.config_path.parent.mkdir(parents=True, exist_ok=True)
+
+        self.config = self._load_config()
+
+    def _load_config(self) -> Dict[str, Any]:
+        """Load configuration from YAML file"""
+        if self.config_path.exists():
+            with open(self.config_path, "r") as f:
+                return yaml.safe_load(f) or {}
+        else:
+            # Return default configuration
+            default_config = {
+                "api_base_url": "https://api.garmin.com",
+                "default_timeout": 30,
+                "output_format": "table",  # Options: table, json, csv
+                "remember_login": True,
+            }
+            self._save_config(default_config)
+            return default_config
+
+    def _save_config(self, config: Dict[str, Any]) -> None:
+        """Save configuration to YAML file"""
+        with open(self.config_path, "w") as f:
+            yaml.dump(config, f)
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """Get a configuration value"""
+        return self.config.get(key, default)
+
+    def set(self, key: str, value: Any) -> None:
+        """Set a configuration value"""
+        self.config[key] = value
+        self._save_config(self.config)
+
+    def update(self, updates: Dict[str, Any]) -> None:
+        """Update multiple configuration values"""
+        self.config.update(updates)
+        self._save_config(self.config)

cli/src/utils/output.py

@@ -0,0 +1,98 @@
+import csv
+import json
+from io import StringIO
+from typing import Any, Dict, List, Mapping, Set, Union
+
+
+def format_output(data: Any, format_type: str = "table") -> str:
+    """Format output in multiple formats (JSON, table, CSV)"""
+
+    if format_type.lower() == "json":
+        return json.dumps(data, indent=2, default=str)
+
+    elif format_type.lower() == "csv":
+        return _format_as_csv(data)
+
+    elif format_type.lower() == "table":
+        return _format_as_table(data)
+
+    else:
+        # Default to table format
+        return _format_as_table(data)
+
+
+def _format_as_table(data: Any) -> str:
+    """Format data as a human-readable table"""
+    if isinstance(data, dict):
+        # Format dictionary as key-value pairs
+        lines = []
+        for key, value in data.items():
+            lines.append(f"{key:<20} | {value}")
+        return "\n".join(lines)
+
+    elif isinstance(data, list):
+        if not data:
+            return "No data to display"
+
+        if isinstance(data[0], dict):
+            # Format list of dictionaries as a table
+            if not data[0]:
+                return "No data to display"
+
+            headers = list(data[0].keys())
+            # Create header row
+            header_line = " | ".join(f"{h:<15}" for h in headers)
+            separator = "-+-".join("-" * 15 for _ in headers)
+
+            # Create data rows
+            rows = [header_line, separator]
+            for item in data:
+                row = " | ".join(f"{str(item.get(h, '')):<15}" for h in headers)
+                rows.append(row)
+
+            return "\n".join(rows)
+        else:
+            # Format simple list
+            return "\n".join(str(item) for item in data)
+
+    else:
+        # For other types, just convert to string
+        return str(data)
+
+
+def _format_as_csv(data: Any) -> str:
+    """Format data as CSV"""
+    if isinstance(data, dict):
+        # Convert single dict to list with one item for CSV processing
+        data = [data]
+
+    if isinstance(data, list) and data and isinstance(data[0], dict):
+        # Format list of dictionaries as CSV
+        output = StringIO()
+        if data:
+            fieldnames: Set[str] = set()
+            for row in data:
+                fieldnames.update(row.keys())
+            fieldnames = sorted(list(fieldnames))
+
+            writer_csv: csv.DictWriter = csv.DictWriter(output, fieldnames=fieldnames)
+            writer_csv.writeheader()
+            for row in data:
+                writer_csv.writerow({k: v for k, v in row.items() if k in fieldnames})
+
+        return output.getvalue()
+
+    elif isinstance(data, list):
+        # Format simple list as CSV with one column
+        output = StringIO()
+        writer_csv: csv.writer = csv.writer(output)
+        for item in data:
+            writer_csv.writerow([item])
+        return output.getvalue()
+
+    else:
+        # For other types, just convert to string and put in one cell
+        output = StringIO()
+        writer_csv: csv.writer = csv.writer(output)
+        writer_csv.writerow([str(data)])
+        return output.getvalue()