Fixed copy error when a team does not exist in target organization

This commit is contained in:
Enrico Ludwig 2024-08-06 21:21:20 +02:00
parent 6aa5172634
commit ff49e11010

View File

@ -1,15 +1,29 @@
import argparse
import os
import socket
import sys
from gitea import *
from rich.console import Console
from rich.spinner import Spinner
from rich.logging import RichHandler
from loguru import logger
from dotenv import load_dotenv
# fmt: off
load_dotenv()
def str_to_bool(value):
return value.lower() in ['true', '1', 'yes', 'y']
GITEA_INSTANCE : str = os.getenv("GITEA_INSTANCE") if os.getenv("GITEA_INSTANCE") != None else None
GITEA_PORT : int = os.getenv("GITEA_PORT") if os.getenv("GITEA_PORT") != None else None
GITEA_TOKEN : str = os.getenv("GITEA_TOKEN") if os.getenv("GITEA_TOKEN") != None else None
GITEA_SSL : bool = str_to_bool(os.getenv("GITEA_SSL")) if os.getenv("GITEA_SSL") != None else False
SOURCE_ORGA : str = os.getenv("SOURCE_ORGA") if os.getenv("SOURCE_ORGA") != None else None
DRY_RUN : bool = str_to_bool(os.getenv("DRY_RUN")) if os.getenv("DRY_RUN") != None else None
EXCLUDE_ORGAS : list = os.getenv("EXCLUDE_ORGAS").split(',') if os.getenv("EXCLUDE_ORGAS") != None else None
UPDATE_TEAMS : bool = str_to_bool(os.getenv("UPDATE_TEAMS")) if os.getenv("UPDATE_TEAMS") != None else False
OVERRIDE_TEAMS : bool = str_to_bool(os.getenv("OVERRIDE_TEAMS")) if os.getenv("OVERRIDE_TEAMS") != None else False
parser = argparse.ArgumentParser()
parser.add_argument("--host", help="Specify the Gitea instance host")
parser.add_argument("--port", help="Specify the Gitea instance port")
@ -23,15 +37,15 @@ parser.add_argument("--update", help="Updates existing teams in target organizat
parser.add_argument("--override", help="Override existing teams in target organizations", action="store_true")
args = parser.parse_args()
GITEA_INSTANCE : str = args.host if args.host else None
GITEA_PORT : int = args.port if args.port else None
GITEA_TOKEN : str = args.token if args.token else None
GITEA_SSL : bool = args.ssl if args.ssl else False
SOURCE_ORGA : str = args.source_orga if args.source_orga else None
DRY_RUN : bool = args.dry_run if args.dry_run else False
EXCLUDE_ORGAS : list = args.exclude if args.exclude else []
UPDATE_TEAMS : bool = args.update if args.update else False
OVERRIDE_TEAMS : bool = args.override if args.override else False
GITEA_INSTANCE : str = args.host if args.host else GITEA_INSTANCE
GITEA_PORT : int = args.port if args.port else GITEA_PORT
GITEA_TOKEN : str = args.token if args.token else GITEA_TOKEN
GITEA_SSL : bool = args.ssl if args.ssl else GITEA_SSL
SOURCE_ORGA : str = args.source_orga if args.source_orga else SOURCE_ORGA
DRY_RUN : bool = args.dry_run if args.dry_run else DRY_RUN
EXCLUDE_ORGAS : list = args.exclude if args.exclude else EXCLUDE_ORGAS
UPDATE_TEAMS : bool = args.update if args.update else UPDATE_TEAMS
OVERRIDE_TEAMS : bool = args.override if args.override else OVERRIDE_TEAMS
# fmt: on
@ -43,6 +57,20 @@ logger.add(
level="DEBUG" if args.debug else "INFO",
)
# fmt: off
logger.info("Starting Gitea Auto Mapper")
logger.debug("Debug logging enabled")
logger.info(f"Dry-run mode: {'Enabled' if DRY_RUN else 'Disabled'}")
logger.info(f"Target Gitea instance: {GITEA_INSTANCE}:{GITEA_PORT}")
logger.info(f"Using SSL: {'Enabled' if GITEA_SSL else 'Disabled'}")
logger.info(f"Source organization: {SOURCE_ORGA}")
logger.info(f"Excluded organizations: {', '.join(EXCLUDE_ORGAS) if EXCLUDE_ORGAS else 'None'}")
logger.info(f"Update mode: {'Enabled' if UPDATE_TEAMS else 'Disabled'}")
logger.info(f"Override mode: {'Enabled' if OVERRIDE_TEAMS else 'Disabled'}")
# fmt: on
def check_host(host, port):
@ -89,9 +117,13 @@ def check_host(host, port):
with console.status("- Checking Gitea Endpoint", spinner="dots") as status:
try:
check_host(GITEA_INSTANCE, GITEA_PORT)
logger.info("Gitea endpoint is valid")
logger.info(f"Gitea endpoint at {GITEA_INSTANCE}:{GITEA_PORT} is valid")
except Exception as e:
logger.error(f"Failed to check Gitea endpoint: {e}")
logger.error(
"Please double check the Gitea endpoint and make sure it is reachable"
)
os._exit(1)
# Create a Gitea API client
gitea_client = None
@ -184,7 +216,13 @@ with console.status("- Copying Teams", spinner="dots") as status:
for team in source_orga_teams:
try:
# check if the team already exists in the target organization
existing_team = orga.get_team(team.name, True)
orga_teams = orga.get_teams()
existing_team = next(
(t for t in orga_teams if t.name == team.name), None
)
if existing_team:
logger.debug(f"\tTeam {team.name} already exists in {orga.name}")
if OVERRIDE_TEAMS and existing_team:
logger.info(f"\tDeleting existing team '{team.name}'")
@ -192,8 +230,8 @@ with console.status("- Copying Teams", spinner="dots") as status:
existing_team.delete()
existing_team = None
logger.info(f"\tCreating team '{team.name}'")
if not DRY_RUN and not existing_team:
logger.info(f"\tCreating team '{team.name}'")
new_team = gitea_client.create_team(
org=orga,
name=team.name,