91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
import os, shutil
|
|
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"]
|
|
remote = f"{git_protocol}://{username}:{password}@{git_remote}"
|
|
|
|
if os.path.exists(repo_path):
|
|
shutil.rmtree(repo_path)
|
|
self.repo_path = repo_path
|
|
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"""
|
|
try:
|
|
self.repo.remotes[remote_name].pull(ref_name=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_branch(self, branch_name, remote_name='origin', ref_name='main'):
|
|
"""Create a new branch in the repository with authentication."""
|
|
try:
|
|
# Use the same remote and ref as before
|
|
self.repo.git.branch(branch_name, commit=True)
|
|
return True
|
|
except GitCommandError as e:
|
|
print(f"Failed to create branch: {e}")
|
|
return False
|
|
|
|
def add_and_commit(self, message=None):
|
|
"""Add and commit changes to the repository."""
|
|
try:
|
|
# 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(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_branch(title)
|
|
|
|
shutil.copy(f"{file_path}", f"{self.repo_path}src/content/")
|
|
|
|
self.add_and_commit(commit_messge)
|
|
|
|
self.repo.git.push(remote_name='origin', ref_name=title, force=True)
|
|
|
|
def remove_repo(self):
|
|
shutil.rmtree(self.repo_path) |