103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
import os, shutil
|
|
from urllib.parse import quote
|
|
from git import Repo
|
|
from git.exc import GitCommandError
|
|
|
|
class GitRepository:
|
|
# This is designed to be transitory it will desctruvtively create the repo at repo_path
|
|
# if you have uncommited changes you can kiss them goodbye!
|
|
# Don't use the repo created by this function for dev -> its a tool!
|
|
# It is expected that when used you will add, commit, push, delete
|
|
def __init__(self, repo_path, username=None, password=None):
|
|
git_protocol = os.environ["GIT_PROTOCOL"]
|
|
git_remote = os.environ["GIT_REMOTE"]
|
|
#if username is not set we don't need parse to the url
|
|
if username==None or password == None:
|
|
remote = f"{git_protocol}://{git_remote}"
|
|
else:
|
|
# of course if it is we need to parse and escape it so that it
|
|
# can generate a url
|
|
git_user = quote(username)
|
|
git_password = quote(password)
|
|
remote = f"{git_protocol}://{git_user}:{git_password}@{git_remote}"
|
|
|
|
if os.path.exists(repo_path):
|
|
shutil.rmtree(repo_path)
|
|
self.repo_path = repo_path
|
|
print("Cloning Repo")
|
|
Repo.clone_from(remote, repo_path)
|
|
self.repo = Repo(repo_path)
|
|
self.username = username
|
|
self.password = password
|
|
|
|
def clone(self, remote_url, destination_path):
|
|
"""Clone a Git repository with authentication"""
|
|
try:
|
|
self.repo.clone(remote_url, destination_path)
|
|
return True
|
|
except GitCommandError as e:
|
|
print(f"Cloning failed: {e}")
|
|
return False
|
|
|
|
def fetch(self, remote_name='origin', ref_name='main'):
|
|
"""Fetch updates from a remote repository with authentication"""
|
|
try:
|
|
self.repo.remotes[remote_name].fetch(ref_name=ref_name)
|
|
return True
|
|
except GitCommandError as e:
|
|
print(f"Fetching failed: {e}")
|
|
return False
|
|
|
|
def pull(self, remote_name='origin', ref_name='main'):
|
|
"""Pull updates from a remote repository with authentication"""
|
|
print("Pulling Latest Updates (if any)")
|
|
try:
|
|
self.repo.remotes[remote_name].pull(ref_name)
|
|
return True
|
|
except GitCommandError as e:
|
|
print(f"Pulling failed: {e}")
|
|
return False
|
|
|
|
def get_branches(self):
|
|
"""List all branches in the repository"""
|
|
return [branch.name for branch in self.repo.branches]
|
|
|
|
|
|
def create_and_switch_branch(self, branch_name, remote_name='origin', ref_name='main'):
|
|
"""Create a new branch in the repository with authentication."""
|
|
try:
|
|
print(f"Creating Branch {branch_name}")
|
|
# Use the same remote and ref as before
|
|
self.repo.git.branch(branch_name)
|
|
except GitCommandError:
|
|
print("Branch already exists switching")
|
|
# ensure remote commits are pulled into local
|
|
self.repo.git.checkout(branch_name)
|
|
|
|
def add_and_commit(self, message=None):
|
|
"""Add and commit changes to the repository."""
|
|
try:
|
|
print("Commiting latest draft")
|
|
# Add all changes
|
|
self.repo.git.add(all=True)
|
|
# Commit with the provided message or a default
|
|
if message is None:
|
|
commit_message = "Added and committed new content"
|
|
else:
|
|
commit_message = message
|
|
self.repo.git.commit(message=commit_message)
|
|
return True
|
|
except GitCommandError as e:
|
|
print(f"Commit failed: {e}")
|
|
return False
|
|
|
|
def create_copy_commit_push(self, file_path, title, commit_messge):
|
|
self.create_and_switch_branch(title)
|
|
|
|
self.pull(ref_name=title)
|
|
shutil.copy(f"{file_path}", f"{self.repo_path}src/content/")
|
|
|
|
self.add_and_commit(f"'{commit_messge}'")
|
|
|
|
self.repo.git.push()
|