coulomb/issue-core
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Projects Releases Wiki Activity

fix: resolve issue-facade ID mapping bugs and enhance functionality

Browse Source 
- Fix Sentinel bug in list command where Click set search params to Sentinel.UNSET
- Fix version command by adding explicit version and package_name parameters
- Fix test isolation by correcting mock patch targets and datetime objects
- Fix critical ID mapping bug: use issue.number consistently instead of mixing with issue.backend_id
- Update all comment operations to use issue numbers instead of internal IDs
- Ensure issue-facade uses upstream issue numbers directly without local ID confusion
- Add comprehensive test coverage with 20 passing tests
- Verify core functionality: list, show, close, version, backend management all working
- Successfully close issue #166 with proper comment handling

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2025-11-10 10:48:31 +01:00
parent 00b9834d2f
commit 34a8bc7d4c
19 changed files with 469 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
issue_tracker/cli/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
Command Line Interface for Universal Issue Tracking
Provides a comprehensive CLI for managing issues across different backends.
The CLI is designed to be intuitive and follows common patterns from
tools like git, gh (GitHub CLI), and similar utilities.
Commands:
- issue list: List issues
- issue show: Show issue details
- issue create: Create new issue
- issue edit: Edit existing issue
- issue close: Close issue
- issue reopen: Reopen issue
- issue comment: Add comment
- issue label: Manage labels
- issue assign: Manage assignments
- backend: Manage backends
- sync: Synchronization operations
"""

149
issue_tracker/cli/backend_commands.py Normal file
View File

@@ -0,0 +1,149 @@
"""
Backend Management CLI Commands
Commands for configuring and managing issue tracking backends.
"""
import click
import os
from .utils import (
load_backend_configs, save_backend_configs, format_backend_list,
test_backend_connection, validate_backend_type, echo_success,
echo_error, echo_warning, echo_info, confirm_action
)
@click.group()
def backend_group():
"""Backend configuration and management."""
pass
@backend_group.command('list')
def list_backends():
"""List configured backends."""
configs = load_backend_configs()
click.echo(format_backend_list(configs))
@backend_group.command('add')
@click.argument('name')
@click.argument('backend_type', type=click.Choice(['local', 'gitea']))
@click.pass_context
def add_backend(ctx, name, backend_type):
"""Add a new backend configuration."""
configs = load_backend_configs()
if name in configs:
if not confirm_action(f"Backend '{name}' already exists. Overwrite?"):
click.echo("Aborted")
return
if backend_type == 'local':
db_path = click.prompt('Database path', default=f'~/.config/issue-tracker/{name}.db')
config = {
'type': 'local',
'db_path': str(db_path)
}
elif backend_type == 'gitea':
base_url = click.prompt('Gitea base URL (e.g., https://git.example.com)')
owner = click.prompt('Repository owner/organization')
repo = click.prompt('Repository name')
# Check for API token in environment variable first
env_token = os.getenv('GITEA_API_TOKEN')
if env_token:
click.echo(f"Using API token from GITEA_API_TOKEN environment variable")
token = env_token
else:
token = click.prompt('Access token', hide_input=True)
config = {
'type': 'gitea',
'base_url': base_url.rstrip('/'),
'owner': owner,
'repo': repo,
'token': token
}
# Test connection
click.echo("Testing connection...")
if test_backend_connection(config):
echo_success("Connection successful!")
else:
echo_warning("Connection test failed, but configuration will be saved anyway.")
# Save configuration
configs[name] = config
save_backend_configs(configs)
echo_success(f"Backend '{name}' added successfully")
# Set as default if it's the first one
if 'default' not in configs:
configs['default'] = name
save_backend_configs(configs)
echo_info(f"Set '{name}' as default backend")
@backend_group.command('remove')
@click.argument('name')
def remove_backend(name):
"""Remove a backend configuration."""
configs = load_backend_configs()
if name not in configs:
echo_error(f"Backend '{name}' not found")
return
if not confirm_action(f"Remove backend '{name}'?"):
click.echo("Aborted")
return
del configs[name]
# Update default if necessary
if configs.get('default') == name:
remaining_backends = [k for k in configs.keys() if k != 'default']
if remaining_backends:
configs['default'] = remaining_backends[0]
echo_info(f"Set '{configs['default']}' as new default backend")
else:
del configs['default']
save_backend_configs(configs)
echo_success(f"Backend '{name}' removed")
@backend_group.command('test')
@click.argument('name')
def test_backend(name):
"""Test backend connection."""
configs = load_backend_configs()
if name not in configs:
echo_error(f"Backend '{name}' not found")
return
config = configs[name]
click.echo(f"Testing connection to '{name}'...")
if test_backend_connection(config):
echo_success("Connection successful!")
else:
echo_error("Connection failed!")
@backend_group.command('set-default')
@click.argument('name')
def set_default_backend(name):
"""Set default backend."""
configs = load_backend_configs()
if name not in configs:
echo_error(f"Backend '{name}' not found")
return
configs['default'] = name
save_backend_configs(configs)
echo_success(f"Set '{name}' as default backend")

422
issue_tracker/cli/commands.py Normal file
View File

@@ -0,0 +1,422 @@
"""
Issue Management CLI Commands
Core commands for managing issues: create, list, show, edit, close, etc.
"""
import click
from datetime import datetime, timezone
from typing import Optional
from ..core.models import Issue, Label, User, IssueState, Priority, IssueType
from ..core.interfaces import IssueFilter
from .utils import get_backend, format_issue, format_issue_list, get_user_input
@click.group()
def issue_group():
"""Issue management commands."""
pass
@issue_group.command('list')
@click.option('--state', type=click.Choice(['open', 'closed', 'all']), default='open', help='Issue state filter')
@click.option('--assignee', help='Filter by assignee')
@click.option('--label', multiple=True, help='Filter by labels')
@click.option('--milestone', help='Filter by milestone')
@click.option('--search', help='Search in title and description')
@click.option('--limit', type=int, default=30, help='Maximum number of issues to show')
@click.option('--format', 'output_format', type=click.Choice(['table', 'json', 'compact']), default='table', help='Output format')
@click.pass_context
def list_issues(ctx, state, assignee, label, milestone, search, limit, output_format):
"""List issues with optional filtering."""
backend = get_backend(ctx)
# Build filter criteria
# Handle Click Sentinel values
search_value = search if search is not None and not str(search).startswith('Sentinel') else None
assignee_value = assignee if assignee is not None and not str(assignee).startswith('Sentinel') else None
milestone_value = milestone if milestone is not None and not str(milestone).startswith('Sentinel') else None
filter_criteria = IssueFilter(
state=None if state == 'all' else state,
assignee=assignee_value,
labels=list(label) if label else None,
milestone=milestone_value,
search=search_value,
limit=limit
)
try:
issues = backend.list_issues(filter_criteria)
if not issues:
click.echo("No issues found.")
return
if output_format == 'json':
import json
click.echo(json.dumps([issue.to_dict() for issue in issues], indent=2))
elif output_format == 'compact':
for issue in issues:
labels_str = ', '.join(label.name for label in issue.labels[:3])
if len(issue.labels) > 3:
labels_str += f' (+{len(issue.labels) - 3} more)'
assignee_str = issue.primary_assignee.username if issue.primary_assignee else 'unassigned'
click.echo(f"#{issue.number:4d} {issue.state.value:10s} {issue.title[:50]:50s} {assignee_str:15s} {labels_str}")
else: # table format
click.echo(format_issue_list(issues))
except Exception as e:
raise click.ClickException(f"Failed to list issues: {e}")
@issue_group.command('show')
@click.argument('issue_number', type=int)
@click.option('--comments', is_flag=True, help='Show comments')
@click.option('--format', 'output_format', type=click.Choice(['detailed', 'json', 'compact']), default='detailed', help='Output format')
@click.pass_context
def show_issue(ctx, issue_number, comments, output_format):
"""Show detailed information about an issue."""
backend = get_backend(ctx)
try:
issue = backend.get_issue_by_number(issue_number)
if not issue:
raise click.ClickException(f"Issue #{issue_number} not found")
if output_format == 'json':
import json
issue_dict = issue.to_dict()
if comments:
issue_dict['comments'] = [
{
'id': c.id,
'body': c.body,
'author': c.author.username,
'created_at': c.created_at.isoformat()
}
for c in backend.get_comments(str(issue.number))
]
click.echo(json.dumps(issue_dict, indent=2))
else:
click.echo(format_issue(issue, show_comments=comments, backend=backend if comments else None))
except Exception as e:
raise click.ClickException(f"Failed to show issue: {e}")
@issue_group.command('create')
@click.argument('title')
@click.option('--description', '-d', help='Issue description')
@click.option('--label', '-l', multiple=True, help='Labels to add')
@click.option('--assignee', '-a', help='Assign to user')
@click.option('--milestone', '-m', help='Milestone')
@click.option('--priority', type=click.Choice(['low', 'medium', 'high', 'critical']), help='Issue priority')
@click.option('--type', 'issue_type', type=click.Choice(['bug', 'feature', 'enhancement', 'task', 'documentation', 'question']), help='Issue type')
@click.option('--interactive', '-i', is_flag=True, help='Interactive mode')
@click.pass_context
def create_issue(ctx, title, description, label, assignee, milestone, priority, issue_type, interactive):
"""Create a new issue."""
backend = get_backend(ctx)
try:
# Interactive mode
if interactive:
title = title or click.prompt('Title')
description = get_user_input('Description (optional)', multiline=True)
# Show available labels
available_labels = backend.get_labels()
if available_labels:
click.echo(f"\nAvailable labels: {', '.join(l.name for l in available_labels)}")
label = click.prompt('Labels (comma-separated, optional)', default='').split(',') if click.prompt('Add labels?', type=bool, default=False) else []
# Show available users
available_users = backend.get_users()
if available_users:
click.echo(f"\nAvailable users: {', '.join(u.username for u in available_users)}")
assignee = click.prompt('Assignee (optional)', default='') or None
# Show available milestones
available_milestones = backend.get_milestones()
if available_milestones:
click.echo(f"\nAvailable milestones: {', '.join(m.title for m in available_milestones)}")
milestone = click.prompt('Milestone (optional)', default='') or None
# Build labels list
labels = []
# Add explicit labels
for label_name in label:
label_name = label_name.strip()
if label_name:
labels.append(Label(name=label_name))
# Add priority label
if priority:
labels.append(Label(name=f'priority:{priority}'))
# Add type label
if issue_type:
labels.append(Label(name=issue_type))
# Build assignees list
assignees = []
if assignee:
# Try to find user
users = backend.search_users(assignee)
if users:
assignees.append(users[0])
else:
# Create a basic user object
assignees.append(User(id=assignee, username=assignee))
# Find milestone
milestone_obj = None
if milestone:
milestones = backend.get_milestones()
for m in milestones:
if m.title == milestone or m.id == milestone:
milestone_obj = m
break
# Create issue
now = datetime.now(timezone.utc)
issue = Issue(
id="", # Will be set by backend
number=0, # Will be set by backend
title=title,
description=description or "",
state=IssueState.OPEN,
created_at=now,
updated_at=now,
labels=labels,
assignees=assignees,
milestone=milestone_obj
)
created_issue = backend.create_issue(issue)
click.echo(f"Created issue #{created_issue.number}: {created_issue.title}")
if ctx.obj.get('verbose'):
click.echo(format_issue(created_issue))
except Exception as e:
raise click.ClickException(f"Failed to create issue: {e}")
@issue_group.command('edit')
@click.argument('issue_number', type=int)
@click.option('--title', help='New title')
@click.option('--description', help='New description')
@click.option('--add-label', multiple=True, help='Labels to add')
@click.option('--remove-label', multiple=True, help='Labels to remove')
@click.option('--assign', help='User to assign')
@click.option('--unassign', help='User to unassign')
@click.option('--milestone', help='Milestone to set')
@click.option('--interactive', '-i', is_flag=True, help='Interactive editing')
@click.pass_context
def edit_issue(ctx, issue_number, title, description, add_label, remove_label, assign, unassign, milestone, interactive):
"""Edit an existing issue."""
backend = get_backend(ctx)
try:
issue = backend.get_issue_by_number(issue_number)
if not issue:
raise click.ClickException(f"Issue #{issue_number} not found")
# Interactive mode
if interactive:
click.echo(f"Editing issue #{issue.number}: {issue.title}")
new_title = click.prompt('Title', default=issue.title)
if new_title != issue.title:
title = new_title
new_description = get_user_input('Description', default=issue.description, multiline=True)
if new_description != issue.description:
description = new_description
# Apply changes
modified = False
if title:
issue.title = title
modified = True
if description is not None:
issue.description = description
modified = True
# Add labels
for label_name in add_label:
issue.add_label(Label(name=label_name.strip()))
modified = True
# Remove labels
for label_name in remove_label:
if issue.remove_label(label_name.strip()):
modified = True
# Assign user
if assign:
users = backend.search_users(assign)
if users:
issue.add_assignee(users[0])
modified = True
else:
issue.add_assignee(User(id=assign, username=assign))
modified = True
# Unassign user
if unassign:
if issue.remove_assignee(unassign):
modified = True
# Set milestone
if milestone:
milestones = backend.get_milestones()
for m in milestones:
if m.title == milestone or m.id == milestone:
issue.milestone = m
modified = True
break
if modified:
issue.updated_at = datetime.now(timezone.utc)
updated_issue = backend.update_issue(issue)
click.echo(f"Updated issue #{updated_issue.number}")
if ctx.obj.get('verbose'):
click.echo(format_issue(updated_issue))
else:
click.echo("No changes made")
except Exception as e:
raise click.ClickException(f"Failed to edit issue: {e}")
@issue_group.command('close')
@click.argument('issue_number', type=int)
@click.option('--comment', '-c', help='Closing comment')
@click.pass_context
def close_issue(ctx, issue_number, comment):
"""Close an issue."""
backend = get_backend(ctx)
try:
issue = backend.get_issue_by_number(issue_number)
if not issue:
raise click.ClickException(f"Issue #{issue_number} not found")
if issue.state == IssueState.CLOSED:
click.echo(f"Issue #{issue_number} is already closed")
return
# Close the issue
issue.close()
# Add closing comment if provided
if comment:
from ..core.models import Comment
current_user = User(id="cli-user", username="cli-user") # TODO: Get actual user
closing_comment = Comment(
id="",
body=comment,
author=current_user,
created_at=datetime.now(timezone.utc)
)
backend.add_comment(str(issue.number), closing_comment)
updated_issue = backend.update_issue(issue)
click.echo(f"Closed issue #{updated_issue.number}: {updated_issue.title}")
except Exception as e:
raise click.ClickException(f"Failed to close issue: {e}")
@issue_group.command('reopen')
@click.argument('issue_number', type=int)
@click.option('--comment', '-c', help='Reopening comment')
@click.pass_context
def reopen_issue(ctx, issue_number, comment):
"""Reopen a closed issue."""
backend = get_backend(ctx)
try:
issue = backend.get_issue_by_number(issue_number)
if not issue:
raise click.ClickException(f"Issue #{issue_number} not found")
if issue.state != IssueState.CLOSED:
click.echo(f"Issue #{issue_number} is not closed (current state: {issue.state.value})")
return
# Reopen the issue
issue.reopen()
# Add reopening comment if provided
if comment:
from ..core.models import Comment
current_user = User(id="cli-user", username="cli-user") # TODO: Get actual user
reopening_comment = Comment(
id="",
body=comment,
author=current_user,
created_at=datetime.now(timezone.utc)
)
backend.add_comment(str(issue.number), reopening_comment)
updated_issue = backend.update_issue(issue)
click.echo(f"Reopened issue #{updated_issue.number}: {updated_issue.title}")
except Exception as e:
raise click.ClickException(f"Failed to reopen issue: {e}")
@issue_group.command('comment')
@click.argument('issue_number', type=int)
@click.argument('comment_text', required=False)
@click.option('--editor', is_flag=True, help='Open editor for comment')
@click.pass_context
def add_comment(ctx, issue_number, comment_text, editor):
"""Add a comment to an issue."""
backend = get_backend(ctx)
try:
issue = backend.get_issue_by_number(issue_number)
if not issue:
raise click.ClickException(f"Issue #{issue_number} not found")
# Get comment text
if editor:
comment_text = click.edit() or ""
elif not comment_text:
comment_text = get_user_input("Comment", multiline=True)
if not comment_text.strip():
click.echo("Empty comment, aborting")
return
# Create comment
from ..core.models import Comment
current_user = User(id="cli-user", username="cli-user") # TODO: Get actual user
comment = Comment(
id="",
body=comment_text.strip(),
author=current_user,
created_at=datetime.now(timezone.utc)
)
added_comment = backend.add_comment(str(issue.number), comment)
click.echo(f"Added comment to issue #{issue_number}")
if ctx.obj.get('verbose'):
click.echo(f"\nComment by {added_comment.author.username} at {added_comment.created_at}:")
click.echo(added_comment.body)
except Exception as e:
raise click.ClickException(f"Failed to add comment: {e}")

118
issue_tracker/cli/main.py Normal file
View File

@@ -0,0 +1,118 @@
"""
Main CLI Entry Point
Universal Issue Tracking System CLI
"""
import click
import sys
from pathlib import Path
from .commands import issue_group
from .backend_commands import backend_group
from .sync_commands import sync_group
from .. import __version__
@click.group()
@click.version_option(version=__version__, package_name='issue-tracker')
@click.option('--config', type=click.Path(), help='Configuration file path')
@click.option('--backend', help='Backend to use (local, gitea)')
@click.option('--verbose', '-v', is_flag=True, help='Verbose output')
@click.pass_context
def cli(ctx, config, backend, verbose):
"""
Universal Issue Tracking System
A backend-agnostic issue tracking tool that works with local SQLite,
Gitea, GitHub, and other issue tracking systems.
Examples:
issue list # List all issues
issue create "Bug in parser" # Create new issue
issue show 42 # Show issue #42
issue close 42 # Close issue #42
backend add local ~/.issues # Add local backend
backend add gitea myrepo # Add Gitea backend
sync pull gitea # Sync from Gitea
sync push gitea # Sync to Gitea
"""
# Ensure the object exists
ctx.ensure_object(dict)
# Store global options in context
ctx.obj['config_path'] = config
ctx.obj['backend'] = backend
ctx.obj['verbose'] = verbose
# Register command groups
cli.add_command(issue_group, name='issue')
cli.add_command(backend_group, name='backend')
cli.add_command(sync_group, name='sync')
# Convenience aliases - direct issue commands
@cli.command('list')
@click.pass_context
def list_issues(ctx):
"""List all issues (alias for 'issue list')."""
ctx.invoke(issue_group.get_command(ctx, 'list'))
@cli.command('show')
@click.argument('issue_number', type=int)
@click.pass_context
def show_issue(ctx, issue_number):
"""Show issue details (alias for 'issue show')."""
ctx.invoke(issue_group.get_command(ctx, 'show'), issue_number=issue_number)
@cli.command('create')
@click.argument('title')
@click.option('--description', '-d', help='Issue description')
@click.option('--label', '-l', multiple=True, help='Labels to add')
@click.option('--assignee', '-a', help='Assign to user')
@click.option('--milestone', '-m', help='Milestone')
@click.pass_context
def create_issue(ctx, title, description, label, assignee, milestone):
"""Create new issue (alias for 'issue create')."""
ctx.invoke(
issue_group.get_command(ctx, 'create'),
title=title,
description=description,
label=label,
assignee=assignee,
milestone=milestone
)
@cli.command('close')
@click.argument('issue_number', type=int)
@click.option('--comment', '-c', help='Closing comment')
@click.pass_context
def close_issue(ctx, issue_number, comment):
"""Close issue (alias for 'issue close')."""
ctx.invoke(
issue_group.get_command(ctx, 'close'),
issue_number=issue_number,
comment=comment
)
def main():
"""Main entry point for the CLI."""
try:
cli(obj={})
except KeyboardInterrupt:
click.echo("\nAborted by user", err=True)
sys.exit(1)
except Exception as e:
click.echo(f"Error: {e}", err=True)
sys.exit(1)
if __name__ == '__main__':
main()

235
issue_tracker/cli/sync_commands.py Normal file
View File

@@ -0,0 +1,235 @@
"""
Synchronization CLI Commands
Commands for synchronizing issues between different backends.
"""
import click
from datetime import datetime, timezone
from .utils import (
load_backend_configs, get_backend, echo_success, echo_error,
echo_warning, echo_info, progress_bar, confirm_action
)
from ..core.interfaces import BackendFactory
@click.group()
def sync_group():
"""Issue synchronization between backends."""
pass
@sync_group.command('status')
@click.argument('backend_name', required=False)
@click.pass_context
def sync_status(ctx, backend_name):
"""Show sync status for backends."""
configs = load_backend_configs()
if backend_name:
backends_to_check = [backend_name] if backend_name in configs else []
if not backends_to_check:
echo_error(f"Backend '{backend_name}' not found")
return
else:
backends_to_check = [k for k in configs.keys() if k != 'default']
for name in backends_to_check:
config = configs[name]
try:
backend = BackendFactory.create_backend(config['type'])
backend.connect(config)
# Get basic stats
all_issues = backend.list_issues()
open_issues = [i for i in all_issues if i.state.value != 'closed']
click.echo(f"\n{name} ({config['type']}):")
click.echo(f" Total issues: {len(all_issues)}")
click.echo(f" Open issues: {len(open_issues)}")
click.echo(f" Closed issues: {len(all_issues) - len(open_issues)}")
# Check last sync
if hasattr(backend, 'get_last_sync_timestamp'):
last_sync = backend.get_last_sync_timestamp()
if last_sync:
click.echo(f" Last sync: {last_sync}")
else:
click.echo(f" Last sync: never")
backend.disconnect()
except Exception as e:
echo_error(f" Error accessing {name}: {e}")
@sync_group.command('pull')
@click.argument('source_backend')
@click.option('--target', default='local', help='Target backend (default: local)')
@click.option('--dry-run', is_flag=True, help='Show what would be synced without making changes')
@click.option('--force', is_flag=True, help='Force sync even with conflicts')
@click.pass_context
def sync_pull(ctx, source_backend, target, dry_run, force):
"""Pull issues from source backend to target backend."""
configs = load_backend_configs()
if source_backend not in configs:
echo_error(f"Source backend '{source_backend}' not found")
return
if target not in configs:
echo_error(f"Target backend '{target}' not found")
return
if source_backend == target:
echo_error("Source and target backends cannot be the same")
return
try:
# Connect to backends
source_config = configs[source_backend]
target_config = configs[target]
source = BackendFactory.create_backend(source_config['type'])
source.connect(source_config)
target = BackendFactory.create_backend(target_config['type'])
target.connect(target_config)
echo_info(f"Syncing from {source_backend} to {target}")
# Get issues from source
source_issues = source.list_issues()
echo_info(f"Found {len(source_issues)} issues in source")
# Get existing issues in target
target_issues = target.list_issues()
target_numbers = {issue.number for issue in target_issues}
# Determine what needs to be synced
new_issues = []
updated_issues = []
for issue in source_issues:
if issue.number not in target_numbers:
new_issues.append(issue)
else:
# Check if update is needed (simplified check by updated_at)
target_issue = next((i for i in target_issues if i.number == issue.number), None)
if target_issue and issue.updated_at > target_issue.updated_at:
updated_issues.append(issue)
echo_info(f"New issues to sync: {len(new_issues)}")
echo_info(f"Updated issues to sync: {len(updated_issues)}")
if dry_run:
if new_issues:
click.echo("\nNew issues:")
for issue in new_issues:
click.echo(f" #{issue.number}: {issue.title}")
if updated_issues:
click.echo("\nUpdated issues:")
for issue in updated_issues:
click.echo(f" #{issue.number}: {issue.title}")
click.echo(f"\nDry run complete. {len(new_issues + updated_issues)} issues would be synced.")
return
# Confirm sync
total_sync = len(new_issues) + len(updated_issues)
if total_sync == 0:
echo_success("No issues need syncing")
return
if not force and not confirm_action(f"Sync {total_sync} issues?"):
click.echo("Aborted")
return
# Perform sync
synced_count = 0
errors = []
all_to_sync = new_issues + updated_issues
with progress_bar(all_to_sync, label="Syncing issues") as items:
for issue in items:
try:
# Clear backend-specific IDs for new backend
issue.backend_id = None
issue.backend_type = target_config['type']
if issue in new_issues:
target.create_issue(issue)
else:
# For updates, we need to find the target issue and update it
target_issue = target.get_issue_by_number(issue.number)
if target_issue:
# Copy relevant fields
target_issue.title = issue.title
target_issue.description = issue.description
target_issue.state = issue.state
target_issue.labels = issue.labels
target_issue.assignees = issue.assignees
target_issue.milestone = issue.milestone
target_issue.updated_at = issue.updated_at
target_issue.closed_at = issue.closed_at
target.update_issue(target_issue)
synced_count += 1
except Exception as e:
errors.append(f"Issue #{issue.number}: {e}")
# Report results
echo_success(f"Synced {synced_count} issues successfully")
if errors:
echo_warning(f"{len(errors)} errors occurred:")
for error in errors[:5]: # Show first 5 errors
echo_error(f" {error}")
if len(errors) > 5:
echo_warning(f" ... and {len(errors) - 5} more errors")
# Cleanup
source.disconnect()
target.disconnect()
except Exception as e:
echo_error(f"Sync failed: {e}")
@sync_group.command('push')
@click.argument('target_backend')
@click.option('--source', default='local', help='Source backend (default: local)')
@click.option('--dry-run', is_flag=True, help='Show what would be synced without making changes')
@click.option('--force', is_flag=True, help='Force sync even with conflicts')
@click.pass_context
def sync_push(ctx, target_backend, source, dry_run, force):
"""Push issues from source backend to target backend."""
# This is essentially the same as pull but with arguments swapped
ctx.invoke(sync_pull, source_backend=source, target=target_backend, dry_run=dry_run, force=force)
@sync_group.command('bidirectional')
@click.argument('backend1')
@click.argument('backend2')
@click.option('--dry-run', is_flag=True, help='Show what would be synced without making changes')
@click.option('--force', is_flag=True, help='Force sync even with conflicts')
@click.pass_context
def sync_bidirectional(ctx, backend1, backend2, dry_run, force):
"""Bidirectional sync between two backends."""
echo_warning("Bidirectional sync is a complex operation that can cause conflicts.")
if not force and not confirm_action("Continue with bidirectional sync?"):
click.echo("Aborted")
return
# First sync backend1 -> backend2
echo_info(f"Step 1: Syncing {backend1} -> {backend2}")
ctx.invoke(sync_pull, source_backend=backend1, target=backend2, dry_run=dry_run, force=True)
# Then sync backend2 -> backend1
echo_info(f"Step 2: Syncing {backend2} -> {backend1}")
ctx.invoke(sync_pull, source_backend=backend2, target=backend1, dry_run=dry_run, force=True)
echo_success("Bidirectional sync completed")

336
issue_tracker/cli/utils.py Normal file
View File

@@ -0,0 +1,336 @@
"""
CLI Utility Functions
Helper functions for the CLI commands including formatting, configuration,
backend management, and user interaction.
"""
import click
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional, List
import tempfile
from ..core.interfaces import IssueBackend, BackendFactory
from ..core.models import Issue, Comment
from ..backends.local import LocalSQLiteBackend
from ..backends.gitea import GiteaBackend
# Register available backends
BackendFactory.register_backend('local', LocalSQLiteBackend)
BackendFactory.register_backend('gitea', GiteaBackend)
def get_config_dir() -> Path:
"""Get configuration directory."""
config_dir = Path.home() / '.config' / 'issue-tracker'
config_dir.mkdir(parents=True, exist_ok=True)
return config_dir
def get_backend_config_path() -> Path:
"""Get backend configuration file path."""
return get_config_dir() / 'backends.json'
def load_backend_configs() -> dict:
"""Load backend configurations."""
config_path = get_backend_config_path()
if not config_path.exists():
return {}
import json
try:
with open(config_path, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return {}
def save_backend_configs(configs: dict) -> None:
"""Save backend configurations."""
config_path = get_backend_config_path()
import json
with open(config_path, 'w') as f:
json.dump(configs, f, indent=2)
def get_default_backend() -> str:
"""Get the default backend name."""
configs = load_backend_configs()
return configs.get('default', 'local')
def get_backend(ctx) -> IssueBackend:
"""Get backend instance from context."""
backend_name = ctx.obj.get('backend') or get_default_backend()
configs = load_backend_configs()
if backend_name not in configs:
if backend_name == 'local':
# Auto-configure local backend
local_config = {
'type': 'local',
'db_path': str(get_config_dir() / 'issues.db')
}
configs['local'] = local_config
save_backend_configs(configs)
else:
raise click.ClickException(f"Backend '{backend_name}' not configured. Use 'backend add' to configure it.")
backend_config = configs[backend_name]
backend_type = backend_config['type']
try:
backend = BackendFactory.create_backend(backend_type)
backend.connect(backend_config)
return backend
except Exception as e:
raise click.ClickException(f"Failed to connect to backend '{backend_name}': {e}")
def format_issue_list(issues: List[Issue]) -> str:
"""Format list of issues as a table."""
if not issues:
return "No issues found."
# Calculate column widths
max_title_width = min(50, max(len(issue.title) for issue in issues))
max_assignee_width = 15
# Header
lines = []
header = f"{'#':<6} {'State':<12} {'Title':<{max_title_width}} {'Assignee':<{max_assignee_width}} Labels"
lines.append(header)
lines.append("-" * len(header))
# Issues
for issue in issues:
title = issue.title[:max_title_width]
if len(issue.title) > max_title_width:
title = title[:-3] + "..."
assignee = issue.primary_assignee.username if issue.primary_assignee else "unassigned"
assignee = assignee[:max_assignee_width]
labels = ", ".join(label.name for label in issue.labels[:3])
if len(issue.labels) > 3:
labels += f" (+{len(issue.labels) - 3})"
line = f"{issue.number:<6} {issue.state.value:<12} {title:<{max_title_width}} {assignee:<{max_assignee_width}} {labels}"
lines.append(line)
return "\n".join(lines)
def format_issue(issue: Issue, show_comments: bool = False, backend: Optional[IssueBackend] = None) -> str:
"""Format a single issue with details."""
lines = []
# Header
lines.append(f"#{issue.number}: {issue.title}")
lines.append("=" * (len(f"#{issue.number}: {issue.title}")))
lines.append("")
# Basic info
lines.append(f"State: {issue.state.value}")
lines.append(f"Created: {format_datetime(issue.created_at)}")
lines.append(f"Updated: {format_datetime(issue.updated_at)}")
if issue.closed_at:
lines.append(f"Closed: {format_datetime(issue.closed_at)}")
# Assignees
if issue.assignees:
assignees_str = ", ".join(assignee.username for assignee in issue.assignees)
lines.append(f"Assignees: {assignees_str}")
else:
lines.append("Assignees: none")
# Milestone
if issue.milestone:
lines.append(f"Milestone: {issue.milestone.title}")
# Labels
if issue.labels:
labels_by_category = {}
for label in issue.labels:
category = label.category
if category not in labels_by_category:
labels_by_category[category] = []
labels_by_category[category].append(label.name)
for category, label_names in labels_by_category.items():
lines.append(f"{category.title()} labels: {', '.join(label_names)}")
else:
lines.append("Labels: none")
# Description
lines.append("")
lines.append("Description:")
lines.append("-" * 12)
if issue.description:
lines.append(issue.description)
else:
lines.append("(no description)")
# Comments
if show_comments and backend:
comments = backend.get_comments(str(issue.number))
if comments:
lines.append("")
lines.append(f"Comments ({len(comments)}):")
lines.append("-" * 20)
for comment in comments:
lines.append("")
lines.append(f"Comment by {comment.author.username} at {format_datetime(comment.created_at)}:")
lines.append(comment.body)
return "\n".join(lines)
def format_datetime(dt: datetime) -> str:
"""Format datetime for display."""
if dt.tzinfo:
dt = dt.astimezone()
return dt.strftime("%Y-%m-%d %H:%M:%S")
def get_user_input(prompt: str, default: str = "", multiline: bool = False) -> str:
"""Get user input with optional default and multiline support."""
if multiline:
click.echo(f"{prompt} (Press Ctrl+D when done, Ctrl+C to cancel):")
if default:
click.echo(f"Current value:\n{default}\n")
lines = []
try:
while True:
try:
line = input()
lines.append(line)
except EOFError:
break
except KeyboardInterrupt:
raise click.Abort()
return "\n".join(lines) if lines else default
else:
return click.prompt(prompt, default=default)
def validate_backend_type(backend_type: str) -> bool:
"""Validate that a backend type is supported."""
return backend_type in BackendFactory.get_available_backends()
def test_backend_connection(backend_config: dict) -> bool:
"""Test if a backend configuration works."""
try:
backend_type = backend_config['type']
backend = BackendFactory.create_backend(backend_type)
backend.connect(backend_config)
result = backend.test_connection()
backend.disconnect()
return result
except Exception:
return False
def format_backend_list(configs: dict) -> str:
"""Format backend configurations for display."""
if not configs:
return "No backends configured."
lines = []
default_backend = configs.get('default', 'local')
lines.append(f"{'Name':<15} {'Type':<10} {'Status':<10} Description")
lines.append("-" * 60)
for name, config in configs.items():
if name == 'default':
continue
backend_type = config.get('type', 'unknown')
is_default = name == default_backend
# Test connection
try:
status = "connected" if test_backend_connection(config) else "error"
except Exception:
status = "error"
# Description
if backend_type == 'local':
desc = f"Local SQLite: {config.get('db_path', 'unknown')}"
elif backend_type == 'gitea':
desc = f"Gitea: {config.get('base_url', 'unknown')}/{config.get('owner', 'unknown')}/{config.get('repo', 'unknown')}"
else:
desc = f"{backend_type} backend"
if is_default:
desc += " (default)"
line = f"{name:<15} {backend_type:<10} {status:<10} {desc}"
lines.append(line)
return "\n".join(lines)
def get_editor() -> str:
"""Get the user's preferred editor."""
return os.environ.get('EDITOR', 'nano')
def edit_text(initial_text: str = "") -> Optional[str]:
"""Open text in editor and return edited content."""
with tempfile.NamedTemporaryFile(mode='w+', suffix='.md', delete=False) as f:
f.write(initial_text)
temp_path = f.name
try:
editor = get_editor()
os.system(f'{editor} {temp_path}')
with open(temp_path, 'r') as f:
edited_text = f.read()
return edited_text if edited_text != initial_text else None
finally:
os.unlink(temp_path)
def confirm_action(message: str, default: bool = False) -> bool:
"""Ask user for confirmation."""
return click.confirm(message, default=default)
def progress_bar(items, label: str = "Processing"):
"""Create a progress bar for iterating over items."""
return click.progressbar(items, label=label)
def echo_success(message: str) -> None:
"""Echo success message in green."""
click.echo(click.style(message, fg='green'))
def echo_warning(message: str) -> None:
"""Echo warning message in yellow."""
click.echo(click.style(message, fg='yellow'))
def echo_error(message: str) -> None:
"""Echo error message in red."""
click.echo(click.style(message, fg='red'))
def echo_info(message: str) -> None:
"""Echo info message in blue."""
click.echo(click.style(message, fg='blue'))