Merge pull request #58 from ryancheley/feature/since-table-logic

Implement automatic since table logic for incremental updates

Closes #14

Author: ryancheley

src/toggl_to_sqlite/_version.py

@@ -5,7 +5,8 @@
 
 TYPE_CHECKING = False
 if TYPE_CHECKING:
-    from typing import Tuple, Union
+    from typing import Tuple
+    from typing import Union
 
     VERSION_TUPLE = Tuple[Union[int, str], ...]
 else:
@@ -16,5 +17,5 @@
 __version_tuple__: VERSION_TUPLE
 version_tuple: VERSION_TUPLE
 
-__version__ = version = "0.8.1.dev0+g7f9fa80.d20250615"
-__version_tuple__ = version_tuple = (0, 8, 1, "dev0", "g7f9fa80.d20250615")
+__version__ = version = '0.8.6.dev0+g6d85260.d20250615'
+__version_tuple__ = version_tuple = (0, 8, 6, 'dev0', 'g6d85260.d20250615')

src/toggl_to_sqlite/cli.py

@@ -54,27 +54,58 @@ def auth(auth):
     default="auth.json",
     help="Path to auth tokens, defaults to auth.json",
 )
-@click.option("-d", "--days", required=True, type=int, default=25)
-@click.option("-s", "--since", type=click.DateTime())
 @click.option(
-    "-t",
-    "--type",
-    default=["time_entries", "workspaces", "projects"],
-    required=True,
-    multiple=True,
+    "-d", "--days", type=int, default=25, help="Number of days to fetch (only used if no automatic since date is available)"
+)
+@click.option("-s", "--since", type=click.DateTime(), help="Fetch data since this date (overrides automatic since detection)")
+@click.option("--force-full", is_flag=True, help="Force a full sync, ignoring previous sync times")
+@click.option(
+    "-t", "--type", default=["time_entries", "workspaces", "projects"], required=True, multiple=True, help="Data types to fetch"
 )
-def fetch(db_path, auth, days, since, type):
+def fetch(db_path, auth, days, since, force_full, type):
     "Save Toggl data to a SQLite database"
+    import datetime
+
     auth = json.load(open(auth))
     db = sqlite_utils.Database(db_path)
-    days = days
-    since = since
+
+    # Get current time for updating sync timestamps
+    sync_time = datetime.datetime.now(datetime.timezone.utc)
+
     if "time_entries" in type:
-        time_entries = utils.get_time_entries(api_token=auth["api_token"], days=days, since=since)
+        # Use automatic since detection for time entries (unless force_full is specified)
+        if force_full:
+            click.echo("Force full sync requested - fetching all time entries")
+            effective_since = None
+        else:
+            effective_since = utils.get_effective_since_date(
+                api_token=auth["api_token"], table_name="time_entries", db=db, user_since=since
+            )
+
+        # Only use automatic since if no explicit since date is provided
+        if effective_since and not since and not force_full:
+            # Convert to date for comparison if it's a datetime
+            effective_date = effective_since.date() if hasattr(effective_since, "date") else effective_since
+            days_since_effective = (datetime.datetime.now().date() - effective_date).days + 1  # Add 1 to ensure overlap
+            click.echo(f"📅 Fetching time entries since {effective_date} ({days_since_effective} days)")
+            time_entries = utils.get_time_entries(api_token=auth["api_token"], days=days_since_effective, since=effective_since)
+        else:
+            if since:
+                since_date = since.date() if hasattr(since, "date") else since
+                click.echo(f"📅 Fetching time entries since user-specified date: {since_date}")
+            else:
+                click.echo(f"📅 Fetching time entries for the last {days} days")
+            time_entries = utils.get_time_entries(api_token=auth["api_token"], days=days, since=since)
+
         utils.save_items(time_entries, "time_entries", db)
+        utils.update_sync_time(db, "time_entries", sync_time)
+
     if "workspaces" in type:
         workspaces = utils.get_workspaces(api_token=auth["api_token"])
         utils.save_items(workspaces, "workspaces", db)
+        utils.update_sync_time(db, "workspaces", sync_time)
+
     if "projects" in type:
         projects = utils.get_projects(api_token=auth["api_token"])
         utils.save_items(projects, "projects", db)
+        utils.update_sync_time(db, "projects", sync_time)

src/toggl_to_sqlite/utils.py

@@ -75,3 +75,50 @@ def save_items(items: list, table: str, db: sqlite_utils.Database) -> None:
             db[table].insert_all(data, pk="id", alter=True, replace=True)
         except AttributeError:
             print(item)
+
+
+def get_last_sync_time(db: sqlite_utils.Database, table_name: str) -> datetime.datetime:
+    """Get the last sync time for a specific table."""
+    since_table = f"{table_name}_since"
+
+    if since_table not in db.table_names():
+        return None
+
+    try:
+        # Explicitly query for row with id=1
+        rows = list(db[since_table].rows_where("id = ?", [1]))
+        if not rows:
+            return None
+        row = rows[0]
+        return datetime.datetime.fromisoformat(row["since"])
+    except (sqlite_utils.db.NotFoundError, KeyError, IndexError, ValueError):
+        return None
+
+
+def update_sync_time(db: sqlite_utils.Database, table_name: str, sync_time: datetime.datetime) -> None:
+    """Update the last sync time for a specific table."""
+    since_table = f"{table_name}_since"
+
+    db[since_table].insert({"id": 1, "since": sync_time.isoformat()}, pk="id", replace=True, alter=True)
+
+
+def get_effective_since_date(
+    api_token: str, table_name: str, db: sqlite_utils.Database, user_since: datetime.datetime = None
+) -> datetime.datetime:
+    """Get the effective 'since' date to use for fetching data.
+
+    Priority:
+    1. User-provided since parameter
+    2. Last sync time from database
+    3. Workspace creation date (fallback)
+    """
+    if user_since:
+        return user_since
+
+    # Check for last sync time
+    last_sync = get_last_sync_time(db, table_name)
+    if last_sync:
+        return last_sync
+
+    # Fallback to workspace creation date
+    return get_start_datetime(api_token)

tests/test_cli_auth.py

@@ -0,0 +1,103 @@
+"""Tests for the CLI auth command."""
+
+import json
+import os
+import tempfile
+from unittest import mock
+
+from click.testing import CliRunner
+
+from toggl_to_sqlite.cli import cli
+
+
+def test_auth_command_creates_file():
+    """Test that auth command creates auth file with API token."""
+    runner = CliRunner()
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        auth_file = os.path.join(temp_dir, "test_auth.json")
+
+        # Mock the input function to provide a test API token
+        with mock.patch("builtins.input", return_value="test_api_token_12345"):
+            result = runner.invoke(cli, ["auth", "--auth", auth_file])
+
+            assert result.exit_code == 0
+            assert "Visit this page and sign in with your Toggl account:" in result.output
+            assert "https://track.toggl.com/profile" in result.output
+            assert f"Authentication tokens written to {auth_file}" in result.output
+
+            # Verify the file was created with correct content
+            assert os.path.exists(auth_file)
+
+            with open(auth_file, "r") as f:
+                auth_data = json.load(f)
+
+            assert auth_data == {"api_token": "test_api_token_12345"}
+
+
+def test_auth_command_default_file():
+    """Test that auth command uses default auth.json filename."""
+    runner = CliRunner()
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        # Change to temp directory so auth.json is created there
+        original_cwd = os.getcwd()
+        os.chdir(temp_dir)
+
+        try:
+            with mock.patch("builtins.input", return_value="default_token"):
+                result = runner.invoke(cli, ["auth"])
+
+                assert result.exit_code == 0
+                assert "Authentication tokens written to auth.json" in result.output
+
+                # Verify default file was created
+                assert os.path.exists("auth.json")
+
+                with open("auth.json", "r") as f:
+                    auth_data = json.load(f)
+
+                assert auth_data["api_token"] == "default_token"
+        finally:
+            os.chdir(original_cwd)
+
+
+def test_auth_command_overwrites_existing():
+    """Test that auth command overwrites existing auth file."""
+    runner = CliRunner()
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        auth_file = os.path.join(temp_dir, "existing_auth.json")
+
+        # Create existing auth file
+        with open(auth_file, "w") as f:
+            json.dump({"api_token": "old_token"}, f)
+
+        # Run auth command with new token
+        with mock.patch("builtins.input", return_value="new_token_54321"):
+            result = runner.invoke(cli, ["auth", "--auth", auth_file])
+
+            assert result.exit_code == 0
+
+            # Verify file was overwritten
+            with open(auth_file, "r") as f:
+                auth_data = json.load(f)
+
+            assert auth_data["api_token"] == "new_token_54321"
+
+
+def test_auth_command_shows_instructions():
+    """Test that auth command displays proper user instructions."""
+    runner = CliRunner()
+
+    with tempfile.TemporaryDirectory() as temp_dir:
+        auth_file = os.path.join(temp_dir, "instructions_test.json")
+
+        with mock.patch("builtins.input", return_value="instruction_token"):
+            result = runner.invoke(cli, ["auth", "--auth", auth_file])
+
+            # Check that expected instruction text is present
+            output = result.output
+            assert "Visit this page and sign in with your Toggl account:" in output
+            assert "https://track.toggl.com/profile" in output
+            assert f"Authentication tokens written to {auth_file}" in output