def update_()

in src/lambda_codebase/initial_commit/initial_commit.py [0:0]

• 42 lines of code
• 11 McCabe index (conditional complexity)

def update_(event: Mapping[str, Any], _context: Any, create_pr=False) -> Tuple[PhysicalResourceId, Data]: #pylint: disable=R0912, R0915
    update_event = UpdateEvent(**event)
    repo_name = repo_arn_to_name(update_event.ResourceProperties.RepositoryArn)
    files_to_delete = get_files_to_delete(repo_name)
    files_to_commit = get_files_to_commit(update_event.ResourceProperties.DirectoryName)

    commit_id = CC_CLIENT.get_branch(
        repositoryName=repo_name,
        branchName="master",
    )["branch"]["commitId"]
    CC_CLIENT.create_branch(
        **generate_create_branch_input(update_event, repo_name, commit_id)
    )

    if files_to_commit:
        try:
            for index, files in enumerate(chunks([f.as_dict() for f in files_to_commit], 99)):
                commit_id = CC_CLIENT.create_commit(**generate_commit_input(
                    repo_name,
                    index,
                    parent_commit_id=commit_id,
                    branch=update_event.ResourceProperties.Version,
                    puts=files
                ))["commitId"]
                create_pr = True # If the commit above was able to be made, we want to create a PR afterwards
        except (CC_CLIENT.exceptions.FileEntryRequiredException, CC_CLIENT.exceptions.NoChangeException):
            pass
    if files_to_delete:
        try:
            for index, deletes in enumerate(chunks([f.as_dict() for f in files_to_delete], 99)):
                commit_id = CC_CLIENT.create_commit(**generate_commit_input(
                    repo_name,
                    index,
                    parent_commit_id=commit_id,
                    branch=update_event.ResourceProperties.Version,
                    deletes=deletes
                ))["commitId"]
        except (CC_CLIENT.exceptions.FileEntryRequiredException, CC_CLIENT.exceptions.NoChangeException):
            pass
    if create_pr or files_to_delete:
        CC_CLIENT.create_pull_request(**generate_pull_request_input(update_event, repo_name))
    else:
        CC_CLIENT.delete_branch(**generate_delete_branch_input(update_event, repo_name))

    return event["PhysicalResourceId"], {}