<s> import os <EOL> from collections import namedtuple <EOL> from shutil import rmtree <EOL> from stat import S_IFDIR , S_IFREG , S_IFLNK <EOL> from pygit2 import ( clone_repository , Signature , GIT_SORT_TOPOLOGICAL , <EOL> GIT_FILEMODE_TREE , GIT_STATUS_CURRENT , <EOL> GIT_FILEMODE_LINK , GIT_FILEMODE_BLOB , GIT_BRANCH_REMOTE , <EOL> GIT_BRANCH_LOCAL , GIT_FILEMODE_BLOB_EXECUTABLE ) <EOL> from six import iteritems <EOL> from gitfs . cache import CommitCache <EOL> from gitfs . log import log <EOL> from gitfs . utils . path import split_path_into_components <EOL> from gitfs . utils . commits import CommitsList <EOL> DivergeCommits = namedtuple ( "<STR_LIT>" , [ "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> class Repository ( object ) : <EOL> def __init__ ( self , repository , commits = None ) : <EOL> self . _repo = repository <EOL> self . commits = commits or CommitCache ( self ) <EOL> self . behind = False <EOL> def __getitem__ ( self , item ) : <EOL> """<STR_LIT>""" <EOL> return self . _repo [ item ] <EOL> def __getattr__ ( self , attr ) : <EOL> """<STR_LIT>""" <EOL> if attr not in self . __dict__ : <EOL> return getattr ( self . _repo , attr ) <EOL> else : <EOL> return self . __dict__ [ attr ] <EOL> def ahead ( self , upstream , branch ) : <EOL> ahead , _ = self . diverge ( upstream , branch ) <EOL> return ahead <EOL> def diverge ( self , upstream , branch ) : <EOL> reference = "<STR_LIT>" . format ( upstream , branch ) <EOL> remote_branch = self . lookup_branch ( reference , GIT_BRANCH_REMOTE ) <EOL> local_branch = self . lookup_branch ( branch , GIT_BRANCH_LOCAL ) <EOL> if remote_branch . target == local_branch . target : <EOL> return False , False <EOL> diverge_commits = self . find_diverge_commits ( local_branch , <EOL> remote_branch ) <EOL> behind = len ( diverge_commits . second_commits ) > <NUM_LIT:0> <EOL> ahead = len ( diverge_commits . first_commits ) > <NUM_LIT:0> <EOL> return ahead , behind <EOL> def checkout ( self , ref , * args , ** kwargs ) : <EOL> result = self . _repo . checkout ( ref , * args , ** kwargs ) <EOL> self . ignore . update ( ) <EOL> status = self . _repo . status ( ) <EOL> for path , status in iteritems ( status ) : <EOL> if status == GIT_STATUS_CURRENT : <EOL> continue <EOL> full_path = self . _full_path ( path ) <EOL> if path not in self . _repo . index : <EOL> if path not in self . ignore : <EOL> try : <EOL> os . unlink ( full_path ) <EOL> except OSError : <EOL> rmtree ( <EOL> full_path , <EOL> onerror = lambda function , fpath , excinfo : log . info ( <EOL> "<STR_LIT>" , fpath <EOL> ) <EOL> ) <EOL> continue <EOL> stats = self . get_git_object_default_stats ( ref , path ) <EOL> current_stat = os . lstat ( full_path ) <EOL> if stats [ '<STR_LIT>' ] != current_stat . st_mode : <EOL> os . chmod ( full_path , current_stat . st_mode ) <EOL> self . _repo . index . add ( self . _sanitize ( path ) ) <EOL> return result <EOL> def _sanitize ( self , path ) : <EOL> if path is not None and path . startswith ( "<STR_LIT:/>" ) : <EOL> path = path [ <NUM_LIT:1> : ] <EOL> return path <EOL> def push ( self , upstream , branch , credentials ) : <EOL> """<STR_LIT>""" <EOL> remote = self . get_remote ( upstream ) <EOL> remote . push ( [ "<STR_LIT>" % ( branch ) ] , callbacks = credentials ) <EOL> def fetch ( self , upstream , branch_name , credentials ) : <EOL> """<STR_LIT>""" <EOL> remote = self . get_remote ( upstream ) <EOL> remote . fetch ( callbacks = credentials ) <EOL> _ , behind = self . diverge ( upstream , branch_name ) <EOL> self . behind = behind <EOL> return behind <EOL> def commit ( self , message , author , commiter , parents = None , ref = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> status = self . _repo . status ( ) <EOL> if status == { } : <EOL> return None <EOL> author = Signature ( author [ <NUM_LIT:0> ] , author [ <NUM_LIT:1> ] ) <EOL> commiter = Signature ( commiter [ <NUM_LIT:0> ] , commiter [ <NUM_LIT:1> ] ) <EOL> tree = self . _repo . index . write_tree ( ) <EOL> self . _repo . index . write ( ) <EOL> if parents is None : <EOL> parents = [ self . _repo . revparse_single ( ref ) . id ] <EOL> return self . _repo . create_commit ( ref , author , commiter , message , <EOL> tree , parents ) <EOL> @ classmethod <EOL> def clone ( cls , remote_url , path , branch = None , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> repo = clone_repository ( remote_url , path , checkout_branch = branch , <EOL> callbacks = credentials ) <EOL> repo . checkout_head ( ) <EOL> return cls ( repo ) <EOL> def _is_searched_entry ( self , entry_name , searched_entry , path_components ) : <EOL> """<STR_LIT>""" <EOL> return ( entry_name == searched_entry and <EOL> len ( path_components ) == <NUM_LIT:1> and <EOL> entry_name == path_components [ <NUM_LIT:0> ] ) <EOL> def _get_git_object ( self , tree , obj_name , path_components , modifier ) : <EOL> """<STR_LIT>""" <EOL> git_obj = None <EOL> for entry in tree : <EOL> if self . _is_searched_entry ( entry . name , obj_name , path_components ) : <EOL> return modifier ( entry ) <EOL> elif entry . filemode == GIT_FILEMODE_TREE : <EOL> git_obj = self . _get_git_object ( self . _repo [ entry . id ] , obj_name , <EOL> path_components [ <NUM_LIT:1> : ] , modifier ) <EOL> if git_obj : <EOL> return git_obj <EOL> return git_obj <EOL> def get_git_object_type ( self , tree , path ) : <EOL> """<STR_LIT>""" <EOL> path_components = split_path_into_components ( path ) <EOL> try : <EOL> return self . _get_git_object ( tree , path_components [ - <NUM_LIT:1> ] , <EOL> path_components , <EOL> lambda entry : entry . filemode ) <EOL> except : <EOL> return GIT_FILEMODE_TREE <EOL> def get_git_object ( self , tree , path ) : <EOL> """<STR_LIT>""" <EOL> path_components = split_path_into_components ( path ) <EOL> return self . _get_git_object ( tree , path_components [ - <NUM_LIT:1> ] , path_components , <EOL> lambda entry : self . _repo [ entry . id ] ) <EOL> def get_git_object_default_stats ( self , ref , path ) : <EOL> types = { <EOL> GIT_FILEMODE_LINK : { <EOL> '<STR_LIT>' : S_IFLNK | <NUM_LIT> , <EOL> } , GIT_FILEMODE_TREE : { <EOL> '<STR_LIT>' : S_IFDIR | <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:2> <EOL> } , GIT_FILEMODE_BLOB : { <EOL> '<STR_LIT>' : S_IFREG | <NUM_LIT> , <EOL> } , GIT_FILEMODE_BLOB_EXECUTABLE : { <EOL> '<STR_LIT>' : S_IFREG | <NUM_LIT> , <EOL> } , <EOL> } <EOL> if path == "<STR_LIT:/>" : <EOL> return types [ GIT_FILEMODE_TREE ] <EOL> obj_type = self . get_git_object_type ( ref , path ) <EOL> if obj_type is None : <EOL> return obj_type <EOL> stats = types [ obj_type ] <EOL> if obj_type in [ GIT_FILEMODE_BLOB , GIT_FILEMODE_BLOB_EXECUTABLE ] : <EOL> stats [ '<STR_LIT>' ] = self . get_blob_size ( ref , path ) <EOL> return stats <EOL> def get_blob_size ( self , tree , path ) : <EOL> """<STR_LIT>""" <EOL> return self . get_git_object ( tree , path ) . size <EOL> def get_blob_data ( self , tree , path ) : <EOL> """<STR_LIT>""" <EOL> return self . get_git_object ( tree , path ) . data <EOL> def get_commit_dates ( self ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . commits . keys ( ) ) <EOL> def get_commits_by_date ( self , date ) : <EOL> """<STR_LIT>""" <EOL> return list ( map ( str , self . commits [ date ] ) ) <EOL> def walk_branches ( self , sort , * branches ) : <EOL> """<STR_LIT>""" <EOL> iterators = [ self . _repo . walk ( branch . target , sort ) <EOL> for branch in branches ] <EOL> stop_iteration = [ False for branch in branches ] <EOL> commits = [ ] <EOL> for iterator in iterators : <EOL> try : <EOL> commit = next ( iterator ) <EOL> except StopIteration : <EOL> commit = None <EOL> commits . append ( commit ) <EOL> yield ( commit for commit in commits ) <EOL> while not all ( stop_iteration ) : <EOL> for index , iterator in enumerate ( iterators ) : <EOL> try : <EOL> commit = next ( iterator ) <EOL> commits [ index ] = commit <EOL> except StopIteration : <EOL> stop_iteration [ index ] = True <EOL> if not all ( stop_iteration ) : <EOL> yield ( commit for commit in commits ) <EOL> def remote_head ( self , upstream , branch ) : <EOL> ref = "<STR_LIT>" % ( upstream , branch ) <EOL> remote = self . _repo . lookup_branch ( ref , GIT_BRANCH_REMOTE ) <EOL> return remote . get_object ( ) <EOL> def get_remote ( self , name ) : <EOL> """<STR_LIT>""" <EOL> remote = [ remote for remote in self . _repo . remotes <EOL> if remote . name == name ] <EOL> if not remote : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> return remote [ <NUM_LIT:0> ] <EOL> def _full_path ( self , partial ) : <EOL> if partial . startswith ( "<STR_LIT:/>" ) : <EOL> partial = partial [ <NUM_LIT:1> : ] <EOL> return os . path . join ( self . _repo . workdir , partial ) <EOL> def find_diverge_commits ( self , first_branch , second_branch ) : <EOL> """<STR_LIT>""" <EOL> common_parent = None <EOL> first_commits = CommitsList ( ) <EOL> second_commits = CommitsList ( ) <EOL> walker = self . walk_branches ( GIT_SORT_TOPOLOGICAL , <EOL> first_branch , second_branch ) <EOL> for first_commit , second_commit in walker : <EOL> if ( first_commit in second_commits or <EOL> second_commit in first_commits ) : <EOL> break <EOL> if first_commit not in first_commits : <EOL> first_commits . append ( first_commit ) <EOL> if second_commit not in second_commits : <EOL> second_commits . append ( second_commit ) <EOL> if second_commit . hex == first_commit . hex : <EOL> break <EOL> try : <EOL> index = second_commits . index ( first_commit ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> second_commits = second_commits [ : index ] <EOL> common_parent = first_commit <EOL> try : <EOL> index = first_commits . index ( second_commit ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> first_commits = first_commits [ : index ] <EOL> common_parent = second_commit <EOL> return DivergeCommits ( common_parent , first_commits , second_commits ) </s>
