|
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