<s> """<STR_LIT>""" <EOL> import base64 <EOL> from itertools import chain <EOL> import shutil <EOL> import tempfile <EOL> import requests <EOL> import sh <EOL> CLONE_DIR = '<STR_LIT>' <EOL> class GitHubException ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GitHubRepoExists ( GitHubException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GitHubRepoDoesNotExist ( GitHubException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GitHubUnknownError ( GitHubException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GitHubNoTeamFound ( GitHubException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class GitHub ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , api_url , oauth2_token ) : <EOL> """<STR_LIT>""" <EOL> self . api_url = api_url <EOL> if not api_url . endswith ( '<STR_LIT:/>' ) : <EOL> self . api_url += '<STR_LIT:/>' <EOL> self . session = requests . Session ( ) <EOL> self . session . headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' . format ( oauth2_token ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def _get_all ( self , url ) : <EOL> """<STR_LIT>""" <EOL> results = None <EOL> response = self . session . get ( url ) <EOL> if response . status_code == <NUM_LIT:200> : <EOL> results = response . json ( ) <EOL> while ( <EOL> response . links . get ( '<STR_LIT>' , False ) and <EOL> response . status_code == <NUM_LIT:200> <EOL> ) : <EOL> response = self . session . get ( response . links [ '<STR_LIT>' ] [ '<STR_LIT:url>' ] ) <EOL> results += response . json ( ) <EOL> if response . status_code not in [ <NUM_LIT:200> , <NUM_LIT> ] : <EOL> raise GitHubUnknownError ( response . text ) <EOL> return results <EOL> def _get_repo ( self , org , repo ) : <EOL> """<STR_LIT>""" <EOL> repo_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org , <EOL> repo = repo <EOL> ) <EOL> repo_response = self . session . get ( repo_url ) <EOL> if repo_response . status_code == <NUM_LIT:200> : <EOL> return repo_response . json ( ) <EOL> if repo_response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( repo_response . text ) <EOL> def _find_team ( self , org , team ) : <EOL> """<STR_LIT>""" <EOL> list_teams_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org <EOL> ) <EOL> teams = self . _get_all ( list_teams_url ) <EOL> if not teams : <EOL> raise GitHubUnknownError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> found_team = [ <EOL> x for x in teams <EOL> if x [ '<STR_LIT:name>' ] . strip ( ) . lower ( ) == team . strip ( ) . lower ( ) <EOL> ] <EOL> if len ( found_team ) != <NUM_LIT:1> : <EOL> raise GitHubNoTeamFound ( <EOL> '<STR_LIT>' . format ( team , org ) <EOL> ) <EOL> found_team = found_team [ <NUM_LIT:0> ] <EOL> return found_team <EOL> def create_repo ( self , org , repo , description ) : <EOL> """<STR_LIT>""" <EOL> repo_dict = self . _get_repo ( org , repo ) <EOL> if repo_dict is not None : <EOL> raise GitHubRepoExists ( '<STR_LIT>' ) <EOL> create_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org <EOL> ) <EOL> payload = { <EOL> '<STR_LIT:name>' : repo , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : True , <EOL> } <EOL> repo_create_response = self . session . post ( create_url , json = payload ) <EOL> if repo_create_response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( repo_create_response . text ) <EOL> return repo_create_response . json ( ) <EOL> def _create_team ( self , org , team_name , read_only ) : <EOL> """<STR_LIT>""" <EOL> if read_only : <EOL> permission = '<STR_LIT>' <EOL> else : <EOL> permission = '<STR_LIT>' <EOL> create_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org <EOL> ) <EOL> response = self . session . post ( create_url , json = { <EOL> '<STR_LIT:name>' : team_name , <EOL> '<STR_LIT>' : permission <EOL> } ) <EOL> if response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( response . text ) <EOL> return response . json ( ) <EOL> def put_team ( self , org , team_name , read_only , members ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> team_dict = self . _find_team ( org , team_name ) <EOL> except GitHubNoTeamFound : <EOL> team_dict = self . _create_team ( org , team_name , read_only ) <EOL> if members is None : <EOL> return team_dict <EOL> members_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> id = team_dict [ '<STR_LIT:id>' ] <EOL> ) <EOL> existing_members = self . _get_all ( members_url ) <EOL> existing_members = [ x [ '<STR_LIT>' ] for x in existing_members ] <EOL> remove_members = dict ( <EOL> [ ( x , False ) for x in existing_members if x not in members ] <EOL> ) <EOL> add_members = dict ( <EOL> [ ( x , True ) for x in members if x not in existing_members ] <EOL> ) <EOL> membership_dict = dict ( <EOL> chain ( remove_members . items ( ) , add_members . items ( ) ) <EOL> ) <EOL> for member , add in membership_dict . items ( ) : <EOL> url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> id = team_dict [ '<STR_LIT:id>' ] , <EOL> member = member <EOL> ) <EOL> if add : <EOL> response = self . session . put ( url ) <EOL> else : <EOL> response = self . session . delete ( url ) <EOL> if response . status_code not in [ <NUM_LIT:200> , <NUM_LIT> ] : <EOL> raise GitHubUnknownError ( <EOL> '<STR_LIT>' . format ( <EOL> member , response . text <EOL> ) <EOL> ) <EOL> return team_dict <EOL> def add_team_repo ( self , org , repo , team ) : <EOL> """<STR_LIT>""" <EOL> found_team = self . _find_team ( org , team ) <EOL> team_repo_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> id = found_team [ '<STR_LIT:id>' ] , <EOL> org = org , <EOL> repo = repo <EOL> ) <EOL> response = self . session . put ( team_repo_url ) <EOL> if response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( response . text ) <EOL> def add_web_hook ( self , org , repo , url ) : <EOL> """<STR_LIT>""" <EOL> hook_url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org , <EOL> repo = repo <EOL> ) <EOL> payload = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:url>' : url , <EOL> } <EOL> } <EOL> response = self . session . post ( hook_url , json = payload ) <EOL> if response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( response . text ) <EOL> return response . json ( ) <EOL> def delete_web_hooks ( self , org , repo ) : <EOL> """<STR_LIT>""" <EOL> repo_dict = self . _get_repo ( org , repo ) <EOL> if repo_dict is None : <EOL> raise GitHubRepoDoesNotExist ( <EOL> '<STR_LIT>' <EOL> ) <EOL> url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org , <EOL> repo = repo <EOL> ) <EOL> hooks = self . _get_all ( url ) <EOL> num_hooks_removed = <NUM_LIT:0> <EOL> for hook in hooks or [ ] : <EOL> response = self . session . delete ( hook [ '<STR_LIT:url>' ] ) <EOL> if response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( response . text ) <EOL> num_hooks_removed += <NUM_LIT:1> <EOL> return num_hooks_removed <EOL> @ staticmethod <EOL> def shallow_copy_repo ( src_repo , dst_repo , committer , branch = None ) : <EOL> """<STR_LIT>""" <EOL> cwd = unicode ( sh . pwd ( ) . rstrip ( '<STR_LIT:\n>' ) ) <EOL> tmp_dir = tempfile . mkdtemp ( prefix = '<STR_LIT>' ) <EOL> try : <EOL> sh . cd ( tmp_dir ) <EOL> if branch is None : <EOL> sh . git . clone ( src_repo , CLONE_DIR , depth = <NUM_LIT:1> ) <EOL> else : <EOL> sh . git . clone ( src_repo , CLONE_DIR , depth = <NUM_LIT:1> , branch = branch ) <EOL> sh . cd ( CLONE_DIR ) <EOL> shutil . rmtree ( '<STR_LIT>' ) <EOL> sh . git . init ( ) <EOL> sh . git . config ( '<STR_LIT>' , committer [ '<STR_LIT:email>' ] ) <EOL> sh . git . config ( '<STR_LIT>' , committer [ '<STR_LIT:name>' ] ) <EOL> sh . git . remote . add . origin ( dst_repo ) <EOL> sh . git . add ( '<STR_LIT:.>' ) <EOL> sh . git . commit ( <EOL> m = '<STR_LIT>' . format ( <EOL> src_repo <EOL> ) <EOL> ) <EOL> sh . git . push . origin . master ( f = True ) <EOL> finally : <EOL> shutil . rmtree ( tmp_dir , ignore_errors = True ) <EOL> sh . cd ( cwd ) <EOL> def add_repo_file ( self , org , repo , committer , message , path , contents ) : <EOL> """<STR_LIT>""" <EOL> repo_dict = self . _get_repo ( org , repo ) <EOL> if repo_dict is None : <EOL> raise GitHubRepoDoesNotExist ( <EOL> '<STR_LIT>' <EOL> ) <EOL> url = '<STR_LIT>' . format ( <EOL> url = self . api_url , <EOL> org = org , <EOL> repo = repo , <EOL> path = path <EOL> ) <EOL> payload = { <EOL> '<STR_LIT:message>' : message , <EOL> '<STR_LIT>' : committer , <EOL> '<STR_LIT:content>' : base64 . b64encode ( contents ) . decode ( '<STR_LIT:ascii>' ) , <EOL> } <EOL> response = self . session . put ( url , json = payload ) <EOL> if response . status_code != <NUM_LIT> : <EOL> raise GitHubUnknownError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> org = org , repo = repo , path = path , response = response . text <EOL> ) <EOL> ) </s>
