Add gitea-shim integration and update workflow services

This commit is contained in:
2025-06-23 13:14:20 -03:00
parent b0e9dcf3a1
commit 1b62e9c3d4
19 changed files with 2856 additions and 7 deletions

View File

@ -0,0 +1,7 @@
"""Model classes for Gitea GitHub shim."""
from .repository import Repository
from .pull_request import PullRequest
from .user import User
__all__ = ['Repository', 'PullRequest', 'User']

View File

@ -0,0 +1,266 @@
"""Pull Request model that provides GitHub-compatible interface for Gitea pull requests."""
from typing import Optional, Dict, Any, List
class PullRequest:
"""
Pull Request wrapper that provides GitHub-compatible interface for Gitea pull requests.
"""
def __init__(self, gitea_pr, gitea_repo, gitea_client):
"""
Initialize pull request wrapper.
Args:
gitea_pr: The Gitea pull request object
gitea_repo: The Gitea repository object
gitea_client: The Gitea client instance
"""
self._pr = gitea_pr
self._repo = gitea_repo
self._gitea = gitea_client
# Map Gitea attributes to GitHub attributes
self.number = gitea_pr.number
self.state = gitea_pr.state
self.title = gitea_pr.title
self.body = gitea_pr.body
self.created_at = gitea_pr.created_at
self.updated_at = gitea_pr.updated_at
self.closed_at = gitea_pr.closed_at
self.merged_at = gitea_pr.merged_at
self.merge_commit_sha = gitea_pr.merge_commit_sha
self.html_url = gitea_pr.html_url
self.diff_url = gitea_pr.diff_url
self.patch_url = gitea_pr.patch_url
self.mergeable = gitea_pr.mergeable
self.merged = gitea_pr.merged
self.draft = getattr(gitea_pr, 'draft', False) # Gitea might not have draft PRs
# User information
self.user = self._create_user_object(gitea_pr.user)
self.assignee = self._create_user_object(gitea_pr.assignee) if gitea_pr.assignee else None
self.assignees = [self._create_user_object(a) for a in getattr(gitea_pr, 'assignees', [])]
# Branch information
self.base = self._create_branch_info(gitea_pr.base)
self.head = self._create_branch_info(gitea_pr.head)
# Labels and milestone
self.labels = [self._create_label_object(l) for l in getattr(gitea_pr, 'labels', [])]
self.milestone = self._create_milestone_object(gitea_pr.milestone) if gitea_pr.milestone else None
def _create_user_object(self, gitea_user) -> Dict[str, Any]:
"""Create a GitHub-compatible user object."""
if not gitea_user:
return None
return {
'login': gitea_user.login,
'id': gitea_user.id,
'avatar_url': gitea_user.avatar_url,
'type': 'User'
}
def _create_branch_info(self, gitea_branch) -> Dict[str, Any]:
"""Create a GitHub-compatible branch info object."""
return {
'ref': gitea_branch.ref,
'sha': gitea_branch.sha,
'repo': {
'name': gitea_branch.repo.name,
'full_name': gitea_branch.repo.full_name,
'owner': self._create_user_object(gitea_branch.repo.owner)
} if gitea_branch.repo else None
}
def _create_label_object(self, gitea_label) -> Dict[str, Any]:
"""Create a GitHub-compatible label object."""
return {
'name': gitea_label.name,
'color': gitea_label.color,
'description': gitea_label.description
}
def _create_milestone_object(self, gitea_milestone) -> Dict[str, Any]:
"""Create a GitHub-compatible milestone object."""
return {
'title': gitea_milestone.title,
'description': gitea_milestone.description,
'state': gitea_milestone.state,
'number': gitea_milestone.id,
'due_on': gitea_milestone.due_on
}
def update(self, **kwargs) -> 'PullRequest':
"""
Update the pull request.
Args:
**kwargs: Fields to update (title, body, state, base, etc.)
Returns:
Updated PullRequest object
"""
# Map GitHub parameters to Gitea parameters
update_params = {}
if 'title' in kwargs:
update_params['title'] = kwargs['title']
if 'body' in kwargs:
update_params['body'] = kwargs['body']
if 'state' in kwargs:
update_params['state'] = kwargs['state']
if 'assignee' in kwargs:
update_params['assignee'] = kwargs['assignee']
if 'assignees' in kwargs:
update_params['assignees'] = kwargs['assignees']
if 'labels' in kwargs:
update_params['labels'] = kwargs['labels']
if 'milestone' in kwargs:
update_params['milestone'] = kwargs['milestone']
# Update using Gitea API
updated_pr = self._pr.update(**update_params)
# Return new PullRequest instance with updated data
return PullRequest(updated_pr, self._repo, self._gitea)
def merge(self, commit_title: Optional[str] = None,
commit_message: Optional[str] = None,
sha: Optional[str] = None,
merge_method: str = 'merge') -> Dict[str, Any]:
"""
Merge the pull request.
Args:
commit_title: Title for the merge commit
commit_message: Message for the merge commit
sha: SHA that must match the HEAD of the PR
merge_method: Merge method ('merge', 'squash', 'rebase')
Returns:
Merge result information
"""
# Map GitHub merge methods to Gitea
gitea_merge_style = {
'merge': 'merge',
'squash': 'squash',
'rebase': 'rebase'
}.get(merge_method, 'merge')
# Merge using Gitea API
result = self._pr.merge(
style=gitea_merge_style,
title=commit_title,
message=commit_message
)
return {
'sha': result.sha if hasattr(result, 'sha') else None,
'merged': True,
'message': 'Pull Request successfully merged'
}
def get_commits(self) -> List[Dict[str, Any]]:
"""
Get commits in the pull request.
Returns:
List of commit objects
"""
gitea_commits = self._pr.get_commits()
commits = []
for commit in gitea_commits:
commits.append({
'sha': commit.sha,
'commit': {
'message': commit.commit.message,
'author': {
'name': commit.commit.author.name,
'email': commit.commit.author.email,
'date': commit.commit.author.date
}
},
'html_url': commit.html_url
})
return commits
def get_files(self) -> List[Dict[str, Any]]:
"""
Get files changed in the pull request.
Returns:
List of file objects
"""
gitea_files = self._pr.get_files()
files = []
for file in gitea_files:
files.append({
'filename': file.filename,
'status': file.status,
'additions': file.additions,
'deletions': file.deletions,
'changes': file.changes,
'patch': file.patch,
'sha': file.sha,
'blob_url': file.blob_url,
'raw_url': file.raw_url
})
return files
def create_review_comment(self, body: str, commit_id: str, path: str,
position: int) -> Dict[str, Any]:
"""
Create a review comment on the pull request.
Args:
body: Comment body
commit_id: Commit SHA to comment on
path: File path to comment on
position: Line position in the diff
Returns:
Comment object
"""
comment = self._pr.create_review_comment(
body=body,
commit_id=commit_id,
path=path,
position=position
)
return {
'id': comment.id,
'body': comment.body,
'path': comment.path,
'position': comment.position,
'commit_id': comment.commit_id,
'user': self._create_user_object(comment.user),
'created_at': comment.created_at,
'updated_at': comment.updated_at
}
def create_issue_comment(self, body: str) -> Dict[str, Any]:
"""
Create a general comment on the pull request.
Args:
body: Comment body
Returns:
Comment object
"""
comment = self._pr.create_comment(body=body)
return {
'id': comment.id,
'body': comment.body,
'user': self._create_user_object(comment.user),
'created_at': comment.created_at,
'updated_at': comment.updated_at
}

View File

@ -0,0 +1,291 @@
"""Repository model that provides GitHub-compatible interface for Gitea repositories."""
from typing import Optional, List, Dict, Any
try:
# When running tests or using as a module
from models.pull_request import PullRequest
except ImportError:
# When using as a package
from .pull_request import PullRequest
class Repository:
"""
Repository wrapper that provides GitHub-compatible interface for Gitea repositories.
"""
def __init__(self, gitea_repo, gitea_client):
"""
Initialize repository wrapper.
Args:
gitea_repo: The Gitea repository object
gitea_client: The Gitea client instance
"""
self._repo = gitea_repo
self._gitea = gitea_client
# Map Gitea attributes to GitHub attributes
self.name = gitea_repo.name
self.full_name = gitea_repo.full_name
self.description = gitea_repo.description
self.private = gitea_repo.private
self.fork = gitea_repo.fork
self.created_at = gitea_repo.created_at
self.updated_at = gitea_repo.updated_at
self.pushed_at = gitea_repo.updated_at # Gitea doesn't have pushed_at
self.size = gitea_repo.size
self.stargazers_count = gitea_repo.stars_count
self.watchers_count = gitea_repo.watchers_count
self.forks_count = gitea_repo.forks_count
self.open_issues_count = gitea_repo.open_issues_count
self.default_branch = gitea_repo.default_branch
self.archived = gitea_repo.archived
self.html_url = gitea_repo.html_url
self.clone_url = gitea_repo.clone_url
self.ssh_url = gitea_repo.ssh_url
# Owner information
self.owner = self._create_owner_object(gitea_repo.owner)
def _create_owner_object(self, gitea_owner) -> Dict[str, Any]:
"""Create a GitHub-compatible owner object."""
return {
'login': gitea_owner.login,
'id': gitea_owner.id,
'avatar_url': gitea_owner.avatar_url,
'type': 'User' if not getattr(gitea_owner, 'is_organization', False) else 'Organization'
}
def get_pull(self, number: int) -> PullRequest:
"""
Get a pull request by number.
Args:
number: Pull request number
Returns:
PullRequest object
"""
gitea_pr = self._repo.get_pull(number)
return PullRequest(gitea_pr, self._repo, self._gitea)
def get_pulls(self, state: str = 'open', sort: str = 'created',
direction: str = 'desc', base: Optional[str] = None,
head: Optional[str] = None) -> List[PullRequest]:
"""
Get pull requests for the repository.
Args:
state: State of PRs to return ('open', 'closed', 'all')
sort: Sort field ('created', 'updated', 'popularity')
direction: Sort direction ('asc', 'desc')
base: Filter by base branch
head: Filter by head branch
Returns:
List of PullRequest objects
"""
# Map GitHub state to Gitea state
gitea_state = state
if state == 'all':
gitea_state = None # Gitea uses None for all
gitea_prs = self._repo.get_pulls(state=gitea_state)
# Convert to PullRequest objects
pulls = [PullRequest(pr, self._repo, self._gitea) for pr in gitea_prs]
# Apply additional filters if needed
if base:
pulls = [pr for pr in pulls if pr.base.ref == base]
if head:
pulls = [pr for pr in pulls if pr.head.ref == head]
return pulls
def create_pull(self, title: str, body: str, head: str, base: str,
draft: bool = False, **kwargs) -> PullRequest:
"""
Create a new pull request.
Args:
title: PR title
body: PR description
head: Source branch
base: Target branch
draft: Whether PR is a draft
**kwargs: Additional parameters
Returns:
PullRequest object
"""
# Create PR using Gitea API
gitea_pr = self._repo.create_pull_request(
title=title,
body=body,
head=head,
base=base
# Note: Gitea might not support draft PRs
)
return PullRequest(gitea_pr, self._repo, self._gitea)
def get_contents(self, path: str, ref: Optional[str] = None) -> Any:
"""
Get file contents from repository.
Args:
path: File path
ref: Git ref (branch, tag, commit SHA)
Returns:
File contents object
"""
# Get file contents using Gitea API
contents = self._repo.get_file_contents(path, ref=ref)
# Create GitHub-compatible response
return {
'name': contents.name,
'path': contents.path,
'sha': contents.sha,
'size': contents.size,
'type': contents.type,
'content': contents.content,
'encoding': contents.encoding,
'download_url': contents.download_url,
'html_url': contents.html_url
}
def get_branches(self) -> List[Dict[str, Any]]:
"""
Get all branches in the repository.
Returns:
List of branch objects
"""
gitea_branches = self._repo.get_branches()
branches = []
for branch in gitea_branches:
branches.append({
'name': branch.name,
'commit': {
'sha': branch.commit.id,
'url': branch.commit.url
},
'protected': branch.protected
})
return branches
def get_branch(self, branch: str) -> Dict[str, Any]:
"""
Get a specific branch.
Args:
branch: Branch name
Returns:
Branch object
"""
gitea_branch = self._repo.get_branch(branch)
return {
'name': gitea_branch.name,
'commit': {
'sha': gitea_branch.commit.id,
'url': gitea_branch.commit.url
},
'protected': gitea_branch.protected
}
def create_fork(self, organization: Optional[str] = None) -> 'Repository':
"""
Create a fork of the repository.
Args:
organization: Organization to fork to (optional)
Returns:
Repository object for the fork
"""
gitea_fork = self._repo.create_fork(organization=organization)
return Repository(gitea_fork, self._gitea)
def get_commits(self, sha: Optional[str] = None, path: Optional[str] = None,
since: Optional[str] = None, until: Optional[str] = None) -> List[Dict[str, Any]]:
"""
Get commits from the repository.
Args:
sha: Starting commit SHA or branch
path: Only commits containing this path
since: Only commits after this date
until: Only commits before this date
Returns:
List of commit objects
"""
# Note: Gitea API might have different parameters
gitea_commits = self._repo.get_commits()
commits = []
for commit in gitea_commits:
commits.append({
'sha': commit.sha,
'commit': {
'message': commit.commit.message,
'author': {
'name': commit.commit.author.name,
'email': commit.commit.author.email,
'date': commit.commit.author.date
},
'committer': {
'name': commit.commit.committer.name,
'email': commit.commit.committer.email,
'date': commit.commit.committer.date
}
},
'html_url': commit.html_url,
'author': commit.author,
'committer': commit.committer
})
return commits
def get_commit(self, sha: str) -> Dict[str, Any]:
"""
Get a specific commit.
Args:
sha: Commit SHA
Returns:
Commit object
"""
gitea_commit = self._repo.get_commit(sha)
return {
'sha': gitea_commit.sha,
'commit': {
'message': gitea_commit.commit.message,
'author': {
'name': gitea_commit.commit.author.name,
'email': gitea_commit.commit.author.email,
'date': gitea_commit.commit.author.date
},
'committer': {
'name': gitea_commit.commit.committer.name,
'email': gitea_commit.commit.committer.email,
'date': gitea_commit.commit.committer.date
}
},
'html_url': gitea_commit.html_url,
'author': gitea_commit.author,
'committer': gitea_commit.committer,
'stats': gitea_commit.stats,
'files': gitea_commit.files
}

View File

@ -0,0 +1,246 @@
"""User model that provides GitHub-compatible interface for Gitea users."""
from typing import Optional, List, Dict, Any
class User:
"""
User wrapper that provides GitHub-compatible interface for Gitea users.
"""
def __init__(self, gitea_user, gitea_client):
"""
Initialize user wrapper.
Args:
gitea_user: The Gitea user object
gitea_client: The Gitea client instance
"""
self._user = gitea_user
self._gitea = gitea_client
# Map Gitea attributes to GitHub attributes
self.login = gitea_user.login
self.id = gitea_user.id
self.avatar_url = gitea_user.avatar_url
self.html_url = gitea_user.html_url
self.type = 'User' # Could be 'User' or 'Organization'
self.name = gitea_user.full_name
self.company = getattr(gitea_user, 'company', None)
self.blog = getattr(gitea_user, 'website', None)
self.location = gitea_user.location
self.email = gitea_user.email
self.bio = getattr(gitea_user, 'description', None)
self.public_repos = getattr(gitea_user, 'public_repos', 0)
self.followers = getattr(gitea_user, 'followers_count', 0)
self.following = getattr(gitea_user, 'following_count', 0)
self.created_at = gitea_user.created
self.updated_at = getattr(gitea_user, 'last_login', gitea_user.created)
def get_repos(self, type: str = 'owner', sort: str = 'full_name',
direction: str = 'asc') -> List[Any]:
"""
Get repositories for the user.
Args:
type: Type of repos to return ('all', 'owner', 'member')
sort: Sort field ('created', 'updated', 'pushed', 'full_name')
direction: Sort direction ('asc', 'desc')
Returns:
List of Repository objects
"""
# Import here to avoid circular imports
try:
# When running tests or using as a module
from models.repository import Repository
except ImportError:
# When using as a package
from .repository import Repository
# Get repos using Gitea API
gitea_repos = self._user.get_repos()
# Convert to Repository objects
repos = [Repository(repo, self._gitea) for repo in gitea_repos]
# Apply sorting
if sort == 'full_name':
repos.sort(key=lambda r: r.full_name, reverse=(direction == 'desc'))
elif sort == 'created':
repos.sort(key=lambda r: r.created_at, reverse=(direction == 'desc'))
elif sort == 'updated':
repos.sort(key=lambda r: r.updated_at, reverse=(direction == 'desc'))
return repos
def get_starred(self) -> List[Any]:
"""
Get repositories starred by the user.
Returns:
List of Repository objects
"""
# Import here to avoid circular imports
try:
# When running tests or using as a module
from models.repository import Repository
except ImportError:
# When using as a package
from .repository import Repository
# Get starred repos using Gitea API
gitea_repos = self._user.get_starred()
# Convert to Repository objects
return [Repository(repo, self._gitea) for repo in gitea_repos]
def get_subscriptions(self) -> List[Any]:
"""
Get repositories watched by the user.
Returns:
List of Repository objects
"""
# Import here to avoid circular imports
try:
# When running tests or using as a module
from models.repository import Repository
except ImportError:
# When using as a package
from .repository import Repository
# Get watched repos using Gitea API
gitea_repos = self._user.get_subscriptions()
# Convert to Repository objects
return [Repository(repo, self._gitea) for repo in gitea_repos]
def get_orgs(self) -> List[Dict[str, Any]]:
"""
Get organizations the user belongs to.
Returns:
List of organization objects
"""
gitea_orgs = self._user.get_orgs()
orgs = []
for org in gitea_orgs:
orgs.append({
'login': org.username,
'id': org.id,
'avatar_url': org.avatar_url,
'description': org.description,
'type': 'Organization'
})
return orgs
def get_followers(self) -> List['User']:
"""
Get users following this user.
Returns:
List of User objects
"""
gitea_followers = self._user.get_followers()
# Convert to User objects
return [User(follower, self._gitea) for follower in gitea_followers]
def get_following(self) -> List['User']:
"""
Get users this user is following.
Returns:
List of User objects
"""
gitea_following = self._user.get_following()
# Convert to User objects
return [User(user, self._gitea) for user in gitea_following]
def has_in_following(self, following: 'User') -> bool:
"""
Check if this user follows another user.
Args:
following: User to check
Returns:
True if following, False otherwise
"""
return self._user.is_following(following.login)
def add_to_following(self, following: 'User') -> None:
"""
Follow another user.
Args:
following: User to follow
"""
self._user.follow(following.login)
def remove_from_following(self, following: 'User') -> None:
"""
Unfollow another user.
Args:
following: User to unfollow
"""
self._user.unfollow(following.login)
def get_keys(self) -> List[Dict[str, Any]]:
"""
Get SSH keys for the user.
Returns:
List of key objects
"""
gitea_keys = self._user.get_keys()
keys = []
for key in gitea_keys:
keys.append({
'id': key.id,
'key': key.key,
'title': key.title,
'created_at': key.created_at,
'read_only': getattr(key, 'read_only', True)
})
return keys
def create_repo(self, name: str, **kwargs) -> Any:
"""
Create a new repository for the user.
Args:
name: Repository name
**kwargs: Additional parameters
Returns:
Repository object
"""
# Import here to avoid circular imports
try:
# When running tests or using as a module
from models.repository import Repository
except ImportError:
# When using as a package
from .repository import Repository
# Map GitHub parameters to Gitea parameters
gitea_params = {
'name': name,
'description': kwargs.get('description', ''),
'private': kwargs.get('private', False),
'auto_init': kwargs.get('auto_init', False),
'gitignores': kwargs.get('gitignore_template', ''),
'license': kwargs.get('license_template', ''),
'readme': kwargs.get('readme', '')
}
gitea_repo = self._user.create_repo(**gitea_params)
return Repository(gitea_repo, self._gitea)