|
4 | 4 | from git import Actor |
5 | 5 | from git import Repo |
6 | 6 | from git.exc import GitCommandError |
7 | | -from git.exc import NoSuchPathError |
8 | 7 | from git.refs.head import Head |
9 | 8 | from git.remote import PushInfo |
10 | 9 | from git.remote import PushInfoList |
11 | 10 | from typing import Any |
12 | 11 | from typing import Union |
13 | | -import json |
14 | 12 | import logging |
15 | | -import os |
16 | 13 | import re |
17 | | -import sys |
18 | 14 |
|
19 | 15 | _logger = logging.getLogger(app_config.APP_NAME) |
20 | 16 |
|
21 | 17 |
|
22 | | -def load_targets_config(file_path: str) -> dict[str, Any]: |
23 | | - """Load configuration from a JSON file. |
24 | | -
|
25 | | - Args: |
26 | | - file_path (str): The path to the JSON configuration file. |
27 | | -
|
28 | | - Returns: |
29 | | - dict: A dictionary containing the loaded configuration. |
30 | | -
|
31 | | - Raises: |
32 | | - FileNotFoundError: If the configuration file is not found. |
33 | | - json.JSONDecodeError: If the JSON content is not valid. |
34 | | - Exception: For any other unexpected error during loading. |
35 | | - """ |
36 | | - try: |
37 | | - file_abspath = os.path.abspath(file_path) |
38 | | - with open(file_abspath) as f: |
39 | | - _logger.info(f"Loading '{file_abspath}'") |
40 | | - return json.load(f) |
41 | | - |
42 | | - except FileNotFoundError: |
43 | | - error_message = f"Configuration file '{ |
44 | | - file_abspath}' not found." |
45 | | - |
46 | | - except json.JSONDecodeError: |
47 | | - error_message = f"Unable to load configuration from '{ |
48 | | - file_abspath}'. Invalid JSON format." |
49 | | - |
50 | | - except Exception as e: |
51 | | - error_message = f"An unexpected error occurred while loading '{ |
52 | | - file_abspath}': {str(e).strip()}" |
53 | | - |
54 | | - _logger.error(error_message) |
55 | | - _logger.info("Exiting..") |
56 | | - sys.exit(1) |
57 | | - |
58 | | - |
59 | | -def load_target_repos(repos: list[dict]) -> list[Repo]: |
60 | | - """Load target repositories. |
61 | | -
|
62 | | - Args: |
63 | | - repos (list[dict]): A list of dictionaries containing repository information. |
64 | | -
|
65 | | - Returns: |
66 | | - list[Repo]: A list of GitPython Repo objects representing the loaded repositories. |
67 | | - """ |
68 | | - result = [] |
69 | | - for repo in repos: |
70 | | - try: |
71 | | - repo_type = repo['type'].strip().lower() |
72 | | - repo_name = repo['name'] |
73 | | - scm_provider_data: dict[str, str] = repo['scmProvider'] |
74 | | - if repo_type == 'local': |
75 | | - _logger.info(f"'{repo_name}' is a 'Local' repository. Using..") |
76 | | - repo_obj = Repo(path=repo['source']) |
77 | | - elif repo_type == 'remote': |
78 | | - _logger.info( |
79 | | - f"'{repo_name}' is a 'Remote' repository. Cloning..") |
80 | | - repo_obj = Repo.clone_from(url=repo['source'], to_path=os.path.join( |
81 | | - app_config.REMOTE_TARGETS_CLONING_PATH, repo_name)) |
82 | | - scm_provider_type = scm_provider_data['type'].strip( |
83 | | - ).lower().replace(' ', '') |
84 | | - if scm_provider_type == "azuredevops": |
85 | | - _logger.debug(f"'{scm_provider_type}' SCM provider detected") |
86 | | - repo_obj.scm_provider = { |
87 | | - 'type': scm_provider_type, |
88 | | - 'base_url': scm_provider_data['baseUrl'], |
89 | | - 'project': scm_provider_data['project'] |
90 | | - } |
91 | | - elif scm_provider_type == "github": |
92 | | - _logger.debug(f"'{scm_provider_type}' SCM provider detected") |
93 | | - repo_obj.scm_provider = { |
94 | | - 'type': scm_provider_type, |
95 | | - 'domain': scm_provider_data['domain'], |
96 | | - 'owner_or_org': scm_provider_data['ownerOrOrg'].strip() |
97 | | - } |
98 | | - result.append(repo_obj) |
99 | | - except GitCommandError as git_cmd_err: |
100 | | - if git_cmd_err.status == 128 and 'already exists' in git_cmd_err.stderr: |
101 | | - repo_abspath = os.path.abspath(os.path.join( |
102 | | - app_config.REMOTE_TARGETS_CLONING_PATH, repo_name)) |
103 | | - _logger.info(f"'{repo_abspath}' already exists, using..") |
104 | | - repo_obj = Repo(path=repo_abspath) |
105 | | - scm_provider_type = scm_provider_data['type'].strip( |
106 | | - ).lower().replace(' ', '') |
107 | | - if scm_provider_type == "azuredevops": |
108 | | - _logger.debug( |
109 | | - f"'{scm_provider_type}' SCM provider detected") |
110 | | - repo_obj.scm_provider = { |
111 | | - 'type': scm_provider_type, |
112 | | - 'base_url': scm_provider_data['baseUrl'], |
113 | | - 'project': scm_provider_data['project'] |
114 | | - } |
115 | | - elif scm_provider_type == "github": |
116 | | - _logger.debug( |
117 | | - f"'{scm_provider_type}' SCM provider detected") |
118 | | - repo_obj.scm_provider = { |
119 | | - 'type': scm_provider_type, |
120 | | - 'domain': scm_provider_data['domain'], |
121 | | - 'owner_or_org': scm_provider_data['ownerOrOrg'].strip() |
122 | | - } |
123 | | - result.append(repo_obj) |
124 | | - else: |
125 | | - _logger.error(f"Unexpected GitCommandError: { |
126 | | - str(git_cmd_err).strip()}") |
127 | | - except NoSuchPathError: |
128 | | - repo_abspath = os.path.abspath(os.path.join( |
129 | | - app_config.REMOTE_TARGETS_CLONING_PATH, repo_name)) |
130 | | - _logger.error(f"Invalid 'Local' repo source path '{ |
131 | | - repo_abspath}': No such path. Check '{repo_name}' source path") |
132 | | - except Exception as e: |
133 | | - _logger.error(f"Unexpected error when loading '{ |
134 | | - repo_name}': {str(e).strip()}") |
135 | | - return result |
136 | | - |
137 | | - |
138 | 18 | def identity_setup(repo: Repo, actor_username: str, actor_email: str) -> bool: |
139 | 19 | """Set up identity configuration for a GitPython repository. |
140 | 20 |
|
@@ -227,7 +107,8 @@ def checkout_branch(repo: Repo, branch_name: str, from_branch: str = None, remot |
227 | 107 | branch.set_tracking_branch(repo.refs[remote_branch_name]) |
228 | 108 | else: |
229 | 109 | _logger.info(f"'{branch_name}' doesn't exist, creating..") |
230 | | - from_branch = from_branch or get_default_branch_name(repo=repo, remote_name=remote_name) |
| 110 | + from_branch = from_branch or get_default_branch_name( |
| 111 | + repo=repo, remote_name=remote_name) |
231 | 112 | remote_from_branch = f"{remote_name}/{from_branch}" |
232 | 113 | if from_branch in repo.branches: |
233 | 114 | branch = repo.create_head(branch_name, commit=from_branch) |
|
0 commit comments