git/AutoMetabuilder
1
0
Fork 0
mirror of https://github.com/johndoe6345789/AutoMetabuilder.git synced 2026-04-24 13:54:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Convert services to workflow plugins and remove old files

Browse Source 
- Moved OpenAI factory code into backend_create_openai.py plugin
- Moved OpenAI client helper code into core_ai_request.py plugin
- Moved GitHub integration code into backend_create_github.py plugin
- Moved context loader code into core_load_context.py plugin
- Removed entire services directory (5 files)
- Removed utils/context_loader.py (1 file)
- Updated utils/__init__.py to remove context_loader references

Co-authored-by: johndoe6345789 <224850594+johndoe6345789@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-01-10 16:37:19 +00:00
parent 6fccefe12a
commit ec681fe63e
11 changed files with 147 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
backend/autometabuilder/services/__init__.py
View File

@@ -1,22 +0,0 @@
"""
Services module for AutoMetabuilder.
This module contains service integrations:
- github_integration: GitHub API integration
- github_service: GitHub service builder
- openai_client: OpenAI client helpers
- openai_factory: OpenAI client factory
"""
from .github_integration import GitHubIntegration, get_repo_name_from_env
from .github_service import create_github_integration
from .openai_client import get_completion
from .openai_factory import create_openai_client
__all__ = [
"GitHubIntegration",
"get_repo_name_from_env",
"create_github_integration",
"get_completion",
"create_openai_client",
]

71
backend/autometabuilder/services/github_integration.py
View File

@@ -1,71 +0,0 @@
"""
GitHub integration module.
"""
import os
from github import Github
from github.Issue import Issue
from github.PullRequest import PullRequest
from tenacity import retry, stop_after_attempt, wait_exponential
from .. import load_messages
class GitHubIntegration:
"""Class to handle GitHub interactions."""
def __init__(self, token: str, repo_name: str):
self.github = Github(token)
self.repo = self.github.get_repo(repo_name)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_open_issues(self):
"""Get open issues from the repository."""
return self.repo.get_issues(state='open')
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_issue(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
return self.repo.get_issue(number=issue_number)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def create_branch(self, branch_name: str, base_branch: str = "main"):
"""Create a new branch from a base branch."""
base_ref = self.repo.get_git_ref(f"heads/{base_branch}")
self.repo.create_git_ref(
ref=f"refs/heads/{branch_name}", sha=base_ref.object.sha
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def create_pull_request(
self,
title: str,
body: str,
head_branch: str,
base_branch: str = "main",
) -> PullRequest:
"""Create a new pull request."""
return self.repo.create_pull(
title=title, body=body, head=head_branch, base=base_branch
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_pull_requests(self, state: str = "open"):
"""Get pull requests from the repository."""
return self.repo.get_pulls(state=state)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_pull_request_comments(self, pr_number: int):
"""Get comments from a specific pull request."""
pr = self.repo.get_pull(pr_number)
return pr.get_issue_comments()
def get_repo_name_from_env() -> str:
"""Retrieve repository name from environment variable."""
# Try to get from environment variable
repo_name = os.environ.get("GITHUB_REPOSITORY")
if not repo_name:
# Fallback or error
msgs = load_messages()
raise ValueError(msgs["error_github_repo_missing"])
return repo_name

19
backend/autometabuilder/services/github_service.py
View File

@@ -1,19 +0,0 @@
"""GitHub integration builder."""
import logging
from .github_integration import GitHubIntegration, get_repo_name_from_env
logger = logging.getLogger("autometabuilder")
def create_github_integration(token: str, msgs: dict):
"""Create GitHub integration if possible."""
if not token:
return None
try:
repo_name = get_repo_name_from_env()
gh = GitHubIntegration(token, repo_name)
logger.info(msgs["info_integrated_repo"].format(repo_name=repo_name))
return gh
except Exception as error: # pylint: disable=broad-exception-caught
logger.warning(msgs["warn_github_init_failed"].format(error=error))
return None

15
backend/autometabuilder/services/openai_client.py
View File

@@ -1,15 +0,0 @@
"""OpenAI client helpers."""
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_completion(client, model, messages, tools):
"""Request a chat completion with retries."""
return client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice="auto",
temperature=1.0,
top_p=1.0,
)

13
backend/autometabuilder/services/openai_factory.py
View File

@@ -1,13 +0,0 @@
"""OpenAI client factory."""
import os
from openai import OpenAI
DEFAULT_ENDPOINT = "https://models.github.ai/inference"
def create_openai_client(token: str):
"""Create an OpenAI client."""
return OpenAI(
base_url=os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_ENDPOINT),
api_key=token,
)

3
backend/autometabuilder/utils/__init__.py
View File

@@ -3,7 +3,6 @@ Utilities module for AutoMetabuilder.
This module contains various utility functions:
- cli_args: CLI argument parsing
- context_loader: Load SDLC context from repo and GitHub
- docker_utils: Docker command utilities
- logging_config: Logging configuration with TRACE support
- model_resolver: Resolve LLM model names
@@ -12,7 +11,6 @@ This module contains various utility functions:
"""
from .cli_args import parse_args
from .context_loader import get_sdlc_context
from .docker_utils import run_command_in_docker
from .logging_config import configure_logging
from .model_resolver import resolve_model_name
@@ -21,7 +19,6 @@ from .tool_map_builder import build_tool_map
__all__ = [
"parse_args",
"get_sdlc_context",
"run_command_in_docker",
"configure_logging",
"resolve_model_name",

38
backend/autometabuilder/utils/context_loader.py
View File

@@ -1,38 +0,0 @@
"""Load SDLC context from repo and GitHub."""
import os
import logging
from ..services.github_integration import GitHubIntegration
logger = logging.getLogger("autometabuilder")
def get_sdlc_context(gh: GitHubIntegration, msgs: dict) -> str:
"""Return SDLC context text from roadmap, issues, and PRs."""
sdlc_context = ""
if os.path.exists("ROADMAP.md"):
with open("ROADMAP.md", "r", encoding="utf-8") as f:
roadmap_content = f.read()
label = msgs.get("roadmap_label", "ROADMAP.md Content:")
sdlc_context += f"\n{label}\n{roadmap_content}\n"
else:
msg = msgs.get(
"missing_roadmap_msg",
"ROADMAP.md is missing. Please analyze the repository and create it."
)
sdlc_context += f"\n{msg}\n"
if gh:
try:
issues = gh.get_open_issues()
issue_list = "\n".join([f"- #{i.number}: {i.title}" for i in issues[:5]])
if issue_list:
sdlc_context += f"\n{msgs['open_issues_label']}\n{issue_list}"
prs = gh.get_pull_requests()
pr_list = "\n".join([f"- #{p.number}: {p.title}" for p in prs[:5]])
if pr_list:
sdlc_context += f"\n{msgs['open_prs_label']}\n{pr_list}"
except Exception as error: # pylint: disable=broad-exception-caught
logger.error(msgs["error_sdlc_context"].format(error=error))
return sdlc_context

85
backend/autometabuilder/workflow/plugins/backend/backend_create_github.py
View File

@@ -1,5 +1,75 @@
"""Workflow plugin: create GitHub integration."""
from ....services.github_service import create_github_integration
import os
import logging
from github import Github
from github.Issue import Issue
from github.PullRequest import PullRequest
from tenacity import retry, stop_after_attempt, wait_exponential
from .... import load_messages
logger = logging.getLogger("autometabuilder")
class GitHubIntegration:
"""Class to handle GitHub interactions."""
def __init__(self, token: str, repo_name: str):
self.github = Github(token)
self.repo = self.github.get_repo(repo_name)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_open_issues(self):
"""Get open issues from the repository."""
return self.repo.get_issues(state='open')
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_issue(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
return self.repo.get_issue(number=issue_number)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def create_branch(self, branch_name: str, base_branch: str = "main"):
"""Create a new branch from a base branch."""
base_ref = self.repo.get_git_ref(f"heads/{base_branch}")
self.repo.create_git_ref(
ref=f"refs/heads/{branch_name}", sha=base_ref.object.sha
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def create_pull_request(
self,
title: str,
body: str,
head_branch: str,
base_branch: str = "main",
) -> PullRequest:
"""Create a new pull request."""
return self.repo.create_pull(
title=title, body=body, head=head_branch, base=base_branch
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_pull_requests(self, state: str = "open"):
"""Get pull requests from the repository."""
return self.repo.get_pulls(state=state)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_pull_request_comments(self, pr_number: int):
"""Get comments from a specific pull request."""
pr = self.repo.get_pull(pr_number)
return pr.get_issue_comments()
def get_repo_name_from_env() -> str:
"""Retrieve repository name from environment variable."""
# Try to get from environment variable
repo_name = os.environ.get("GITHUB_REPOSITORY")
if not repo_name:
# Fallback or error
msgs = load_messages()
raise ValueError(msgs["error_github_repo_missing"])
return repo_name
def run(runtime, _inputs):
@@ -7,7 +77,18 @@ def run(runtime, _inputs):
token = runtime.context.get("github_token")
msgs = runtime.context.get("msgs", {})
gh = create_github_integration(token, msgs)
# Create GitHub integration if possible
if not token:
gh = None
else:
try:
repo_name = get_repo_name_from_env()
gh = GitHubIntegration(token, repo_name)
logger.info(msgs["info_integrated_repo"].format(repo_name=repo_name))
except Exception as error: # pylint: disable=broad-exception-caught
logger.warning(msgs["warn_github_init_failed"].format(error=error))
gh = None
# Store in both store (for workflow) and context (for other plugins)
runtime.context["gh"] = gh
return {"result": gh, "initialized": gh is not None}

12
backend/autometabuilder/workflow/plugins/backend/backend_create_openai.py
View File

@@ -1,12 +1,20 @@
"""Workflow plugin: create OpenAI client."""
from ....services.openai_factory import create_openai_client
import os
from openai import OpenAI
DEFAULT_ENDPOINT = "https://models.github.ai/inference"
def run(runtime, _inputs):
"""Initialize OpenAI client."""
token = runtime.context.get("github_token")
client = create_openai_client(token)
# Create OpenAI client
client = OpenAI(
base_url=os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_ENDPOINT),
api_key=token,
)
# Store in both store (for workflow) and context (for other plugins)
runtime.context["client"] = client
return {"result": client, "initialized": client is not None}

17
backend/autometabuilder/workflow/plugins/core/core_ai_request.py
View File

@@ -1,11 +1,24 @@
"""Workflow plugin: AI request."""
from ....services.openai_client import get_completion
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def _get_completion(client, model, messages, tools):
"""Request a chat completion with retries."""
return client.chat.completions.create(
model=model,
messages=messages,
tools=tools,
tool_choice="auto",
temperature=1.0,
top_p=1.0,
)
def run(runtime, inputs):
"""Invoke the model with current messages."""
messages = list(inputs.get("messages") or [])
response = get_completion(
response = _get_completion(
runtime.context["client"],
runtime.context["model_name"],
messages,

41
backend/autometabuilder/workflow/plugins/core/core_load_context.py
View File

@@ -1,7 +1,44 @@
"""Workflow plugin: load SDLC context."""
from ....utils.context_loader import get_sdlc_context
import os
import logging
logger = logging.getLogger("autometabuilder")
def run(runtime, _inputs):
"""Load SDLC context into the workflow store."""
return {"context": get_sdlc_context(runtime.context["gh"], runtime.context["msgs"])}
gh = runtime.context.get("gh")
msgs = runtime.context.get("msgs", {})
# Build SDLC context from roadmap, issues, and PRs
sdlc_context = ""
# Load ROADMAP.md if it exists
if os.path.exists("ROADMAP.md"):
with open("ROADMAP.md", "r", encoding="utf-8") as f:
roadmap_content = f.read()
label = msgs.get("roadmap_label", "ROADMAP.md Content:")
sdlc_context += f"\n{label}\n{roadmap_content}\n"
else:
msg = msgs.get(
"missing_roadmap_msg",
"ROADMAP.md is missing. Please analyze the repository and create it."
)
sdlc_context += f"\n{msg}\n"
# Load GitHub issues and PRs if integration is available
if gh:
try:
issues = gh.get_open_issues()
issue_list = "\n".join([f"- #{i.number}: {i.title}" for i in issues[:5]])
if issue_list:
sdlc_context += f"\n{msgs['open_issues_label']}\n{issue_list}"
prs = gh.get_pull_requests()
pr_list = "\n".join([f"- #{p.number}: {p.title}" for p in prs[:5]])
if pr_list:
sdlc_context += f"\n{msgs['open_prs_label']}\n{pr_list}"
except Exception as error: # pylint: disable=broad-exception-caught
logger.error(msgs["error_sdlc_context"].format(error=error))
return {"context": sdlc_context}