def _get_codeartifact_index()

in src/sagemaker_training/modules.py [0:0]

• 41 lines of code
• 3 McCabe index (conditional complexity)

def _get_codeartifact_index():
    """
    Build the authenticated codeartifact index url based on the arn provided
    via CA_REPOSITORY_ARN environment variable following the form
    `arn:${Partition}:codeartifact:${Region}:${Account}:repository/${DomainName}/${RepositoryName}`
    https://docs.aws.amazon.com/codeartifact/latest/ug/python-configure-pip.html
    https://docs.aws.amazon.com/service-authorization/latest/reference/list_awscodeartifact.html#awscodeartifact-resources-for-iam-policies
    :return: authenticated codeartifact index url
    """
    repository_arn = os.getenv(CA_REPOSITORY_ARN_ENV)
    arn_regex = (
        "arn:(?P<partition>[^:]+):codeartifact:(?P<region>[^:]+):(?P<account>[^:]+)"
        ":repository/(?P<domain>[^/]+)/(?P<repository>.+)"
    )
    m = re.match(arn_regex, repository_arn)
    if not m:
        raise Exception("invalid CodeArtifact repository arn {}".format(repository_arn))
    domain = m.group("domain")
    owner = m.group("account")
    repository = m.group("repository")
    region = m.group("region")

    logger.info(
        "configuring pip to use codeartifact "
        "(domain: %s, domain owner: %s, repository: %s, region: %s)",
        domain,
        owner,
        repository,
        region,
    )
    try:
        client = boto3.client("codeartifact", region_name=region)
        auth_token_response = client.get_authorization_token(domain=domain, domainOwner=owner)
        token = auth_token_response["authorizationToken"]
        endpoint_response = client.get_repository_endpoint(
            domain=domain, domainOwner=owner, repository=repository, format="pypi"
        )
        unauthenticated_index = endpoint_response["repositoryEndpoint"]
        return re.sub(
            "https://",
            "https://aws:{}@".format(token),
            re.sub(
                "{}/?$".format(repository),
                "{}/simple/".format(repository),
                unauthenticated_index,
            ),
        )
    except Exception:
        logger.error("failed to configure pip to use codeartifact")
        raise Exception("failed to configure pip to use codeartifact")