Skip to content

Support syncing by label #19

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ithinkihaveacat wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Support syncing by label #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
df473b8
Add sync by label
ithinkihaveacat Jun 14, 2025
3f2359b
Merge branch 'main' into sync-by-label
ithinkihaveacat Jun 15, 2025
0aed7a8
Add --label info to README
ithinkihaveacat Jun 15, 2025
bbbb761
Replace --label filtering with more generic --query
ithinkihaveacat Jun 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ A robust Python application that syncs Gmail messages to a local SQLite database
- **Robust Error Handling**: Automatic retries with exponential backoff
- **Graceful Shutdown**: Handles interruption signals cleanly
- **Type Safety**: Comprehensive type hints throughout the codebase
- **Label Filtering**: Sync only messages with specific Gmail labels

## Installation

Expand Down Expand Up @@ -56,12 +57,18 @@ python main.py sync --data-dir ./data
# Full sync with deletion detection
python main.py sync --data-dir ./data --full-sync

# Sync only messages with a specific label
python main.py sync --data-dir ./data --query "label:Transactional"

# Sync a specific message
python main.py sync-message --data-dir ./data --message-id MESSAGE_ID

# Detect and mark deleted messages only
python main.py sync-deleted-messages --data-dir ./data

# Detect deleted messages for a specific label
python main.py sync-deleted-messages --data-dir ./data --query "label:Work"

# Use custom number of worker threads
python main.py sync --data-dir ./data --workers 8

Expand All @@ -76,6 +83,7 @@ python main.py sync --help
- `--data-dir`: Required. Directory where the SQLite database will be stored
- `--full-sync`: Optional. Forces a complete sync of all messages
- `--message-id`: Required for `sync-message`. The ID of a specific message to sync
- `--query`: Optional. A Gmail query string to filter messages (e.g., 'label:Transactional', 'from:user@example.com').
- `--workers`: Optional. Number of worker threads (default: number of CPU cores)
- `--help`: Show help information for commands and options

Expand Down
63 changes: 46 additions & 17 deletions gmail_to_sqlite/sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -153,15 +153,15 @@ def _create_service(credentials: Any) -> Any:

def get_message_ids_from_gmail(
service: Any,
query: Optional[List[str]] = None,
query: Optional[str] = None,
check_shutdown: Optional[Callable[[], bool]] = None,
) -> List[str]:
"""
Fetches all message IDs from Gmail matching the query.

Args:
service: The Gmail API service object.
query: Optional list of query strings to filter messages.
query: Optional query string to filter messages.
check_shutdown: Callback that returns True if shutdown is requested.

Returns:
Expand All @@ -175,6 +175,8 @@ def get_message_ids_from_gmail(
collected_count = 0

logging.info("Collecting all message IDs from Gmail...")
if query:
logging.info(f"Filtering messages with query: {query}")

try:
while not (check_shutdown and check_shutdown()):
Expand All @@ -187,7 +189,7 @@ def get_message_ids_from_gmail(
list_params["pageToken"] = page_token

if query:
list_params["q"] = " | ".join(query)
list_params["q"] = query

results = service.users().messages().list(**list_params).execute()
messages_page = results.get("messages", [])
Expand Down Expand Up @@ -282,19 +284,20 @@ def _detect_and_mark_deleted_messages(
def all_messages(
credentials: Any,
full_sync: bool = False,
num_workers: int = 4,
num_workers: int = DEFAULT_WORKERS,
check_shutdown: Optional[Callable[[], bool]] = None,
query: Optional[str] = None,
) -> int:
"""
Fetches messages from the Gmail API using the provided credentials, in parallel.
Also detects and marks deleted messages.

Args:
credentials (object): The credentials object used to authenticate the API request.
db_conn (object): The database connection object.
full_sync (bool): Whether to do a full sync or not.
credentials (object): The credentials object for API authentication.
full_sync (bool): If True, syncs all messages regardless of previous syncs.
num_workers (int): Number of worker threads for parallel fetching.
check_shutdown (callable): A callback function that returns True if shutdown is requested.
check_shutdown (callable): A callback that returns True if shutdown is requested.
query (str, optional): A Gmail query string to filter messages.

Returns:
int: The number of messages successfully synced.
Expand All @@ -303,19 +306,42 @@ def all_messages(
future_to_id = {}

try:
query = []
service = _create_service(credentials)
labels = get_labels(service)
all_message_ids = []
base_query = query if query else ""

if not full_sync:
# Fetch messages newer than the last indexed
last = db.last_indexed()
if last:
query.append(f"after:{int(last.timestamp())}")
new_query = f"{base_query} after:{int(last.timestamp())}".strip()
logging.info(f"Fetching new messages with query: {new_query}")
all_message_ids.extend(
get_message_ids_from_gmail(service, new_query, check_shutdown)
)

# Fetch messages older than the first indexed (backfill)
first = db.first_indexed()
if first:
query.append(f"before:{int(first.timestamp())}")

service = _create_service(credentials)
labels = get_labels(service)
old_query = f"{base_query} before:{int(first.timestamp())}".strip()
logging.info(f"Backfilling old messages with query: {old_query}")
all_message_ids.extend(
get_message_ids_from_gmail(service, old_query, check_shutdown)
)

all_message_ids = get_message_ids_from_gmail(service, query, check_shutdown)
# If it's the very first sync, there's no last or first, so run with the base query
if not last and not first:
logging.info(f"Performing initial sync with query: {base_query}")
all_message_ids.extend(
get_message_ids_from_gmail(service, base_query, check_shutdown)
)
else:
# Full sync requested
logging.info(f"Performing full sync with query: {base_query}")
all_message_ids = get_message_ids_from_gmail(
service, base_query, check_shutdown
)
Comment on lines +309 to +344

@marcboeker marcboeker Jun 30, 2025

Copy link
Copy Markdown
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe we can simplify this a bit. WDYT?

try:
    service = _create_service(credentials)
    labels = get_labels(service)
    base_query = query or ""

    if full_sync:
        # Full sync uses only the base query
        query_str = base_query
        logging.info(f"Performing full sync with query: {query_str}")
    else:
        # Build after/before filters into one compound query
        last = db.last_indexed()
        first = db.first_indexed()
        filters = []
        if last:
            filters.append(f"after:{int(last.timestamp())}")
        if first:
            filters.append(f"before:{int(first.timestamp())}")

        query_str = " ".join([base_query] + filters).strip()
        logging.info(f"Fetching messages with compound query: {query_str}")

    all_message_ids = get_message_ids_from_gmail(
        service, query_str, check_shutdown
    )
except Exception:
    logging.exception("Failed to sync messages")


if check_shutdown and check_shutdown():
logging.info(
Expand Down Expand Up @@ -412,7 +438,9 @@ def thread_worker(message_id: str) -> bool:


def sync_deleted_messages(
credentials: Any, check_shutdown: Optional[Callable[[], bool]] = None
credentials: Any,
check_shutdown: Optional[Callable[[], bool]] = None,
query: Optional[str] = None,
) -> None:
"""
Compares message IDs in Gmail with those in the database and marks missing messages as deleted.
Expand All @@ -422,14 +450,15 @@ def sync_deleted_messages(
Args:
credentials: The credentials used to authenticate the Gmail API.
check_shutdown (callable): A callback function that returns True if shutdown is requested.
query (str, optional): A Gmail query string to filter messages.

Returns:
int: Number of messages marked as deleted.
"""
try:
service = _create_service(credentials)
gmail_message_ids = get_message_ids_from_gmail(
service, check_shutdown=check_shutdown
service, check_shutdown=check_shutdown, query=query
)

if check_shutdown and check_shutdown():
Expand Down
9 changes: 8 additions & 1 deletion main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,8 @@ def create_argument_parser() -> argparse.ArgumentParser:
Examples:
%(prog)s sync --data-dir ./data
%(prog)s sync --data-dir ./data --full-sync
%(prog)s sync --data-dir ./data --query "label:Important"
%(prog)s sync-deleted-messages --data-dir ./data --query "label:Work"
%(prog)s sync-message --data-dir ./data --message-id abc123
""",
)
Expand All @@ -120,6 +122,10 @@ def create_argument_parser() -> argparse.ArgumentParser:
"--message-id",
help="The ID of the message to sync (required for sync-message command)",
)
parser.add_argument(
"--query",
help="A Gmail query string to filter messages (e.g., 'label:Important', 'from:user@example.com')",
)
parser.add_argument(
"--workers",
type=int,
Expand Down Expand Up @@ -164,13 +170,14 @@ def check_shutdown() -> bool:
full_sync=args.full_sync,
num_workers=args.workers,
check_shutdown=check_shutdown,
query=args.query,
)
elif args.command == "sync-message":
sync.single_message(
credentials, args.message_id, check_shutdown=check_shutdown
)
elif args.command == "sync-deleted-messages":
sync.sync_deleted_messages(credentials, check_shutdown=check_shutdown)
sync.sync_deleted_messages(credentials, check_shutdown=check_shutdown, query=args.query)

db_conn.close()
logging.info("Operation completed successfully")
Expand Down