apb_extra_utils.github 

  1import functools
  2import json
  3import os
  4import shutil
  5from tempfile import mkdtemp
  6from urllib.error import HTTPError
  7from urllib.request import Request, urlopen
  8
  9from apb_extra_utils.misc import download_and_unzip, remove_content_dir, zip_dir, create_dir
 10
 11PREFIX_FILE_LAST_TAG_REPO = 'last_tag_repo_github_'
 12
 13
 14def get_api_github(owner, repo, api_request, token=None):
 15    """
 16    GET Request for repository GITHUB via API GITHUB.
 17
 18    See the REST API DOCS here https://docs.github.com/en/rest
 19
 20    Args:
 21        owner (str):
 22        repo (str):
 23        api_request (str):
 24        token (str=None):
 25
 26    Returns:
 27        info_response (dict)
 28    """
 29    request_headers = {
 30        'Accept': 'application/vnd.github.v3+json'
 31    }
 32    if token:
 33        request_headers['Authorization'] = f'token {token}'
 34
 35    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
 36    req = Request(url_github, headers=request_headers, method='GET')
 37    info_response = {}
 38    try:
 39        with urlopen(req) as resp_request:
 40            if resp_request:
 41                info_response = json.load(resp_request)
 42    except HTTPError as exc:
 43        info_response[HTTPError.__name__] = str(exc)
 44
 45    return info_response
 46
 47
 48def post_api_github(owner, repo, api_request, post_data, token=None):
 49    """
 50    POST Request on repository GITHUB via API GITHUB
 51
 52    See the REST API DOCS here https://docs.github.com/en/rest
 53
 54    Args:
 55        owner (str):
 56        repo (str):
 57        api_request (str):
 58        post_data (dict):
 59        token (str=None):
 60
 61    Returns:
 62        info_response (dict)
 63    """
 64    request_headers = {
 65        'Accept': 'application/vnd.github.v3+json',
 66        'Content-Type': 'application/json; charset=utf-8',
 67    }
 68    if token:
 69        request_headers['Authorization'] = f'token {token}'
 70
 71    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
 72
 73    post_data_enc = json.dumps(post_data).encode('utf-8')
 74
 75    req = Request(url_github, headers=request_headers, method='POST')
 76    info_response = {}
 77    try:
 78        with urlopen(req, post_data_enc) as resp_request:
 79            if resp_request:
 80                info_response = resp_request.__dict__
 81    except HTTPError as exc:
 82        info_response[HTTPError.__name__] = str(exc)
 83
 84    return info_response
 85
 86
 87@functools.cache
 88def has_changes_in_github(owner, repo, branch, download_to, token=None):
 89    """
 90    Check if the GitHub repository branch has changes.
 91    
 92    Args:
 93        owner (str): Owner repository Github
 94        repo (str): Name repository Github
 95        branch (str): Branch repository to check
 96        download_to (str): Path to the local repository
 97        token (str=None): Github token for private access
 98
 99    Returns:
100        bool: True if there are changes, False otherwise
101        str: sha_commit of the current branch state
102    """
103    branch = branch.lower()
104    info_branches = get_api_github(owner, repo, 'branches', token)
105    info_branch = next(filter(lambda el: el.get('name', '').lower() == branch,
106                              info_branches), None)
107
108    if not info_branch:
109        return False, None
110
111    sha_commit = info_branch.get('commit').get('sha')
112    expected_name_zip_repo = f'{repo}-{branch}'
113    log_last_tag = os.path.join(download_to, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
114
115    if os.path.exists(log_last_tag):
116        with open(log_last_tag) as fr:
117            last_tag = fr.read()
118            if last_tag and last_tag.strip() == sha_commit.strip():
119                return False, sha_commit
120
121    return True, sha_commit
122
123
124def get_resources_from_repo_github(html_repo, tag, expected_name_zip_repo, path_repo, header=None,
125                                   force_update=False, remove_prev=False, as_zip=False):
126    """
127    
128    Args:
129        html_repo (str):
130        tag (str):
131        expected_name_zip_repo (str):
132        path_repo (str):
133        header (dict=None):
134        force_update (bool=False):
135        remove_prev (bool=False):
136        as_zip (bool=False):
137
138    Returns:
139        updated (bool)
140    """
141    updated = False
142    log_last_tag = os.path.join(path_repo, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
143
144    if not force_update:
145        last_tag = None
146        if os.path.exists(log_last_tag):
147            with open(log_last_tag) as fr:
148                last_tag = fr.read()
149
150        if last_tag and last_tag.strip() == tag.strip():
151            return updated
152
153    if not header:
154        header = {}
155    header['Accept'] = 'application/octet-stream'
156
157    dir_temp = mkdtemp()
158    download_and_unzip(html_repo, extract_to=dir_temp, headers=[(k, v) for k, v in header.items()])
159    path_res = os.path.join(dir_temp, expected_name_zip_repo)
160
161    if os.path.exists(path_res):
162        create_dir(path_repo)
163
164        if as_zip:
165            zip_dir(path_res, os.path.join(path_repo, f'{expected_name_zip_repo}.zip'))
166        else:
167            if remove_prev and os.path.exists(path_repo):
168                remove_content_dir(path_repo)
169            shutil.copytree(path_res, path_repo, dirs_exist_ok=True)
170
171        shutil.rmtree(path_res, ignore_errors=True)
172
173        with open(log_last_tag, "w+") as fw:
174            fw.write(tag)
175
176        updated = True
177
178    return updated
179
180
181def download_release_repo_github(owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False,
182                                 remove_prev=False):
183    """
184    Download release Github repository on the path selected.
185
186    Args:
187        owner (str): Owner repository Github
188        repo (str): Name repository Github
189        download_to (str): Path to download
190        tag_release (str=None): if not informed get 'latest' release
191        token (str=None): Github token for private access
192        force (bool=False): Force update if exists previous sources
193        remove_prev (bool=False): Remove all previous resources
194        as_zip (bool=False): Retorna como ZIP
195
196    Returns:
197        tag_name (str)
198    """
199    if not tag_release:
200        info_release = get_api_github(owner, repo, 'releases/latest', token)
201    else:
202        info_release = get_api_github(owner, repo, f'releases/tags/{tag_release}', token)
203
204    tag_name = info_release.get('tag_name')
205    if tag_name:
206        html_release = f'https://github.com/{owner}/{repo}/archive/refs/tags/{tag_name}.zip'
207        header = {}
208        if token:
209            header['Authorization'] = f'token {token}'
210
211        get_resources_from_repo_github(html_release, tag_name, f'{repo}-{tag_name}', download_to,
212                                       header=header, force_update=force, remove_prev=remove_prev, as_zip=as_zip)
213
214        return tag_name
215
216
217def download_branch_repo_github(owner, repo, branch, download_to, token=None, force=False, as_zip=False,
218                                remove_prev=False):
219    """
220    Download the branch selected for the Github repo on the path selected
221
222    Args:
223        owner (str): Owner repository Github
224        repo (str): Name repository Github
225        branch (str): Branch repository to download
226        download_to (str): Path to download
227        token (str=None): Github token for private access
228        force (bool=False): Force update if exists previous sources
229        remove_prev (bool=False): Remove all previous resources
230        as_zip (bool=False): Retorna como ZIP
231
232    Returns:
233        sha_commit (str), updated (boolean)
234    """
235    has_changes, sha_commit = has_changes_in_github(owner, repo, branch, download_to, token)
236    if not has_changes and not force:
237        return sha_commit, False
238    html_branch = f'https://github.com/{owner}/{repo}/archive/refs/heads/{branch}.zip'
239    header = {}
240    if token:
241        header['Authorization'] = f'token {token}'
242
243    name_zip = f'{repo}-{branch}'
244    updated = get_resources_from_repo_github(html_branch, sha_commit, name_zip, download_to,
245                                             header=header, force_update=force, remove_prev=remove_prev, as_zip=as_zip)
246
247    if as_zip:
248        path_zip = os.path.join(download_to, f'{name_zip}.zip')
249        if os.path.exists(path_zip):
250            new_path_zip = os.path.join(download_to, f'{name_zip}-{sha_commit}.zip')
251            if os.path.exists(new_path_zip):
252                os.remove(new_path_zip)
253            os.rename(path_zip, new_path_zip)
254
255    return sha_commit, updated
256
257
258if __name__ == '__main__':
259    import fire, sys
260
261    sys.exit(fire.Fire(
262        {
263            get_api_github.__name__: get_api_github,
264            post_api_github.__name__: post_api_github,
265            download_release_repo_github.__name__: download_release_repo_github,
266            download_branch_repo_github.__name__: download_branch_repo_github
267        }
268    ))
PREFIX_FILE_LAST_TAG_REPO = 'last_tag_repo_github_'
def get_api_github(owner, repo, api_request, token=None): 
15def get_api_github(owner, repo, api_request, token=None):
16    """
17    GET Request for repository GITHUB via API GITHUB.
18
19    See the REST API DOCS here https://docs.github.com/en/rest
20
21    Args:
22        owner (str):
23        repo (str):
24        api_request (str):
25        token (str=None):
26
27    Returns:
28        info_response (dict)
29    """
30    request_headers = {
31        'Accept': 'application/vnd.github.v3+json'
32    }
33    if token:
34        request_headers['Authorization'] = f'token {token}'
35
36    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
37    req = Request(url_github, headers=request_headers, method='GET')
38    info_response = {}
39    try:
40        with urlopen(req) as resp_request:
41            if resp_request:
42                info_response = json.load(resp_request)
43    except HTTPError as exc:
44        info_response[HTTPError.__name__] = str(exc)
45
46    return info_response

GET Request for repository GITHUB via API GITHUB.

See the REST API DOCS here apb_extra_utils.github.com/en/rest">https://docsapb_extra_utils.github.com/en/rest

Arguments:
  • owner (str):
  • repo (str):
  • api_request (str):
  • token (str=None):
Returns:

info_response (dict)

def post_api_github(owner, repo, api_request, post_data, token=None): 
49def post_api_github(owner, repo, api_request, post_data, token=None):
50    """
51    POST Request on repository GITHUB via API GITHUB
52
53    See the REST API DOCS here https://docs.github.com/en/rest
54
55    Args:
56        owner (str):
57        repo (str):
58        api_request (str):
59        post_data (dict):
60        token (str=None):
61
62    Returns:
63        info_response (dict)
64    """
65    request_headers = {
66        'Accept': 'application/vnd.github.v3+json',
67        'Content-Type': 'application/json; charset=utf-8',
68    }
69    if token:
70        request_headers['Authorization'] = f'token {token}'
71
72    url_github = f'https://api.github.com/repos/{owner}/{repo}/{api_request}'
73
74    post_data_enc = json.dumps(post_data).encode('utf-8')
75
76    req = Request(url_github, headers=request_headers, method='POST')
77    info_response = {}
78    try:
79        with urlopen(req, post_data_enc) as resp_request:
80            if resp_request:
81                info_response = resp_request.__dict__
82    except HTTPError as exc:
83        info_response[HTTPError.__name__] = str(exc)
84
85    return info_response

POST Request on repository GITHUB via API GITHUB

See the REST API DOCS here apb_extra_utils.github.com/en/rest">https://docsapb_extra_utils.github.com/en/rest

Arguments:
  • owner (str):
  • repo (str):
  • api_request (str):
  • post_data (dict):
  • token (str=None):
Returns:

info_response (dict)

@functools.cache
def has_changes_in_github(owner, repo, branch, download_to, token=None): 
 88@functools.cache
 89def has_changes_in_github(owner, repo, branch, download_to, token=None):
 90    """
 91    Check if the GitHub repository branch has changes.
 92    
 93    Args:
 94        owner (str): Owner repository Github
 95        repo (str): Name repository Github
 96        branch (str): Branch repository to check
 97        download_to (str): Path to the local repository
 98        token (str=None): Github token for private access
 99
100    Returns:
101        bool: True if there are changes, False otherwise
102        str: sha_commit of the current branch state
103    """
104    branch = branch.lower()
105    info_branches = get_api_github(owner, repo, 'branches', token)
106    info_branch = next(filter(lambda el: el.get('name', '').lower() == branch,
107                              info_branches), None)
108
109    if not info_branch:
110        return False, None
111
112    sha_commit = info_branch.get('commit').get('sha')
113    expected_name_zip_repo = f'{repo}-{branch}'
114    log_last_tag = os.path.join(download_to, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
115
116    if os.path.exists(log_last_tag):
117        with open(log_last_tag) as fr:
118            last_tag = fr.read()
119            if last_tag and last_tag.strip() == sha_commit.strip():
120                return False, sha_commit
121
122    return True, sha_commit

Check if the GitHub repository branch has changes.

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • branch (str): Branch repository to check
  • download_to (str): Path to the local repository
  • token (str=None): Github token for private access
Returns:

bool: True if there are changes, False otherwise str: sha_commit of the current branch state

def get_resources_from_repo_github( html_repo, tag, expected_name_zip_repo, path_repo, header=None, force_update=False, remove_prev=False, as_zip=False): 
125def get_resources_from_repo_github(html_repo, tag, expected_name_zip_repo, path_repo, header=None,
126                                   force_update=False, remove_prev=False, as_zip=False):
127    """
128    
129    Args:
130        html_repo (str):
131        tag (str):
132        expected_name_zip_repo (str):
133        path_repo (str):
134        header (dict=None):
135        force_update (bool=False):
136        remove_prev (bool=False):
137        as_zip (bool=False):
138
139    Returns:
140        updated (bool)
141    """
142    updated = False
143    log_last_tag = os.path.join(path_repo, f'.{PREFIX_FILE_LAST_TAG_REPO}{expected_name_zip_repo}')
144
145    if not force_update:
146        last_tag = None
147        if os.path.exists(log_last_tag):
148            with open(log_last_tag) as fr:
149                last_tag = fr.read()
150
151        if last_tag and last_tag.strip() == tag.strip():
152            return updated
153
154    if not header:
155        header = {}
156    header['Accept'] = 'application/octet-stream'
157
158    dir_temp = mkdtemp()
159    download_and_unzip(html_repo, extract_to=dir_temp, headers=[(k, v) for k, v in header.items()])
160    path_res = os.path.join(dir_temp, expected_name_zip_repo)
161
162    if os.path.exists(path_res):
163        create_dir(path_repo)
164
165        if as_zip:
166            zip_dir(path_res, os.path.join(path_repo, f'{expected_name_zip_repo}.zip'))
167        else:
168            if remove_prev and os.path.exists(path_repo):
169                remove_content_dir(path_repo)
170            shutil.copytree(path_res, path_repo, dirs_exist_ok=True)
171
172        shutil.rmtree(path_res, ignore_errors=True)
173
174        with open(log_last_tag, "w+") as fw:
175            fw.write(tag)
176
177        updated = True
178
179    return updated
Arguments:
  • html_repo (str):
  • tag (str):
  • expected_name_zip_repo (str):
  • path_repo (str):
  • header (dict=None):
  • force_update (bool=False):
  • remove_prev (bool=False):
  • as_zip (bool=False):
Returns:

updated (bool)

def download_release_repo_github( owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False, remove_prev=False): 
182def download_release_repo_github(owner, repo, download_to, tag_release=None, token=None, force=False, as_zip=False,
183                                 remove_prev=False):
184    """
185    Download release Github repository on the path selected.
186
187    Args:
188        owner (str): Owner repository Github
189        repo (str): Name repository Github
190        download_to (str): Path to download
191        tag_release (str=None): if not informed get 'latest' release
192        token (str=None): Github token for private access
193        force (bool=False): Force update if exists previous sources
194        remove_prev (bool=False): Remove all previous resources
195        as_zip (bool=False): Retorna como ZIP
196
197    Returns:
198        tag_name (str)
199    """
200    if not tag_release:
201        info_release = get_api_github(owner, repo, 'releases/latest', token)
202    else:
203        info_release = get_api_github(owner, repo, f'releases/tags/{tag_release}', token)
204
205    tag_name = info_release.get('tag_name')
206    if tag_name:
207        html_release = f'https://github.com/{owner}/{repo}/archive/refs/tags/{tag_name}.zip'
208        header = {}
209        if token:
210            header['Authorization'] = f'token {token}'
211
212        get_resources_from_repo_github(html_release, tag_name, f'{repo}-{tag_name}', download_to,
213                                       header=header, force_update=force, remove_prev=remove_prev, as_zip=as_zip)
214
215        return tag_name

Download release Github repository on the path selected.

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • download_to (str): Path to download
  • tag_release (str=None): if not informed get 'latest' release
  • token (str=None): Github token for private access
  • force (bool=False): Force update if exists previous sources
  • remove_prev (bool=False): Remove all previous resources
  • as_zip (bool=False): Retorna como ZIP
Returns:

tag_name (str)

def download_branch_repo_github( owner, repo, branch, download_to, token=None, force=False, as_zip=False, remove_prev=False): 
218def download_branch_repo_github(owner, repo, branch, download_to, token=None, force=False, as_zip=False,
219                                remove_prev=False):
220    """
221    Download the branch selected for the Github repo on the path selected
222
223    Args:
224        owner (str): Owner repository Github
225        repo (str): Name repository Github
226        branch (str): Branch repository to download
227        download_to (str): Path to download
228        token (str=None): Github token for private access
229        force (bool=False): Force update if exists previous sources
230        remove_prev (bool=False): Remove all previous resources
231        as_zip (bool=False): Retorna como ZIP
232
233    Returns:
234        sha_commit (str), updated (boolean)
235    """
236    has_changes, sha_commit = has_changes_in_github(owner, repo, branch, download_to, token)
237    if not has_changes and not force:
238        return sha_commit, False
239    html_branch = f'https://github.com/{owner}/{repo}/archive/refs/heads/{branch}.zip'
240    header = {}
241    if token:
242        header['Authorization'] = f'token {token}'
243
244    name_zip = f'{repo}-{branch}'
245    updated = get_resources_from_repo_github(html_branch, sha_commit, name_zip, download_to,
246                                             header=header, force_update=force, remove_prev=remove_prev, as_zip=as_zip)
247
248    if as_zip:
249        path_zip = os.path.join(download_to, f'{name_zip}.zip')
250        if os.path.exists(path_zip):
251            new_path_zip = os.path.join(download_to, f'{name_zip}-{sha_commit}.zip')
252            if os.path.exists(new_path_zip):
253                os.remove(new_path_zip)
254            os.rename(path_zip, new_path_zip)
255
256    return sha_commit, updated

Download the branch selected for the Github repo on the path selected

Arguments:
  • owner (str): Owner repository Github
  • repo (str): Name repository Github
  • branch (str): Branch repository to download
  • download_to (str): Path to download
  • token (str=None): Github token for private access
  • force (bool=False): Force update if exists previous sources
  • remove_prev (bool=False): Remove all previous resources
  • as_zip (bool=False): Retorna como ZIP
Returns:

sha_commit (str), updated (boolean)