""" Git operation model for tracking repository changes. """ from datetime import datetime from typing import Optional, List from sqlalchemy import String, Text, Integer, DateTime, JSON, ForeignKey, Boolean from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.sql import func from .base import Base, TimestampMixin class GitOperation(Base, TimestampMixin): """ Represents a git operation performed during a development session. Tracks commits, pushes, pulls, branch operations, and other git commands to provide insight into version control workflow. """ __tablename__ = "git_operations" # Primary key id: Mapped[int] = mapped_column(primary_key=True) # Foreign key to session session_id: Mapped[int] = mapped_column(ForeignKey("sessions.id"), nullable=False, index=True) # Operation timing timestamp: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, default=func.now(), index=True ) # Operation details operation: Mapped[str] = mapped_column(String(50), nullable=False, index=True) # commit, push, pull, branch, etc. command: Mapped[str] = mapped_column(Text, nullable=False) result: Mapped[Optional[str]] = mapped_column(Text, nullable=True) success: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False) # File and change tracking files_changed: Mapped[Optional[List[str]]] = mapped_column(JSON, nullable=True) lines_added: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) lines_removed: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Commit-specific fields commit_hash: Mapped[Optional[str]] = mapped_column(String(40), nullable=True, index=True) # Branch operation fields branch_from: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) branch_to: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) # Relationships session: Mapped["Session"] = relationship("Session", back_populates="git_operations") def __repr__(self) -> str: return f"" @property def is_commit(self) -> bool: """Check if this is a commit operation.""" return self.operation == "commit" @property def is_push(self) -> bool: """Check if this is a push operation.""" return self.operation == "push" @property def is_pull(self) -> bool: """Check if this is a pull operation.""" return self.operation == "pull" @property def is_branch_operation(self) -> bool: """Check if this is a branch-related operation.""" return self.operation in {"branch", "checkout", "merge", "rebase"} @property def total_lines_changed(self) -> int: """Get total lines changed (added + removed).""" added = self.lines_added or 0 removed = self.lines_removed or 0 return added + removed @property def net_lines_changed(self) -> int: """Get net lines changed (added - removed).""" added = self.lines_added or 0 removed = self.lines_removed or 0 return added - removed @property def files_count(self) -> int: """Get number of files changed.""" return len(self.files_changed) if self.files_changed else 0 def get_commit_message(self) -> Optional[str]: """Extract commit message from the command.""" if not self.is_commit: return None command = self.command if "-m" in command: # Extract message between quotes after -m parts = command.split("-m") if len(parts) > 1: message_part = parts[1].strip() # Remove quotes if message_part.startswith('"') and message_part.endswith('"'): return message_part[1:-1] elif message_part.startswith("'") and message_part.endswith("'"): return message_part[1:-1] else: # Find first quoted string import re match = re.search(r'["\']([^"\']*)["\']', message_part) if match: return match.group(1) return None def get_branch_name(self) -> Optional[str]: """Get branch name for branch operations.""" if self.branch_to: return self.branch_to elif self.branch_from: return self.branch_from # Try to extract from command if "checkout" in self.command: parts = self.command.split() if len(parts) > 2: return parts[-1] # Last argument is usually the branch return None def is_merge_commit(self) -> bool: """Check if this is a merge commit.""" commit_msg = self.get_commit_message() return commit_msg is not None and "merge" in commit_msg.lower() def is_feature_commit(self) -> bool: """Check if this appears to be a feature commit.""" commit_msg = self.get_commit_message() if not commit_msg: return False feature_keywords = ["add", "implement", "create", "new", "feature"] return any(keyword in commit_msg.lower() for keyword in feature_keywords) def is_bugfix_commit(self) -> bool: """Check if this appears to be a bugfix commit.""" commit_msg = self.get_commit_message() if not commit_msg: return False bugfix_keywords = ["fix", "bug", "resolve", "correct", "patch"] return any(keyword in commit_msg.lower() for keyword in bugfix_keywords) def is_refactor_commit(self) -> bool: """Check if this appears to be a refactoring commit.""" commit_msg = self.get_commit_message() if not commit_msg: return False refactor_keywords = ["refactor", "cleanup", "improve", "optimize", "reorganize"] return any(keyword in commit_msg.lower() for keyword in refactor_keywords) def get_commit_category(self) -> str: """Categorize the commit based on its message.""" if not self.is_commit: return "non-commit" if self.is_merge_commit(): return "merge" elif self.is_feature_commit(): return "feature" elif self.is_bugfix_commit(): return "bugfix" elif self.is_refactor_commit(): return "refactor" else: return "other" def get_change_size_category(self) -> str: """Categorize the size of changes in this operation.""" total_changes = self.total_lines_changed if total_changes == 0: return "no-changes" elif total_changes < 10: return "small" elif total_changes < 50: return "medium" elif total_changes < 200: return "large" else: return "very-large"